ACE/apps/JAWS3/jaws3/THYBRID_Concurrency.cpp

   1 #include "ace/OS_NS_stdlib.h"
   2 #include "ace/Message_Block.h"
   3
   4 #ifndef JAWS_BUILD_DLL
   5 #define JAWS_BUILD_DLL
   6 #endif
   7
   8 #include "jaws3/Concurrency.h"
   9 #include "jaws3/THYBRID_Concurrency.h"
  10 #include "jaws3/Protocol_Handler.h"
  11 #include "jaws3/Options.h"
  12
  13
  14 JAWS_THYBRID_Concurrency::JAWS_THYBRID_Concurrency ()
  15   : getting_ (0)
  16   , min_number_of_threads_ (1)
  17   , max_number_of_threads_ (-1)
  18   , shutdown_task_ (0)
  19   , error_ (0)
  20 {
  21   const char *value;
  22
  23   value = JAWS_Options::instance ()->getenv ("JAWS_MIN_THYBRID_THREADS");
  24   if (value != 0)
  25     this->min_number_of_threads_ = ACE_OS::atoi (value);
  26   else
  27     this->min_number_of_threads_ =
  28       ACE_OS::atoi (JAWS_DEFAULT_MIN_THYBRID_THREADS);
  29
  30   if (this->min_number_of_threads_ <= 0)
  31     this->min_number_of_threads_ = 1;
  32
  33   value = JAWS_Options::instance ()->getenv ("JAWS_MAX_THYBRID_THREADS");
  34   if (value != 0)
  35     this->max_number_of_threads_ = ACE_OS::atoi (value);
  36   else
  37     this->max_number_of_threads_ =
  38       ACE_OS::atoi (JAWS_DEFAULT_MAX_THYBRID_THREADS);
  39
  40   if (this->max_number_of_threads_ <= 0)
  41     this->max_number_of_threads_ = -1;
  42   else if (this->max_number_of_threads_ < this->min_number_of_threads_)
  43     this->max_number_of_threads_ = this->min_number_of_threads_;
  44
  45   int r;
  46   r = this->activate (THR_BOUND | THR_JOINABLE, this->min_number_of_threads_);
  47   if (r < 0)
  48     {
  49       this->shutdown_task_ = 1;
  50       this->error_ = 1;
  51     }
  52 }
  53
  54 int
  55 JAWS_THYBRID_Concurrency::putq (JAWS_Protocol_Handler *ph)
  56 {
  57   if (this->error_)
  58     return -1;
  59
  60   JAWS_CONCURRENCY_TASK *task = this;
  61   int result = task->putq (& ph->mb_);
  62
  63   if (result != -1)
  64     {
  65       if (this->getting_ < this->min_number_of_threads_
  66           && (this->max_number_of_threads_ < 0
  67               || this->thr_count () < (size_t) this->max_number_of_threads_))
  68         {
  69           int r;
  70           r = this->activate ( THR_BOUND | THR_JOINABLE
  71                              , 1 // number of threads
  72                              , 1 // force active
  73                              );
  74           if (r < 0)
  75             {
  76               // ACE_ERROR
  77               return -1;
  78             }
  79         }
  80     }
  81
  82   return result;
  83 }
  84
  85 int
  86 JAWS_THYBRID_Concurrency::getq (JAWS_Protocol_Handler *&ph)
  87 {
  88   ph = 0;
  89
  90   JAWS_CONCURRENCY_TASK *task = this;
  91
  92   if (this->shutdown_task_ && task->msg_queue ()->message_count () == 0)
  93     return -1;
  94
  95   int getting = ++(this->getting_);
  96
  97   if (getting > this->min_number_of_threads_)
  98     {
  99       if (task->msg_queue ()->message_count () == 0)
 100         {
 101           --(this->getting_);
 102           return -1;
 103         }
 104     }
 105
 106   ACE_Message_Block *mb = 0;
 107   int result = task->getq (mb);
 108
 109   if (result != -1)
 110     {
 111       ph = (JAWS_Protocol_Handler *) mb->base ();
 112
 113       if (ph == 0)
 114         {
 115           // Shutdown this task;
 116           this->shutdown_task_ = 1;
 117           if (this->getting_ > 1)
 118             {
 119               task->putq (mb);
 120               result = -1;
 121             }
 122
 123         }
 124     }
 125
 126   --(this->getting_);
 127   return result;
 128 }