TAO/tests/Bug_1361_Regression/Server_Thread_Pool.cpp

   1 #include "Server_Thread_Pool.h"
   2 #include "TestC.h"
   3 #include "ace/OS_NS_unistd.h"
   4
   5 time_t last_success;
   6
   7 int
   8 Thread_Pool::close (u_long)
   9 {
  10   ACE_DEBUG ((LM_DEBUG,
  11               "(%t) worker thread closing down\n"));
  12   return 0;
  13 }
  14
  15 Thread_Pool::Thread_Pool (CORBA::ORB_ptr orb,
  16                           ACE_Thread_Manager *thr_mgr,
  17                           int n_threads)
  18   : ACE_Task<ACE_SYNCH> (thr_mgr),
  19   orb_(CORBA::ORB::_duplicate(orb)),
  20   nt_(n_threads)
  21 {
  22   if (this->activate (THR_NEW_LWP,
  23                       n_threads) == -1)
  24     ACE_ERROR ((LM_ERROR,
  25                 "%p\n",
  26                 "activate failed\n"));
  27 }
  28
  29 Thread_Pool::~Thread_Pool ()
  30 {
  31 }
  32
  33 int
  34 Thread_Pool::shutdown ()
  35 {
  36   thr_mgr_->cancel_grp (grp_id_);
  37
  38   int n_threads = nt_.value ();
  39   for (int i = 0; i < n_threads; ++i)
  40     {
  41       ACE_DEBUG ((LM_DEBUG,
  42                   "(%t) eof, sending block for thread=%d\n",
  43                   i + 1));
  44       ACE_Message_Block *mb1;
  45       ACE_NEW_RETURN (mb1,
  46                       ACE_Message_Block ((char*) Test::Echo::_nil ()),
  47                       -1);
  48       mb1->length (0);
  49
  50       if (this->put (mb1) == -1)
  51         ACE_ERROR ((LM_ERROR,
  52                     "(%t) %p\n",
  53                     "put"));
  54
  55       // this sleep helps to shutdown correctly -> was an error!
  56       ACE_OS::sleep (1);
  57     }
  58
  59   return 0;
  60 }
  61
  62 // Simply enqueue the Message_Block into the end of the queue.
  63 int
  64 Thread_Pool::put (Test::Echo_ptr echoptr)
  65 {
  66   char * charData = (char *) Test::Echo::_duplicate (echoptr);
  67
  68   ACE_Message_Block *mb = 0;
  69   ACE_NEW_RETURN(mb, ACE_Message_Block(charData), -1);
  70   return this->put (mb);
  71 }
  72
  73 int
  74 Thread_Pool::put (ACE_Message_Block *mb,
  75                   ACE_Time_Value *tv)
  76 {
  77   return this->putq (mb, tv);
  78 }
  79
  80 // Iterate <n_iterations> time printing off a message and "waiting"
  81 // for all other threads to complete this iteration.
  82
  83 int
  84 Thread_Pool::svc ()
  85 {
  86   // Note that the <ACE_Task::svc_run> method automatically adds us to
  87   // the Thread_Manager when the thread begins.
  88
  89   int count = 1;
  90
  91   // Keep looping, reading a message out of the queue, until we get a
  92   // message with a length == 0, which signals us to quit.
  93
  94   for (;; count++)
  95     {
  96       ACE_Message_Block *mb = 0;
  97
  98 #if 0
  99       ACE_DEBUG ((LM_DEBUG,
 100                   "(%t) in iteration %d before getq ()\n",
 101                   count));
 102 #endif
 103
 104       if (this->getq (mb) == -1)
 105         {
 106           ACE_ERROR ((LM_ERROR,
 107                       "(%t) in iteration %d, got result -1, exiting\n",
 108                       count));
 109           break;
 110         }
 111
 112 #if 0
 113       if (mb->length() == 0)
 114         {
 115           //FUZZ: disable check_for_NULL
 116           ACE_DEBUG ((LM_DEBUG,
 117                       "(%t) in iteration %d, got NULL message, exiting\n",
 118                       count));
 119           //FUZZ: enable check_for_NULL
 120           break;
 121         }
 122 #endif
 123
 124       Test::Echo_var echo = (Test::Echo_ptr)mb->base();
 125
 126       mb->release ();
 127
 128       if (CORBA::is_nil(echo.in()))
 129         {
 130           //FUZZ: disable check_for_NULL
 131           ACE_DEBUG ((LM_DEBUG,
 132                       "(%t) in iteration %d, got NULL message, exiting\n",
 133                       count));
 134           //FUZZ: enable check_for_NULL
 135           break;
 136         }
 137
 138       // Keep calling a few times after receiving exceptions
 139       for(int exception_count = 50; exception_count; --exception_count)
 140         {
 141           try
 142             {
 143               // keep calling until get an exception
 144               while (true)
 145                 {
 146 #if 0
 147                   if (0)
 148                     {
 149                       Test::Payload pload (10);
 150                       pload.length (10);
 151                       ACE_OS::memset (pload.get_buffer(), pload.length(), 0);
 152                       echo->echo_payload (pload);
 153                     }
 154                   else
 155 #endif /*if 0*/
 156                     {
 157                       Test::Payload_var pout;
 158                       echo->echo_payload_out (pout.out());
 159
 160                       // time_t last_success = ACE_OS::time();
 161                     }
 162               }
 163             }
 164           catch (const CORBA::Exception&)
 165             {
 166               // Just forget the exception and continue
 167             }
 168         }
 169
 170     }
 171
 172   --nt_;
 173   if (nt_ == 0)
 174     {
 175       orb_->shutdown (false);
 176     }
 177
 178   // Note that the <ACE_Task::svc_run> method automatically removes us
 179   // from the <ACE_Thread_Manager> when the thread exits.
 180   return 0;
 181 }