/********************************************************************************
*	ファイル名：RingQueue.cpp													*
*	用　　　途：リングバッファを使用した高速な固定長FIFOバッファ				*
*	備　　　考：																*
*	履　　　歴：																*
********************************************************************************/
/****************************
*	 コンパイル環境設定		*
****************************/
/****************************
*	インクルードファイル	*
****************************/
#include "stdafx.h"
#include "RingQueue.h"

/****************************
*		 マクロ定数			*
****************************/
// アライメント計算
#ifndef	IALIGN
#define IALIGN( n, d )		( ( ((n) + (d)-1) / (d) ) * (d) )
#endif

/****************************
*		 構造体定義			*
****************************/
/****************************
*		 クラス定義			*
****************************/
/****************************
*		 関数　宣言			*
****************************/
/****************************
*		 変数　宣言			*
****************************/

/********************************************************************************
*	関数名：CRingQueue															*
*	用　途：デフォルトコンストラクタ											*
*	引き数：メモリ確保方法														*
*	戻り値：																	*
*	備　考：																	*
********************************************************************************/
CRingQueue::CRingQueue()
	: m_MemQueue  ()
	, m_MemPackets() 
{
	InitMember( 0, 0 ) ;
}


/********************************************************************************
*	関数名：CRingQueue															*
*	用　途：待ち行列に格納する要素サイズおよび数量設定コンストラクタ			*
*	引き数：要素サイズ、数量、マルチスレッド対応モードフラグ					*
*	戻り値：																	*
*	備　考：																	*
********************************************************************************/
CRingQueue::CRingQueue( UINT32 nDataSize,	 /* : 要素サイズ               */
					    UINT32 nQuantity,	 /* : 数量                     */ 
						UINT32 nMTMode		 /* : マルチスレッド対応モード */ )
	: m_ppvQueue  ( NULL )
	, m_MemQueue  ()
	, m_MemPackets()
{
	/***** ↑ m_pvQueueを初期化していないとSetDataSizeは失敗する ******/

	SetDataSize( nDataSize, nQuantity, nMTMode ) ;
}


/********************************************************************************
*	関数名：~CRingQueue															*
*	用　途：デストラクタ														*
*	引き数：																	*
*	戻り値：																	*
*	備　考：																	*
********************************************************************************/
CRingQueue::~CRingQueue()
{
	RemoveAll() ;
}


/********************************************************************************
*	関　数　名：SetDataSize														*
*	用　　　途：キューに保持するデータサイズと個数を指定						*
*	引　き　数：データサイズ、個数												*
*	戻　り　数：設定成功でＴＲＵＥ、失敗でＦＡＬＳＥ							*
*	備　　　考：すでにキューが作成済みの場合は、失敗する						*
********************************************************************************/
BOOL CRingQueue::SetDataSize( UINT32 nDataSize, /* : 要素サイズ               */
							  UINT32 nQuantity, /* : 数量                     */  
							  UINT32 nMTMode	  /* : マルチスレッド対応モード */ )
{
	/***** 引数チェック nQuantityはUINT32だが、セマフォの個数の関係で最大INT_MAXまで *****/
	VERIFY( nDataSize<UINT_MAX && nQuantity<INT_MAX && nMTMode<=RQ_MT_FULL_ACCESS ) ;

	/***** 未初期化チェック *****/
	if( m_ppvQueue != NULL )					return FALSE ;

	InitMember( nDataSize, nQuantity ) ;

	UINT32	nQueueSize  = sizeof(void*)*m_nQuantity ;
	UINT32	nBufferSize = m_nPacketSize*m_nQuantity ;
	BOOL	bResult ;
	
	/***** メモリ割り当て *****/
	bResult = m_MemQueue.Alloc( nQueueSize ) ;
	if( !bResult ) 
	{ 
		return	FALSE ;
	}

	bResult = m_MemPackets.Alloc( nBufferSize ) ;
	if( !bResult ) 
	{ 
		m_MemQueue.Free() ;
		return	FALSE ;
	}

	m_ppvQueue		= (ATOMIC_PTR*)m_MemQueue  .GetPtr() ;
	m_pbyPackets	=			   m_MemPackets.GetPtr() ;

	/***** メモリアロケーション評価 *****/
	ASSERT( m_ppvQueue   != NULL ) ;
	ASSERT( m_pbyPackets != NULL ) ;

	/***** マルチスレッド対応 *****/
	if( nMTMode & RQ_MT_MULTI_WRITER )
	{
		m_plkWrite   = new CCriticalSection() ;
		VERIFY( m_plkWrite != NULL ) ;
	}
	if( nMTMode & RQ_MT_BLOCKING_READER )
	{
		m_hsmRead	= ::CreateSemaphore( NULL, 0, (LONG)m_nQuantity+1, NULL ) ;
		VERIFY( m_hsmRead != NULL ) ;
	}

	return TRUE ;
}


/********************************************************************************
*	関　数　名：GetDataSize														*
*	用　　　途：キューに保持するデータサイズを取得								*
*	引　き　数：																*
*	戻　り　数：データサイズ													*
*	備　　　考：																*
********************************************************************************/
UINT32 CRingQueue::GetDataSize() const 
{
	return m_nDataSize ;
}


/********************************************************************************
*	関　数　名：GetQuantity														*
*	用　　　途：キューに保持するデータの最大個数を取得							*
*	引　き　数：																*
*	戻　り　数：個数															*
*	備　　　考：																*
********************************************************************************/
UINT32 CRingQueue::GetQuantity() const 
{
	return m_nQuantity ;
}


/********************************************************************************
*	関　数　名：SetMTMode														*
*	用　　　途：マルチスレッド対応モードの設定									*
*	引　き　数：モード															*
*	戻　り　数：設定成功でＴＲＵＥ、失敗でＦＡＬＳＥ							*
*	備　　　考：マルチスレッド対応モードの場合、待機(ブロッキング中)のスレッド	*
*				がいない状態で使用すること										*
********************************************************************************/
BOOL CRingQueue::SetMTMode( UINT32 nMTMode )
{
	/***** 初期化チェック *****/
	if( m_ppvQueue == NULL ) return FALSE ;
	
	/***** キューをリセット *****/
	Reset() ;

	/***** 既にライターロック設定で、ライターロック解除の場合 *****/
	if( (m_plkWrite != NULL) && ((nMTMode & RQ_MT_MULTI_WRITER) == 0) )
	{
		delete m_plkWrite ;
		m_plkWrite = NULL ;
	}
	/***** ライターロック未設定で、ライターロック有りにする場合 *****/
	else if( (m_plkWrite == NULL) && ((nMTMode & RQ_MT_MULTI_WRITER) != 0) )
	{
		m_plkWrite   = new CCriticalSection() ;
		VERIFY( m_plkWrite != NULL ) ;
	}

	/***** 既にリーダーロック設定で、リーダーロック解除の場合 *****/
	if( (m_hsmRead != NULL) && ((nMTMode & RQ_MT_BLOCKING_READER) == 0) )
	{
		::CloseHandle( m_hsmRead ) ;
		m_hsmRead	  = NULL ;
	}
	/***** リーダー未設定で、リーダーロック有りにする場合 *****/
	else if( (m_hsmRead == NULL) && ((nMTMode & RQ_MT_BLOCKING_READER) != 0) )
	{
		m_hsmRead     = ::CreateSemaphore( NULL, 0, (LONG)m_nQuantity+1, NULL ) ;
		VERIFY( m_hsmRead != NULL ) ;
	}

	return TRUE ;
}


/********************************************************************************
*	関　数　名：RemoveAll														*
*	用　　　途：キューをリセットしてバッファを解放する							*
*	引　き　数：																*
*	戻　り　数：																*
*	備　　　考：マルチスレッド対応モードの場合、待機(ブロッキング中)のスレッド	*
*				がいない状態で使用すること										*
********************************************************************************/
BOOL CRingQueue::RemoveAll()
{
	/***** 未初期化チェック *****/
	if( m_ppvQueue == NULL ) return FALSE ;

	if( m_hsmRead  != NULL )
	{
		::CloseHandle( m_hsmRead ) ;
	}

	if( m_plkWrite != NULL )
	{
		delete m_plkWrite ;
	}

	m_MemQueue  .Free() ;
	m_MemPackets.Free() ;

	InitMember( 0, 0 ) ;

	return TRUE ;
}


/********************************************************************************
*	関　数　名：Reset															*
*	用　　　途：キューをリセットする（バッファを解放しない）					*
*	引　き　数：																*
*	戻　り　数：																*
*	備　　　考：マルチスレッド対応モードの場合、待機(ブロッキング)中のスレッド	*
*				がいない状態で使用すること										*
********************************************************************************/
BOOL CRingQueue::Reset()
{
	/***** 初期化チェック *****/
	if( m_ppvQueue == NULL ) return FALSE ;

	/***** リーダー用セマフォを再作成(状態を初期化するため) *****/
	if( m_hsmRead != NULL )
	{
		::CloseHandle( m_hsmRead ) ;
		m_hsmRead     = ::CreateSemaphore( NULL, 0, (LONG)m_nQuantity+1, NULL ) ;
		VERIFY( m_hsmRead != NULL ) ;
	}

	m_MemQueue.Clear() ;

	m_nReadIndex  = 0 ;
	m_nWriteIndex = 0 ;

	ATOMIC_RESET( m_lbShutDown );

	return TRUE ;
}


/********************************************************************************
*	関　数　名：ShutDown														*
*	用　　　途：キューのアクセスを停止する										*
*	引　き　数：																*
*	戻　り　数：																*
*	備　　　考：RemoveAllもしくはResetされるまで状態が保持される				*
********************************************************************************/
BOOL CRingQueue::ShutDown()
{
	/***** 初期化チェック *****/
	if( m_ppvQueue == NULL )	return FALSE ;

	/***** マルチスレッドライターロック *****/
	if( m_plkWrite != NULL )
	{
		m_plkWrite->Lock() ;

		if( ATOMIC_TEST(m_lbShutDown) )
		{
			m_plkWrite->Unlock() ;
								return TRUE ;
		}

		ATOMIC_SET(m_lbShutDown) ;

		/***** セマフォをインクリメントし、受信待機中のスレッドを起こす *****/
		if( m_hsmRead != NULL )
		{
			::ReleaseSemaphore( m_hsmRead, 1, NULL ) ;
		}

		m_plkWrite->Unlock() ;
	}
	/***** シングルスレッド *****/
	else
	{
		if( ATOMIC_TEST(m_lbShutDown) )		return TRUE ;

		ATOMIC_SET( m_lbShutDown ) ;

		/***** セマフォをインクリメントし、受信待機中のスレッドを起こす *****/
		if( m_hsmRead != NULL )
		{
			::ReleaseSemaphore( m_hsmRead, 1, NULL ) ;
		}
	}

	return TRUE ;
}


/********************************************************************************
*	関　数　名：IsShutDown														*
*	用　　　途：キューのアクセスを停止中か確認									*
*	引　き　数：																*
*	戻　り　数：																*
*	備　　　考：RemoveAllもしくはResetされるまで状態が保持される				*
********************************************************************************/
BOOL CRingQueue::IsShutDown()
{
	return ATOMIC_TEST(m_lbShutDown) ;
}


/********************************************************************************
*	関　数　名：InitMember														*
*	用　　　途：全メンバー変数を初期化する内部関数								*
*	引　き　数：																*
*	戻　り　数：																*
*	備　　　考：バッファの解放とは関係ない										*
********************************************************************************/
void CRingQueue::InitMember( UINT32 nDataSize, /* : 要素サイズ               */ 
							 UINT32 nQuantity  /* : 数量                     */ )
{
	m_nDataSize   = nDataSize ;
	m_nQuantity   = nQuantity ;
	m_nPacketSize = offsetof( Packet, aData ) + m_nDataSize ;

	m_nWriteIndex = 0 ;
	m_nReadIndex  = 0 ;

	m_ppvQueue    = NULL ;
	m_pbyPackets  = NULL ;
	ATOMIC_RESET( m_lbShutDown ) ;

	m_plkWrite    = NULL ;
	m_hsmRead     = NULL ;
}


/********************************************************************************
*	関　数　名：RingIncrement													*
*	用　　　途：リングバッファのインデックスを１つ先に移動する					*
*	引　き　数：インクリメント対象変数											*
*	戻　り　数：インクリメント前の値											*
*	備　　　考：																*
********************************************************************************/
INT32 CRingQueue::RingIncrement( INT32* piIndex /* : 対象インデックス */ )
{
	INT32		iNowIndex = *piIndex ;

	*piIndex = ( *piIndex + 1 ) % m_nQuantity ;

	return iNowIndex ;
}

/********************************************************************************
*	関　数　名：RingDecrement													*
*	用　　　途：リングバッファのインデックスを１つ前に移動する					*
*	引　き　数：デクリメント対象変数											*
*	戻　り　数：デクリメント前の値												*
*	備　　　考：																*
********************************************************************************/
INT32 CRingQueue::RingDecrement( INT32* piIndex /* : 対象インデックス */ )
{
	INT32		iNowIndex = *piIndex ;

	if( *piIndex > 0 ){
		*piIndex = *piIndex - 1 ;
	}
	else{
		*piIndex = m_nQuantity ;
	}

	return iNowIndex ;
}

/********************************************************************************
*	関　数　名：RingDecrement													*
*	用　　　途：リングバッファのインデックスを任意数前に移動する					*
*	引　き　数：デクリメント対象変数											*
*	戻　り　数：デクリメント前の値												*
*	備　　　考：																*
********************************************************************************/
INT32 CRingQueue::RingDecrement( INT32 nDecrement, INT32* piIndex )
{
	INT32		iNowIndex = *piIndex ;
	INT32		iTempIndex = *piIndex - nDecrement;

	if( iTempIndex >= 0 ){
		*piIndex = iTempIndex ;
	}
	else{
		*piIndex = m_nQuantity + iTempIndex;
	}

	return iNowIndex ;
}


/********************************************************************************
*	関　数　名：Push															*
*	用　　　途：待ち行列にデータを追加											*
*	引　き　数：キューデータ(要素)												*
*	戻　り　数：追加成功でＴＲＵＥ、失敗でＦＡＬＳＥ							*
*	備　　　考：																*
********************************************************************************/
BOOL CRingQueue::Push( void*	pvSrc, /* : 追加データ	 */
					   UINT32	nSize  /* : データサイズ */ )
{
	ASSERT( pvSrc != NULL ) ;

	INT32	nIndex ;		/* リングインデックス		 */
	Packet*	pPacket ;		/* パケットバッファへの参照	 */

	/***** 初期化チェック *****/
	if( m_ppvQueue == NULL   )						return FALSE ;
	if( nSize > m_nDataSize )						return FALSE ;

	/***** マルチスレッドロック *****/
	if( m_plkWrite != NULL )
	{
		m_plkWrite->Lock() ;

		if( ATOMIC_TEST(m_lbShutDown) )
		{
			m_plkWrite->Unlock() ;
													return FALSE ;
		}

		nIndex = m_nWriteIndex ;

		/***** 満杯チェック *****/
		if( ATOMIC_TEST_PTR(m_ppvQueue[nIndex]) )
		{	
			m_plkWrite->Unlock() ;
													return FALSE ;
		}

		RingIncrement( &m_nWriteIndex ) ;
	}
	else
	/***** シングルスレッド *****/
	{
		/***** 送受信停止チェック *****/
		if( ATOMIC_TEST(m_lbShutDown) )				return FALSE ;

		/***** 満杯チェック *****/
		nIndex = m_nWriteIndex ;
		if( ATOMIC_TEST_PTR(m_ppvQueue[nIndex]) )	return FALSE ;

		RingIncrement( &m_nWriteIndex ) ;
	}

	// パケットにデータセット
	pPacket = (Packet*)( m_pbyPackets + (nIndex * m_nPacketSize) ) ;
	if( nSize == 0 )
	{
		pPacket->iSize = (INT64)m_nDataSize ;
	}
	else
	{
		pPacket->iSize = (INT64)nSize ;
	}
	::CopyMemory( pPacket->aData, pvSrc, (SIZE_T)pPacket->iSize ) ;

	// キューに追加
	ATOMIC_SET_PTR( m_ppvQueue[nIndex], pPacket ) ;

	/***** リーダーロック用セマフォ１個インクリメント ****/
	if( m_hsmRead != NULL )
	{
		::ReleaseSemaphore( m_hsmRead, 1, NULL ) ;
	}

	if ( m_plkWrite != NULL )
	{
		m_plkWrite->Unlock() ;
	}

	return TRUE ;
}


/********************************************************************************
*	関　数　名：Pop																*
*	用　　　途：待ち行列からデータを取得										*
*	引　き　数：キューデータ格納先、タイムアウト時間[ms]						*
*	戻　り　数：取得成功でデータサイズ、失敗で-1								*
*	備　　　考：アクセスモードがRQ_MT_BLOCKING_READERの場合は、					*
*				タイムアウト値が有効											*
*				そうでない場合は、タイムアウト値は無効							*
********************************************************************************/
INT32 CRingQueue::Pop( void*	pvDst,		/* : 取り出し先				    */
					   UINT32	nTimeout	/* : 受け取り待ちタイムリミット */ )
{
	ASSERT( pvDst!= NULL ) ;

	Packet*		pPacket ;		/* パケットバッファ		 */
	INT32		nSize ;			/* 取得サイズ			 */

	/***** 初期化チェック *****/
	if( m_ppvQueue == NULL )					return -1 ;

	/***** リーダーブロッキング(マルチスレッド) *****/
	if( m_hsmRead != NULL )
	{
		/***** バッファ待ちと送受信停止確認及び
			   リーダーロック用セマフォ１個デクリメント ****/
		DWORD	dwWait = ::WaitForSingleObject( m_hsmRead, (DWORD)nTimeout ) ;

			 if( dwWait == WAIT_TIMEOUT	 )		return  0 ;
		else if( dwWait != WAIT_OBJECT_0 )		return -1 ;
	}
	/***** シングルスレッド *****/
	//else
	//{
	//}

	/***** パケット取得 *****/
	pPacket = (Packet*)ATOMIC_GET_PTR( m_ppvQueue[m_nReadIndex] ) ;
	if ( pPacket == NULL ) 
	{ 
		if ( ATOMIC_TEST(m_lbShutDown) )
		{
			TRACE( _T("■■■■■■■■■■ キューはシャットダウン(読み書き不可)しました!  ")
			       _T("Pop()の返値には-1が返されます。  ")
			       _T("TRUE / FALSEでないことに注意してください!\n") ) ;
												return -1 ;
		}
												return 0 ;
	}

	nSize = (INT32)pPacket->iSize ;
	ASSERT( nSize>0 && nSize<=(INT32)m_nDataSize ) ;

	::CopyMemory( pvDst, pPacket->aData, (SIZE_T)nSize ) ;
	ATOMIC_SET_PTR( m_ppvQueue[m_nReadIndex], NULL ) ;

	RingIncrement( &m_nReadIndex ) ;

	return nSize ;
}


/********************************************************************************
*	関　数　名：Peek															*
*	用　　　途：待ち行列の先頭データを取得										*
*	引　き　数：キューデータ格納先												*
*	戻　り　数：取得成功でデータサイズ、失敗で-1								*
*	備　　　考：覗き見機能なのでPopと異なりキューからは外されない				*
*				引数の格納先が０の場合は、データの有無・サイズのみの確認になる	*
********************************************************************************/
INT32 CRingQueue::Peek( void* pvDst /* : 取り出し先 */ )
{
	Packet*		pPacket ;		/* パケットバッファ		 */
	INT32		nSize ;		 	/* 取得サイズ			 */

	/***** 初期化チェック *****/
	if( m_ppvQueue == NULL )							return -1 ;

	/***** 送受信停止確認 *****/
	//if( ATOMIC_TEST(m_lbShutDown) )					return -1 ;

	/***** パケット取得 *****/
	pPacket = (Packet*)ATOMIC_GET_PTR( m_ppvQueue[m_nReadIndex] ) ;
	if ( pPacket == NULL )								return 0 ; 

	nSize = (INT32)pPacket->iSize ;
	ASSERT( nSize>0 && nSize<=(INT32)m_nDataSize ) ;

	if( pvDst != NULL )	
	{
		::CopyMemory( pvDst, pPacket->aData, (SIZE_T)nSize ) ;
	}

	return nSize ;
}

/********************************************************************************
*	関　数　名：Peek															*
*	用　　　途：最後にPushしたデータから任意位置のデータを取得					*
*	引　き　数：キューデータ格納先												*
*	戻　り　数：取得成功でデータサイズ、失敗で-1								*
*	備　　　考：覗き見機能なのでPopと異なりキューからは外されない				*
*				引数の格納先が０の場合は、データの有無・サイズのみの確認になる	*
*				対象データがキューに格納されていない場合は失敗(-1)する			*
********************************************************************************/
INT32 CRingQueue::Peek( INT32 nPeekIndexNum, void* pvDst )
{
	Packet*		pPacket ;		/* パケットバッファ		 */
	INT32		nSize ;		 	/* 取得サイズ			 */
	INT32		nPeekIndex ;	/* 参照先インデックス		 */

	/***** 初期化チェック *****/
	if( m_ppvQueue == NULL )									return -1 ;

	/***** 参照インデックス取得 *****/
	nPeekIndex = m_nWriteIndex;
	RingDecrement( nPeekIndexNum, &nPeekIndex );

	/***** パケット取得 *****/
	pPacket = (Packet*)ATOMIC_GET_PTR( m_ppvQueue[nPeekIndex] ) ;
	if ( pPacket == NULL )										return 0 ; 

	nSize = (INT32)pPacket->iSize ;
	ASSERT( nSize>0 && nSize<=(INT32)m_nDataSize ) ;

	if( pvDst != NULL )	
	{
		::CopyMemory( pvDst, pPacket->aData, (SIZE_T)nSize ) ;
	}

	return nSize ;
}


