ACE学習(五)ACE_Semaphore

この信号量メカニズムはwin 32とlinuxでサポートされており、特に感じません.直接Demoは以下のように、このDemoも古典的な生産者消費者の実現である.
 

// thread_semaphore.cpp

#include "ace/Task.h"
#include "ace/Semaphore.h"


class Consumer:public ACE_Task <ACE_MT_SYNCH>
{
public:
 enum{ N_THREADS = 5 };

 Consumer(ACE_Semaphore& psema, ACE_Semaphore& csema):m_psema(psema), m_csema(csema), m_exitCondition(0)
 {}

 int isClose()
 {
  return m_exitCondition;
 }

 int svc(void)
 {
  while (!isClose())
  {
   consumeItem();
  }
  return 0;
 }


 void consumeItem()
 {
  m_csema.acquire();
  if (!isClose())
  {
   ACE_Message_Block *mb;
   this->getq(mb);
   if (mb->msg_type() == ACE_Message_Block::MB_HANGUP)
   {
    shutdown();
    mb->release();
    return;
   }
   else
   {
    ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%t) Consumed %d 
"), *((int *)mb->rd_ptr())));
    mb->release();
   }
   m_psema.release();
  }
 }

 void shutdown(void)
 {
  m_exitCondition = 1;
  this->msg_queue()->deactivate();
  m_csema.release(N_THREADS);
 }

private:
 ACE_Semaphore& m_psema;
 ACE_Semaphore& m_csema;
 int m_exitCondition;

};

class Producer:public ACE_Task_Base
{
public:
 enum{MAX_PROD = 128};

 Producer(ACE_Semaphore &psema, ACE_Semaphore &csema, Consumer &consumer)
  :m_psema(psema), m_csema(csema), m_consumer(consumer)
 {}

 void producItem(int item)
 {
  m_psema.acquire();
  ACE_Message_Block *mb = new ACE_Message_Block(sizeof(int),ACE_Message_Block::MB_DATA);
  ACE_OS::memcpy(mb->wr_ptr(), &item, sizeof(item));
  mb->wr_ptr(sizeof(int));
  this->m_consumer.putq(mb);

  ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%t) Prodeced %d
"), item));
  m_csema.release();
 }

 int svc(void)
 {
  for (int i=0; i<=MAX_PROD; i++)
  {
   producItem(i);
  }
  hangUp();
  return 0;
 }

 void hangUp()
 {
  m_psema.acquire();
  ACE_Message_Block *mb = new ACE_Message_Block(0, ACE_Message_Block::MB_HANGUP);
  this->m_consumer.putq(mb);
  m_csema.release();
 }

private:
 ACE_Semaphore& m_psema;
 ACE_Semaphore& m_csema;
 Consumer& m_consumer;
};

 

int ACE_TMAIN(int, ACE_TCHAR *[])
{
 ACE_Semaphore psem(5);
 ACE_Semaphore csem(0);

 Consumer consumer(psem, csem);
 Producer producer(psem, csem, consumer);

 producer.activate();
 consumer.activate(THR_NEW_LWP | THR_JOINABLE, Consumer::N_THREADS);

 producer.wait();
 consumer.wait();

 return 0;
};