ACE/examples/APG/ThreadSafety/Semaphore.cpp

   1 #include "ace/config-lite.h"
   2 #if defined (ACE_HAS_THREADS)
   3
   4 #include "ace/OS_NS_string.h"
   5 #include "ace/Task.h"
   6 #include "ace/Synch.h"
   7
   8 // Listing 2 code/ch14
   9 class Consumer : public ACE_Task<ACE_MT_SYNCH>
  10 {
  11 public:
  12   enum { N_THREADS = 5 };
  13
  14   Consumer (ACE_Semaphore& psema, ACE_Semaphore& csema)
  15     : psema_(psema), csema_(csema), exit_condition_(0)
  16   { }
  17
  18   int svc ()
  19   {
  20     while (!is_closed ())
  21       consume_item ();
  22     return 0;
  23   }
  24
  25   void consume_item ()
  26   {
  27     csema_.acquire ();
  28     if (!is_closed ())
  29       {
  30         ACE_Message_Block *mb = 0;
  31         this->getq (mb);
  32         if (mb->msg_type () == ACE_Message_Block::MB_HANGUP)
  33           {
  34             this->shutdown ();
  35             mb->release ();
  36             return;
  37           }
  38         else
  39           {
  40             ACE_DEBUG ((LM_DEBUG,
  41                         ACE_TEXT ("(%t) Consumed %d\n"),
  42                         *((int*)mb->rd_ptr ())));
  43             mb->release();
  44           }
  45         psema_.release ();
  46       }
  47   }
  48
  49   //FUZZ: disable check_for_lack_ACE_OS
  50   void shutdown ()
  51   {
  52   //FUZZ: enable check_for_lack_ACE_OS
  53     exit_condition_ = 1;
  54     this->msg_queue ()->deactivate ();
  55     csema_.release (N_THREADS);
  56   }
  57
  58   int is_closed ()
  59   {
  60     return exit_condition_;
  61   }
  62
  63 private:
  64   ACE_Semaphore& psema_;
  65   ACE_Semaphore& csema_;
  66   int exit_condition_;
  67 };
  68 // Listing 2
  69 // Listing 1 code/ch14
  70 class Producer : public ACE_Task_Base
  71 {
  72 public:
  73   enum { MAX_PROD = 128 };
  74
  75   Producer (ACE_Semaphore& psema, ACE_Semaphore& csema,
  76             Consumer &consumer)
  77       : psema_(psema), csema_(csema), consumer_(consumer)
  78   { }
  79
  80   int svc ()
  81   {
  82     for (int i = 0; i <= MAX_PROD; i++)
  83       produce_item (i);
  84     hang_up ();
  85     return 0;
  86   }
  87
  88   void produce_item (int item)
  89   {
  90     psema_.acquire ();
  91     ACE_Message_Block *mb
  92       = new ACE_Message_Block (sizeof (int),
  93                                ACE_Message_Block::MB_DATA);
  94     ACE_OS::memcpy (mb->wr_ptr (), &item, sizeof item);
  95     mb->wr_ptr (sizeof (int));
  96     this->consumer_.putq (mb);
  97
  98     ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) Produced %d\n"), item));
  99     csema_.release();
 100   }
 101
 102   void hang_up ()
 103   {
 104     psema_.acquire ();
 105     ACE_Message_Block *mb =
 106       new ACE_Message_Block (0, ACE_Message_Block::MB_HANGUP);
 107     this->consumer_.putq (mb);
 108     csema_.release ();
 109   }
 110
 111 private:
 112   ACE_Semaphore& psema_;
 113   ACE_Semaphore& csema_;
 114   Consumer& consumer_;
 115 };
 116 // Listing 1
 117 // Listing 3 code/ch14
 118 int ACE_TMAIN (int, ACE_TCHAR *[])
 119 {
 120   ACE_Semaphore psem (5);
 121   ACE_Semaphore csem (0);
 122
 123   Consumer consumer (psem, csem);
 124   Producer producer (psem, csem, consumer);
 125
 126   producer.activate ();
 127   consumer.activate (THR_NEW_LWP | THR_JOINABLE,
 128                      Consumer::N_THREADS);
 129
 130   producer.wait ();
 131   consumer.wait ();
 132
 133   return 0;
 134 }
 135 // Listing 3
 136
 137 #else
 138 #include "ace/OS_main.h"
 139 #include "ace/OS_NS_stdio.h"
 140
 141 int ACE_TMAIN (int, ACE_TCHAR *[])
 142 {
 143   ACE_OS::puts (ACE_TEXT ("This example requires threads."));
 144   return 0;
 145 }
 146
 147 #endif /* ACE_HAS_THREADS */