/home/bes3soft/bes3soft/Boss/7.0.2/dist/7.0.2/DistBoss/DistBossUtil/DistBossUtil-00-00-04/DistBossUtil/template/PthrWriterBufPool.cc

Go to the documentation of this file.

/**********************************************************
  +-> 1      2      3      4      5      6  ...  PoolSize
  |               ahead current  next               |
  |_________________________________________________|
 
  0, a looped buffer pool
  1, we set semIn as (PoolSize-1), so to protect the
  current event buffer not be changed by writeEvent
  2, we can't update the current event buffer until we
  move to the next event(in thread_writing()).
  3, the ahead event buffer is ready to update when we
  at the beginning of current event.
**********************************************************/

#include <iostream>


template <typename Writer, int PoolSize>
PthrWriterBufPool<Writer, PoolSize>::PthrWriterBufPool(typename Writer::WriterArgType& arg)
   : m_inLoop(-1),
     m_outLoop(-1)
{
   if ( PoolSize < 2 ) {
      std::cout << "[PthrWriterBufPool] The PoolSize of buffer must > 1" << std::endl;
      exit(1);
   }

   // use(PoolSize-1) to protect the current event buffer
   sem_init(&m_semIn,  0, (PoolSize-1));
   sem_init(&m_semOut, 0, 0);
   sem_init(&m_semFinalize, 0, 0);

   for ( int i = 0; i < PoolSize; ++i ) {
      // buffer in the loop for each event
      m_buf[i] = new AutoEnlargeBuffer(128*1024);
   }

   m_writer = new Writer(arg);

   // create an individual thread for writing
   pthread_create(&m_tid, NULL, thread_writing, (void*)this);
}

template <typename Writer, int PoolSize>
PthrWriterBufPool<Writer, PoolSize>::~PthrWriterBufPool()
{
   // wait for the end of thread_writing
   sem_wait(&m_semFinalize);

   for ( int i = 0; i < PoolSize; ++i ) {
      delete m_buf[i];
   }

   sem_destroy(&m_semIn);
   sem_destroy(&m_semOut);
   sem_destroy(&m_semFinalize);

   delete m_writer;
}

template <typename Writer, int PoolSize>
int PthrWriterBufPool<Writer, PoolSize>::writeEvent(void* pevt, int size)
{
   if ( m_writer->stat() != 0 ) return m_writer->stat();

   sem_wait(&m_semIn);

   int index = (++m_inLoop) % PoolSize;

   m_buf[index]->copy( pevt, size );

   sem_post(&m_semOut);

   return 0;
}

template <typename Writer, int PoolSize>
int PthrWriterBufPool<Writer, PoolSize>::stat()
{
   return m_writer->stat();
}

template <typename Writer, int PoolSize>
void* PthrWriterBufPool<Writer, PoolSize>::thread_writing(void* arg)
{
   PthrWriterBufPool<Writer, PoolSize>* pthis = (PthrWriterBufPool<Writer, PoolSize>*)arg;

   uint32_t        index;
   AutoEnlargeBuffer* pbuf;


   while ( true ) {
      // the ahead event buffer is ready for update now
      sem_post(&(pthis->m_semIn));
      // waiting for an event that ready to use
      sem_wait(&(pthis->m_semOut));

      index = (++(pthis->m_outLoop)) % PoolSize;

      pbuf = pthis->m_buf[index];

      try {
         pthis->m_writer->writeEvent( pbuf->data(), pbuf->size() );
      }
      catch (ReachEndOfFileList& e) {
         //std::cout << "[PthrWriterBufPool] Finalized Successfully!" << std::endl;
         break;
      }
      catch (RawExMessage& e) {
         e.print();
         break;
      }
      catch ( ... ) {
         std::cout << "[PthrWriterBufPool] Catch unexpected exception !" << std::endl;
         exit(1);
      }
   }


   pthis->m_writer->close();

   sem_post(&(pthis->m_semFinalize));

   pthread_exit(NULL);
}

---