#include <PthrReaderBufPool.h>
Inheritance diagram for PthrReaderBufPool< Reader, PoolSize >:
Public Member Functions | |
| const uint32_t * | currentEvent () const |
| const uint32_t * | currentEvent () const |
| std::string | currentFile () |
| std::string | currentFile () |
| const uint32_t * | nextEvent () |
| const uint32_t * | nextEvent () |
| PthrReaderBufPool (typename Reader::ReaderArgType &arg) | |
| PthrReaderBufPool (typename Reader::ReaderArgType &arg) | |
| uint32_t | stat () |
| uint32_t | stat () |
| virtual | ~PthrReaderBufPool () |
| virtual | ~PthrReaderBufPool () |
Private Member Functions | |
| PthrReaderBufPool () | |
| PthrReaderBufPool () | |
Static Private Member Functions | |
| void * | thread_filling (void *arg) |
| void * | thread_filling (void *arg) |
Private Attributes | |
| AutoEnlargeBuffer * | m_buf [PoolSize] |
| AutoEnlargeBuffer * | m_buf [PoolSize] |
| int | m_inLoop |
| pthread_mutex_t | m_lock |
| int | m_outLoop |
| RawFileException * | m_RawFileException |
| RawFileException * | m_RawFileException |
| Reader * | m_reader |
| Reader * | m_reader |
| sem_t | m_semIn |
| sem_t | m_semOut |
| sem_t | m_semSyn |
| pthread_t | m_tid |
|
||||||||||
|
00020 : m_inLoop(-1), 00021 m_outLoop(-1), 00022 m_RawFileException(0) 00023 { 00024 if ( PoolSize < 2 ) { 00025 std::cout << "[PthrReaderBufPool] The PoolSize of buffer must > 1" << std::endl; 00026 exit(1); 00027 } 00028 00029 // use(PoolSize-1) to protect the current event buffer 00030 sem_init(&m_semIn, 0, (PoolSize-1)); 00031 sem_init(&m_semOut, 0, 0); 00032 pthread_mutex_init(&m_lock, NULL); 00033 00034 sem_init(&m_semSyn, 0, 0); 00035 00036 for ( int i = 0; i < PoolSize; ++i ) { 00037 // buffer in the loop for each event 00038 m_buf[i] = new AutoEnlargeBuffer(128*1024); 00039 } 00040 00041 m_reader = new Reader(arg); 00042 00043 // create an individual thread to fill to buffer 00044 pthread_create(&m_tid, NULL, thread_filling, (void*)this); 00045 }
|
|
|||||||||
|
00049 {
00050 for ( int i = 0; i < PoolSize; ++i ) {
00051 delete m_buf[i];
00052 }
00053
00054 delete m_RawFileException;
00055
00056 sem_destroy(&m_semIn);
00057 sem_destroy(&m_semOut);
00058
00059 delete m_reader;
00060 }
|
|
|||||||||
|
|
|
||||||||||
|
|
|
|||||||||
|
|
|
|||||||||
|
|
|
|||||||||
|
Implements IRawReader. |
|
|||||||||
|
Implements IRawReader. 00093 {
00094 //can't be called before any nextEvent() call
00095 const uint32_t* pevt = (uint32_t*)m_buf[ m_outLoop%PoolSize ]->data();
00096 return pevt;
00097 }
|
|
|||||||||
|
Implements IRawReader. |
|
|||||||||
|
Implements IRawReader. 00101 {
00102 return m_reader->currentFile();
00103 }
|
|
|||||||||
|
Implements IRawReader. |
|
|||||||||
|
Implements IRawReader. 00064 {
00065 if ( m_outLoop < 0 ) sem_post(&m_semSyn);
00066
00067 // the ahead event buffer is ready for update now
00068 sem_post(&m_semIn);
00069 // waiting for a ready to use event
00070 sem_wait(&m_semOut);
00071
00072 pthread_mutex_lock(&m_lock);
00073 int inLoop = m_inLoop;
00074 pthread_mutex_unlock(&m_lock);
00075
00076 if ( m_RawFileException != 0 && m_outLoop >= inLoop ) {
00077 if ( dynamic_cast<RawExMessage*>(m_RawFileException) ) {
00078 throw RawExMessage(*(RawExMessage*)m_RawFileException);
00079 }
00080 if ( dynamic_cast<ReachEndOfFileList*>(m_RawFileException) ) {
00081 throw ReachEndOfFileList(*(ReachEndOfFileList*)m_RawFileException);
00082 }
00083 }
00084
00085 ++m_outLoop;
00086
00087 const uint32_t* pevt = (uint32_t*)m_buf[ m_outLoop%PoolSize ]->data();
00088 return pevt;
00089 }
|
|
|||||||||
|
Implements IRawReader. |
|
|||||||||
|
Implements IRawReader. 00107 {
00108 return m_reader->stat();
00109 }
|
|
||||||||||
|
|
|
||||||||||
|
00113 {
00114 PthrReaderBufPool<Reader, PoolSize>* pthis = (PthrReaderBufPool<Reader, PoolSize>*)arg;
00115
00116 uint32_t index;
00117 const uint32_t* pevt;
00118
00119 sem_wait(&(pthis->m_semSyn));
00120
00121 while ( true ) {
00122 // waiting for a buffer that ready to update
00123 sem_wait(&(pthis->m_semIn));
00124
00125 try {
00126 pevt = pthis->m_reader->nextEvent();
00127 }
00128 catch (RawExMessage& e) {
00129 pthis->m_RawFileException = new RawExMessage(e);
00130 break;
00131 }
00132 catch (ReachEndOfFileList& e) {
00133 pthis->m_RawFileException = new ReachEndOfFileList(e);
00134 break;
00135 }
00136 catch ( ... ) {
00137 std::cout << "[PthrReaderBufPool] Catch unexpected exception !" << std::endl;
00138 exit(1);
00139 }
00140
00141 pthread_mutex_lock(&(pthis->m_lock));
00142 index = (++(pthis->m_inLoop)) % PoolSize;
00143 pthread_mutex_unlock(&(pthis->m_lock));
00144
00145 pthis->m_buf[index]->copy( (void*)pevt, pevt[1]*sizeof(uint32_t) );
00146
00147 // post an event that ready to use
00148 sem_post(&(pthis->m_semOut));
00149 }
00150
00151 sem_post(&(pthis->m_semOut));
00152
00153 pthread_exit(NULL);
00154 }
|
|
|||||
|
|
|
|||||
|
|
|
|||||
|
|
|
|||||
|
|
|
|||||
|
|
|
|||||
|
|
|
|||||
|
|
|
|||||
|
|
|
|||||
|
|
|
|||||
|
|
|
|||||
|
|
|
|||||
|
|
|
|||||
|
|
1.3.9.1