TAO/tao/Dynamic_TP/DTP_Task.cpp

   1 #include "tao/Dynamic_TP/DTP_Task.h"
   2
   3 #if defined (TAO_HAS_CORBA_MESSAGING) && TAO_HAS_CORBA_MESSAGING != 0
   4
   5 #include "tao/CSD_ThreadPool/CSD_TP_Request.h"
   6 #include "tao/CSD_ThreadPool/CSD_TP_Dispatchable_Visitor.h"
   7 #include "tao/CSD_ThreadPool/CSD_TP_Cancel_Visitor.h"
   8
   9 #if !defined (__ACE_INLINE__)
  10 # include "tao/Dynamic_TP/DTP_Task.inl"
  11 #endif /* ! __ACE_INLINE__ */
  12
  13 TAO_BEGIN_VERSIONED_NAMESPACE_DECL
  14
  15 TAO_DTP_Task::TAO_DTP_Task ()
  16   : aw_lock_ (),
  17     queue_lock_ (),
  18     work_lock_ (),
  19     work_available_ (this->work_lock_),
  20     active_workers_ (this->aw_lock_),
  21     active_count_ (0),
  22     accepting_requests_ (false),
  23     shutdown_ (false),
  24     check_queue_ (false),
  25     opened_ (false),
  26     num_queue_requests_ ((size_t)0),
  27     init_pool_threads_ ((size_t)0),
  28     min_pool_threads_ ((size_t)0),
  29     max_pool_threads_ ((size_t)0),
  30     max_request_queue_depth_ ((size_t)0),
  31     thread_stack_size_ ((size_t)0)
  32 {
  33 }
  34
  35 TAO_DTP_Task::~TAO_DTP_Task()
  36 {
  37 }
  38
  39 bool
  40 TAO_DTP_Task::add_request (TAO::CSD::TP_Request* request)
  41 {
  42   {
  43     ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, guard, this->queue_lock_, false);
  44     ++this->num_queue_requests_;
  45     if ((this->num_queue_requests_ > this->max_request_queue_depth_) &&
  46         (this->max_request_queue_depth_ != 0))
  47       {
  48         this->accepting_requests_ = false;
  49       }
  50
  51     if (!this->accepting_requests_)
  52       {
  53         if (TAO_debug_level > 4)
  54           {
  55             TAOLIB_DEBUG ((LM_DEBUG,
  56                         ACE_TEXT ("TAO (%P|%t) - DTP_Task::add_request() ")
  57                         ACE_TEXT ("not accepting requests.\n")
  58                         ACE_TEXT ("TAO (%P|%t) - DTP_Task::add_request() ")
  59                         ACE_TEXT ("num_queue_requests_ : [%d]\n")
  60                         ACE_TEXT ("TAO (%P|%t) - DTP_Task::add_request() ")
  61                         ACE_TEXT ("max_request_queue_depth_ : [%d]\n"),
  62                         this->num_queue_requests_,
  63                         this->max_request_queue_depth_));
  64           }
  65         --this->num_queue_requests_;
  66         return false;
  67       }
  68
  69     // We have made the decision that the request is going to be placed upon
  70     // the queue_.  Inform the request that it is about to be placed into
  71     // a request queue.  Some requests may not need to do anything in
  72     // preparation of being placed into a queue.  Others, however, may need
  73     // to perfom a "clone" operation on some underlying request data before
  74     // the request can be properly placed into a queue.
  75     request->prepare_for_queue();
  76     this->queue_.put(request);
  77   }
  78   {
  79     ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, guard, this->work_lock_, false);
  80     this->check_queue_ = true;
  81     this->work_available_.signal ();
  82     if (TAO_debug_level > 4 )
  83       {
  84         TAOLIB_DEBUG((LM_DEBUG,
  85                    ACE_TEXT ("TAO (%P|%t) - DTP_Task::add_request() - ")
  86                    ACE_TEXT ("work available\n")));
  87       }
  88   }
  89
  90   return true;
  91 }
  92
  93 size_t
  94 TAO_DTP_Task::get_init_pool_threads ()
  95 {
  96   return this->init_pool_threads_;
  97 }
  98
  99 size_t
 100 TAO_DTP_Task::get_min_pool_threads ()
 101 {
 102   return this->min_pool_threads_;
 103 }
 104
 105 size_t TAO_DTP_Task::get_max_pool_threads ()
 106 {
 107   return this->max_pool_threads_;
 108 }
 109
 110 size_t
 111 TAO_DTP_Task::get_max_request_queue_depth ()
 112 {
 113   return this->max_request_queue_depth_;
 114 }
 115
 116 size_t
 117 TAO_DTP_Task::get_thread_stack_size ()
 118 {
 119   return this->thread_stack_size_;
 120 }
 121
 122 time_t
 123 TAO_DTP_Task::get_thread_idle_time ()
 124 {
 125   return this->thread_idle_time_.sec();
 126 }
 127
 128 int
 129 TAO_DTP_Task::open (void* /* args */)
 130 {
 131   ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, guard, this->aw_lock_, -1);
 132   // We can assume that we are in the proper state to handle this open()
 133   // call as long as we haven't been opened before.
 134   if (this->opened_)
 135     {
 136       return 0;
 137     }
 138
 139   int num = static_cast<int> (this->init_pool_threads_);
 140
 141   if (TAO_debug_level > 4)
 142     {
 143       TAOLIB_DEBUG ((LM_DEBUG,
 144                     ACE_TEXT ("TAO (%P|%t) - DTP_Task::open() initialized with:\n")
 145                     ACE_TEXT ("TAO (%P|%t) - DTP_Task::open() init_threads_ \t\t: [%d]\n")
 146                     ACE_TEXT ("TAO (%P|%t) - DTP_Task::open() min_pool_threads_ \t\t: [%d]\n")
 147                     ACE_TEXT ("TAO (%P|%t) - DTP_Task::open() max_pool_threads_ \t\t: [%d]\n")
 148                     ACE_TEXT ("TAO (%P|%t) - DTP_Task::open() max_request_queue_depth_ \t: [%d]\n")
 149                     ACE_TEXT ("TAO (%P|%t) - DTP_Task::open() thread_stack_size_ \t\t: [%d]\n")
 150                     ACE_TEXT ("TAO (%P|%t) - DTP_Task::open() thread_idle_time_ \t\t: [%d]\n"),
 151                     this->init_pool_threads_,
 152                     this->min_pool_threads_,
 153                     this->max_pool_threads_,
 154                     this->max_request_queue_depth_,
 155                     this->thread_stack_size_,
 156                     this->thread_idle_time_.sec ()));
 157     }
 158
 159   // We can't activate 0 threads.  Make sure this isn't the case.
 160   if (num < 1)
 161     {
 162       if (TAO_debug_level > 0)
 163         {
 164           TAOLIB_ERROR ((LM_ERROR,
 165                          ACE_TEXT ("TAO (%P|%t) DTP_Task::open() failed to open.  ")
 166                          ACE_TEXT ("num_threads (%u) is less-than 1.\n"),
 167                          num));
 168         }
 169       return -1;
 170     }
 171
 172   // Set the busy_threads_ to the number of init_threads
 173   // now. When they startup they will decrement themselves
 174   // as they go into a wait state.
 175
 176   this->busy_threads_ = 0;
 177
 178   // Create the stack size arrays if the stack size is set > 0.
 179
 180   // Activate this task object with 'num' worker threads.
 181   if (this->thread_stack_size_ == 0)
 182     {
 183       if (this->activate (THR_NEW_LWP | THR_DETACHED, num, 1) != 0)
 184         {
 185           TAOLIB_ERROR_RETURN ((LM_ERROR,
 186                              ACE_TEXT ("(%P|%t) DTP_Task::open() failed to activate ")
 187                              ACE_TEXT ("(%d) worker threads.\n"),
 188                              num),
 189                             -1);
 190         }
 191     }
 192   else
 193     {
 194       size_t * stack_sz_arr = new size_t[num];
 195       for (int z = 0; z < num; z++)
 196         {
 197           stack_sz_arr[z] = this->thread_stack_size_;
 198         }
 199
 200       if (this->activate (THR_NEW_LWP | THR_DETACHED,
 201                           num,
 202                           1,
 203                           ACE_DEFAULT_THREAD_PRIORITY,
 204                           -1,
 205                           0,
 206                           0,
 207                           0,
 208                           stack_sz_arr) != 0)
 209         {
 210           TAOLIB_ERROR_RETURN ((LM_ERROR,
 211                              ACE_TEXT ("(%P|%t) DTP_Task::open() failed to activate ")
 212                              ACE_TEXT ("(%d) worker threads.\n"),
 213                              num),
 214                             -1);
 215         }
 216
 217       delete[] stack_sz_arr;
 218     }
 219
 220   if (TAO_debug_level > 4)
 221     {
 222       TAOLIB_DEBUG ((LM_DEBUG,
 223                      ACE_TEXT ("(%P|%t) DTP_Task::open() activated %d initial threads\n"),
 224                      num));
 225     }
 226
 227   this->active_count_ = static_cast<size_t> (num);
 228
 229   this->opened_ = true;
 230   this->accepting_requests_ = true;
 231
 232   return 0;
 233 }
 234
 235 bool
 236 TAO_DTP_Task::request_ready (TAO::CSD::TP_Dispatchable_Visitor &v,
 237                                     TAO::CSD::TP_Request_Handle &r)
 238 {
 239   ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, guard, this->queue_lock_, false);
 240   if (!this->queue_.is_empty())
 241     {
 242       this->queue_.accept_visitor(v);
 243       r = v.request();
 244       return !r.is_nil();
 245     }
 246   return false;
 247 }
 248
 249 void
 250 TAO_DTP_Task::clear_request (TAO::CSD::TP_Request_Handle &r)
 251 {
 252   ACE_GUARD (TAO_SYNCH_MUTEX, guard, this->queue_lock_);
 253   --this->num_queue_requests_;
 254   if (this->max_request_queue_depth_ > 0)
 255     {
 256       this->accepting_requests_ = true;
 257     }
 258
 259   if (TAO_debug_level > 4 )
 260     {
 261       TAOLIB_DEBUG ((LM_DEBUG,
 262                   ACE_TEXT ("TAO (%P|%t) - DTP_Task::clear_request() ")
 263                   ACE_TEXT ("Decrementing num_queue_requests.")
 264                   ACE_TEXT ("New queue depth:%d\n"),
 265                   this->num_queue_requests_));
 266     }
 267
 268   r->mark_as_ready ();
 269 }
 270
 271 void
 272 TAO_DTP_Task::add_busy ()
 273 {
 274   ACE_GUARD (TAO_SYNCH_MUTEX, mon, this->aw_lock_);
 275   ++this->busy_threads_;
 276 }
 277
 278 void
 279 TAO_DTP_Task::remove_busy ()
 280 {
 281   ACE_GUARD (TAO_SYNCH_MUTEX, mon, this->aw_lock_);
 282   --this->busy_threads_;
 283 }
 284
 285 void
 286 TAO_DTP_Task::add_active ()
 287 {
 288   ACE_GUARD (TAO_SYNCH_MUTEX, mon, this->aw_lock_);
 289   ++this->active_count_;
 290 }
 291
 292 bool
 293 TAO_DTP_Task::remove_active (bool force)
 294 {
 295   ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, mon, this->aw_lock_, false);
 296   if (force || this->above_minimum())
 297     {
 298       --this->active_count_;
 299       this->active_workers_.signal ();
 300       return true;
 301     }
 302   return false;
 303 }
 304
 305 bool
 306 TAO_DTP_Task::need_active ()
 307 {
 308   ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, mon, this->aw_lock_, false);
 309   return ((this->busy_threads_ == static_cast<unsigned long> (this->active_count_)) &&
 310           ((this->max_pool_threads_ < 1) ||
 311            (this->active_count_ < this->max_pool_threads_)));
 312 }
 313
 314 bool
 315 TAO_DTP_Task::above_minimum ()
 316 {
 317   return this->min_pool_threads_ > 0 &&
 318     this->active_count_ > this->min_pool_threads_;
 319 }
 320
 321 int
 322 TAO_DTP_Task::svc ()
 323 {
 324   this->add_busy ();
 325   if (TAO_debug_level > 4)
 326     {
 327       TAOLIB_DEBUG ((LM_DEBUG,
 328                   ACE_TEXT ("TAO (%P|%t) - DTP_Task::svc() ")
 329                   ACE_TEXT ("New thread created.\n")));
 330     }
 331   TAO::CSD::TP_Dispatchable_Visitor dispatchable_visitor;
 332   while (!this->shutdown_)
 333     {
 334       TAO::CSD::TP_Request_Handle request;
 335
 336       while (!this->shutdown_ && request.is_nil ())
 337         {
 338           if (!this->request_ready (dispatchable_visitor, request))
 339             {
 340               this->remove_busy ();
 341
 342               if (TAO_debug_level > 4)
 343                 {
 344                   TAOLIB_DEBUG ((LM_DEBUG,
 345                               ACE_TEXT ("TAO (%P|%t) - DTP_Task::svc() ")
 346                               ACE_TEXT ("Decrementing busy_threads_. ")
 347                               ACE_TEXT ("Busy thread count:%d\n"),
 348                               this->busy_threads_.load()));
 349                 }
 350
 351               ACE_Time_Value tmp_sec = this->thread_idle_time_.to_absolute_time();
 352
 353               {
 354                 ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, guard, this->work_lock_, false);
 355                 int wait_state = 0;
 356                 while (!(this->shutdown_ || this->check_queue_) && wait_state != -1)
 357                   {
 358                     wait_state = this->thread_idle_time_.sec () == 0
 359                       ? this->work_available_.wait ()
 360                       : this->work_available_.wait (&tmp_sec);
 361                   }
 362                 // Check for timeout
 363                 if (this->shutdown_)
 364                   return 0;
 365                 if (wait_state == -1)
 366                   {
 367                     if (errno != ETIME || this->remove_active (false))
 368                       {
 369                         if (TAO_debug_level > 4)
 370                           {
 371                             TAOLIB_DEBUG ((LM_DEBUG,
 372                                         ACE_TEXT ("TAO (%P|%t) - DTP_Task::svc() ")
 373                                         ACE_TEXT ("Existing thread expiring.\n")));
 374                           }
 375                         return 0;
 376                       }
 377                   }
 378                 this->check_queue_ = false;
 379               }
 380
 381               this->add_busy ();
 382               if (TAO_debug_level > 4)
 383                 {
 384                   TAOLIB_DEBUG ((LM_DEBUG,
 385                               ACE_TEXT ("TAO (%P|%t) - DTP_Task::svc() ")
 386                               ACE_TEXT ("Incrementing busy_threads_. ")
 387                               ACE_TEXT ("Busy thread count:%d\n"),
 388                               this->busy_threads_.load ()));
 389                 }
 390             }
 391         }
 392
 393       if (this->need_active ())
 394         {
 395           if (this->activate (THR_NEW_LWP | THR_DETACHED,
 396                               1,
 397                               1,
 398                               ACE_DEFAULT_THREAD_PRIORITY,
 399                               -1,
 400                               0,
 401                               0,
 402                               0,
 403                               this->thread_stack_size_ == 0 ? 0 :
 404                               &this->thread_stack_size_) != 0)
 405             {
 406               TAOLIB_ERROR ((LM_ERROR,
 407                              ACE_TEXT ("(%P|%t) DTP_Task::svc() failed to ")
 408                              ACE_TEXT ("grow thread pool.\n")));
 409             }
 410           else
 411             {
 412               this->add_active ();
 413               if (TAO_debug_level > 4)
 414                 {
 415                   TAOLIB_DEBUG ((LM_DEBUG,
 416                                  ACE_TEXT ("TAO (%P|%t) - DTP_Task::svc() ")
 417                                  ACE_TEXT ("Growing threadcount. ")
 418                                  ACE_TEXT ("New thread count:%d\n"),
 419                                  this->thr_count ()));
 420                 }
 421             }
 422         }
 423
 424       request->dispatch ();
 425       this->clear_request (request);
 426       dispatchable_visitor.reset ();
 427     }
 428   this->remove_active (true);
 429   return 0;
 430 }
 431
 432
 433 int
 434 TAO_DTP_Task::close (u_long flag)
 435 {
 436   {
 437     ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, guard, this->aw_lock_, 0);
 438     if (flag == 0)
 439       {
 440         this->active_workers_.signal ();
 441         return 0;
 442       }
 443
 444     if (!this->opened_)
 445       {
 446         return 0;
 447       }
 448     this->opened_ = false;
 449     this->shutdown_ = true;
 450     this->accepting_requests_ = false;
 451   }
 452
 453   {
 454     ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, guard, this->work_lock_, 0);
 455     this->work_available_.broadcast();
 456   }
 457
 458   size_t in_task = (this->thr_mgr ()->task () == this) ? 1 : 0;
 459   if (TAO_debug_level > 4)
 460     {
 461       TAOLIB_DEBUG ((LM_DEBUG,
 462                   ACE_TEXT ("TAO (%P|%t) - DTP_Task::close() ")
 463                   ACE_TEXT ("shutting down. in_task = %d,  Count = %d \n"),
 464                   in_task,  this->thr_count ()));
 465     }
 466
 467   while (this->thr_count () > in_task)
 468     {
 469       ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, guard, this->aw_lock_, 0);
 470       this->active_workers_.wait ();
 471     }
 472
 473   {
 474     ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, guard, this->queue_lock_, 0);
 475     TAO::CSD::TP_Cancel_Visitor v;
 476     this->queue_.accept_visitor (v);
 477   }
 478   return 0;
 479 }
 480
 481
 482 void
 483 TAO_DTP_Task::set_init_pool_threads (size_t thr_count)
 484 {
 485   this->init_pool_threads_ = thr_count;
 486 }
 487
 488 void
 489 TAO_DTP_Task::set_min_pool_threads (size_t thr_count)
 490 {
 491   this->min_pool_threads_ = thr_count;
 492 }
 493
 494 void
 495 TAO_DTP_Task::set_max_pool_threads (size_t thr_count)
 496 {
 497   this->max_pool_threads_ = thr_count;
 498 }
 499
 500 void
 501 TAO_DTP_Task::set_thread_stack_size (size_t stack_sz)
 502 {
 503   this->thread_stack_size_ = stack_sz;
 504 }
 505
 506 void
 507 TAO_DTP_Task::set_thread_idle_time(ACE_Time_Value thr_timeout)
 508 {
 509   this->thread_idle_time_ = thr_timeout;
 510 }
 511
 512 void
 513 TAO_DTP_Task::set_max_request_queue_depth (size_t queue_depth)
 514 {
 515   this->max_request_queue_depth_ = queue_depth;
 516 }
 517
 518 void
 519 TAO_DTP_Task::cancel_servant (PortableServer::Servant servant)
 520 {
 521   if (servant == 0)
 522     {
 523       if (TAO_debug_level > 0)
 524         {
 525           TAOLIB_DEBUG ((LM_DEBUG,
 526                          ACE_TEXT ("TAO (%P|%t) - DTP_Task::cancel_servant ")
 527                          ACE_TEXT ("called with null servant\n")
 528                          ));
 529         }
 530       return;
 531     }
 532
 533   ACE_GUARD (TAO_SYNCH_MUTEX, guard, this->queue_lock_);
 534
 535   // Cancel the requests targeted for the provided servant.
 536   TAO::CSD::TP_Cancel_Visitor cancel_visitor (servant);
 537   this->queue_.accept_visitor (cancel_visitor);
 538 }
 539
 540 TAO_END_VERSIONED_NAMESPACE_DECL
 541
 542 #endif /* (TAO_HAS_CORBA_MESSAGING) && TAO_HAS_CORBA_MESSAGING != 0 */