ACE/apps/JAWS3/jaws3/Concurrency.cpp

   1 #include "ace/OS_NS_strings.h"
   2 #include "ace/Message_Block.h"
   3 #include "ace/Singleton.h"
   4 #include "ace/Synch_Traits.h"
   5 #include "ace/Null_Mutex.h"
   6
   7 #ifndef JAWS_BUILD_DLL
   8 #define JAWS_BUILD_DLL
   9 #endif
  10
  11 #include "jaws3/Concurrency.h"
  12 #include "jaws3/TPOOL_Concurrency.h"
  13 #include "jaws3/TPR_Concurrency.h"
  14 #include "jaws3/THYBRID_Concurrency.h"
  15 #include "jaws3/Options.h"
  16
  17
  18 typedef ACE_Singleton<ACE_Message_Block, ACE_SYNCH_NULL_MUTEX>
  19         JAWS_Empty_Message_Block;
  20
  21
  22 int
  23 JAWS_Concurrency_Impl::svc ()
  24 {
  25   JAWS_Protocol_Handler *ph;
  26
  27   while (this->getq (ph) != -1)
  28     {
  29       if (ph == 0)
  30         continue;
  31
  32       if (ph->service () == -1)
  33         {
  34           // TODO: Do I really want to call close here, or
  35           //       let the protocol handler close itself?
  36           ph->dismiss ();
  37           continue;
  38         }
  39     }
  40
  41   return 0;
  42 }
  43
  44
  45 JAWS_Concurrency_Bridge<JAWS_Concurrency_Impl>
  46 ::JAWS_Concurrency_Bridge (JAWS_Concurrency_Impl *impl)
  47   : impl_ (impl)
  48 {
  49   if (this->impl_ == 0)
  50     {
  51       const char *concurrency;
  52       concurrency = JAWS_Options::instance ()->getenv ("JAWS_CONCURRENCY");
  53       if (concurrency == 0)
  54         concurrency = JAWS_DEFAULT_CONCURRENCY;
  55
  56       if (ACE_OS::strcasecmp (concurrency, "TPR") == 0)
  57         this->impl_ = JAWS_TPR_Concurrency::instance ();
  58       else if (ACE_OS::strcasecmp (concurrency, "TPOOL") == 0)
  59         this->impl_ = JAWS_TPOOL_Concurrency::instance ();
  60       else if (ACE_OS::strcasecmp (concurrency, "THYBRID") == 0)
  61         this->impl_ = JAWS_THYBRID_Concurrency::instance ();
  62       else
  63         this->impl_ = JAWS_THYBRID_Concurrency::instance ();
  64         // Since synchronous IO is the default IO, need an aggressive
  65         // default concurrency mechanism.
  66     }
  67 }
  68
  69
  70 int
  71 JAWS_Concurrency_Bridge<JAWS_Concurrency_Impl>
  72 ::putq (JAWS_Protocol_Handler *ph)
  73 {
  74   return this->impl_->putq (ph);
  75 }
  76
  77
  78 int
  79 JAWS_Concurrency_Bridge<JAWS_Concurrency_Impl>
  80 ::getq (JAWS_Protocol_Handler *&ph)
  81 {
  82   return this->impl_->getq (ph);
  83 }
  84
  85
  86 void
  87 JAWS_Concurrency_Bridge<JAWS_Concurrency_Impl>::shutdown ()
  88 {
  89   ACE_Message_Block *empty_mb = JAWS_Empty_Message_Block::instance ();
  90   JAWS_CONCURRENCY_TASK *task;
  91
  92   task = JAWS_THYBRID_Concurrency::instance ();
  93   task->putq (empty_mb);
  94   task->wait ();
  95
  96   task = JAWS_TPOOL_Concurrency::instance ();
  97   task->putq (empty_mb);
  98   task->wait ();
  99
 100   task = JAWS_TPR_Concurrency::instance ();
 101   task->putq (empty_mb);
 102   task->wait ();
 103 }