TAO/orbsvcs/ImplRepo_Service/LiveCheck.cpp

   1 // -*- C++ -*-
   2 #include "LiveCheck.h"
   3 #include "ImR_Locator_i.h"
   4
   5 #include "orbsvcs/Log_Macros.h"
   6
   7 #include "tao/ORB_Core.h"
   8 #include "ace/Reactor.h"
   9 #include "ace/OS_NS_sys_time.h"
  10 #include "ace/Timer_Queue.h"
  11 #include "ace/Timer_Queue_Iterator.h"
  12
  13 LiveListener::LiveListener (const char *server)
  14   : server_ (server),
  15     refcount_ (1)
  16 {
  17 }
  18
  19 LiveListener::~LiveListener (void)
  20 {
  21 }
  22
  23 const char *
  24 LiveListener::server (void) const
  25 {
  26   return this->server_.c_str ();
  27 }
  28
  29 LiveListener *
  30 LiveListener::_add_ref (void)
  31 {
  32   int const refcount = ++this->refcount_;
  33   if (ImR_Locator_i::debug () > 5)
  34     {
  35       ORBSVCS_DEBUG ((LM_DEBUG,
  36                       ACE_TEXT ("(%P|%t) LiveListener::add_ref <%C> count <%d>\n"),
  37                       server_.c_str(), refcount));
  38     }
  39   return this;
  40 }
  41
  42 void
  43 LiveListener::_remove_ref (void)
  44 {
  45   int const count = --this->refcount_;
  46   if (ImR_Locator_i::debug () > 5)
  47     {
  48       ORBSVCS_DEBUG ((LM_DEBUG,
  49                       ACE_TEXT  ("(%P|%t) LiveListener::remove_ref <%C> count <%d>\n"),
  50                       server_.c_str(), count));
  51     }
  52   if (count == 0)
  53     {
  54       delete this;
  55     }
  56 }
  57
  58 //---------------------------------------------------------------------------
  59 //---------------------------------------------------------------------------
  60
  61 const int LiveEntry::reping_msec_[] = {10, 100, 500, 1000, 1000, 2000, 2000, 5000, 5000};
  62 int LiveEntry::reping_limit_ = sizeof (LiveEntry::reping_msec_) / sizeof (int);
  63
  64 const char *
  65 LiveEntry::status_name (LiveStatus s)
  66 {
  67   switch (s)
  68     {
  69     case LS_INIT:
  70       return "INIT";
  71     case LS_UNKNOWN:
  72       return "UNKNOWN";
  73     case LS_PING_AWAY:
  74       return "PING_AWAY";
  75     case LS_DEAD:
  76       return "DEAD";
  77     case LS_ALIVE:
  78       return "ALIVE";
  79     case LS_TRANSIENT:
  80       return "TRANSIENT";
  81     case LS_LAST_TRANSIENT:
  82       return "LAST_TRANSIENT";
  83     case LS_TIMEDOUT:
  84       return "TIMEDOUT";
  85     case LS_CANCELED:
  86       return "CANCELED";
  87     }
  88   return "<undefined status>";
  89 }
  90
  91 void
  92 LiveEntry::set_reping_limit (int max)
  93 {
  94   int array_max =  sizeof (LiveEntry::reping_msec_) / sizeof (int);
  95   LiveEntry::reping_limit_ = max < array_max && max >= 0 ? max : array_max;
  96 }
  97
  98 bool
  99 LiveEntry::reping_available (void) const
 100 {
 101   return this->repings_ < this->max_retry_;
 102 }
 103
 104 int
 105 LiveEntry::next_reping (void)
 106 {
 107   ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, mon, this->lock_, -1);
 108   return this->reping_available() ? LiveEntry::reping_msec_[this->repings_++] : -1;
 109 }
 110
 111 void
 112 LiveEntry::max_retry_msec (int msec)
 113 {
 114   ACE_GUARD (TAO_SYNCH_MUTEX, mon, this->lock_);
 115   for (this->max_retry_ = 0;
 116        this->max_retry_ <  LiveEntry::reping_limit_ && msec > 0;
 117        ++this->max_retry_)
 118     {
 119       msec -= LiveEntry::reping_msec_[this->repings_];
 120     }
 121 }
 122
 123 LiveEntry::LiveEntry (LiveCheck *owner,
 124                       const char *server,
 125                       bool may_ping,
 126                       ImplementationRepository::ServerObject_ptr ref,
 127                       int pid)
 128   : owner_ (owner),
 129     server_ (server),
 130     ref_ (ImplementationRepository::ServerObject::_duplicate (ref)),
 131     liveliness_ (LS_INIT),
 132     next_check_ (ACE_OS::gettimeofday()),
 133     repings_ (0),
 134     max_retry_ (LiveEntry::reping_limit_),
 135     may_ping_ (may_ping),
 136     listeners_ (),
 137     lock_ (),
 138     callback_ (0),
 139     pid_ (pid)
 140 {
 141   if (ImR_Locator_i::debug () > 4)
 142     {
 143       ORBSVCS_DEBUG ((LM_DEBUG,
 144                       ACE_TEXT ("(%P|%t) LiveEntry::ctor server <%C> status <%C> may_ping <%d> pid <%d>\n"),
 145                       server, status_name (this->liveliness_), may_ping, pid));
 146     }
 147 }
 148
 149 LiveEntry::~LiveEntry (void)
 150 {
 151   if (this->callback_.in () != 0)
 152     {
 153       PingReceiver *rec = dynamic_cast<PingReceiver *>(this->callback_.in());
 154       if (rec != 0)
 155         {
 156           rec->cancel ();
 157         }
 158     }
 159 }
 160
 161 void
 162 LiveEntry::release_callback (void)
 163 {
 164   this->callback_ = 0;
 165 }
 166
 167 void
 168 LiveEntry::add_listener (LiveListener *ll)
 169 {
 170   ACE_GUARD (TAO_SYNCH_MUTEX, mon, this->lock_);
 171   LiveListener_ptr llp(ll->_add_ref());
 172   int const result = this->listeners_.insert (llp);
 173   if (ImR_Locator_i::debug() > 4)
 174     {
 175       ORBSVCS_DEBUG ((LM_DEBUG,
 176                       ACE_TEXT ("(%P|%t) LiveEntry::add_listener server <%C> result <%d>\n"),
 177                       this->server_.c_str(),
 178                       result));
 179     }
 180 }
 181
 182 void
 183 LiveEntry::remove_listener (LiveListener *ll)
 184 {
 185   ACE_GUARD (TAO_SYNCH_MUTEX, mon, this->lock_);
 186   LiveListener_ptr llp(ll->_add_ref());
 187   int const result = this->listeners_.remove (llp);
 188   if (ImR_Locator_i::debug() > 4)
 189     {
 190       ORBSVCS_DEBUG ((LM_DEBUG,
 191                       ACE_TEXT ("(%P|%t) LiveEntry::remove_listener server <%C> result <%d>\n"),
 192                       this->server_.c_str(),
 193                       result));
 194     }
 195 }
 196
 197 void
 198 LiveEntry::reset_status (void)
 199 {
 200   ACE_GUARD (TAO_SYNCH_MUTEX, mon, this->lock_);
 201   if ( this->liveliness_ == LS_ALIVE ||
 202        this->liveliness_ == LS_LAST_TRANSIENT ||
 203        this->liveliness_ == LS_TIMEDOUT)
 204     {
 205       this->liveliness_ = LS_UNKNOWN;
 206       this->repings_ = 0;
 207       this->next_check_ = ACE_OS::gettimeofday();
 208     }
 209   if (ImR_Locator_i::debug () > 2)
 210     {
 211       ORBSVCS_DEBUG ((LM_DEBUG,
 212                       ACE_TEXT ("(%P|%t) LiveEntry::reset_status this <%x> ")
 213                       ACE_TEXT ("server <%C> status <%C>\n"),
 214                       this, this->server_.c_str(),
 215                       status_name (this->liveliness_)));
 216     }
 217
 218 }
 219
 220 LiveStatus
 221 LiveEntry::status (void) const
 222 {
 223   if (!this->may_ping_)
 224     {
 225       return LS_ALIVE;
 226     }
 227
 228   if (this->liveliness_ == LS_ALIVE &&
 229       this->owner_->ping_interval() != ACE_Time_Value::zero)
 230     {
 231       ACE_Time_Value now (ACE_OS::gettimeofday());
 232       if (now >= this->next_check_)
 233         {
 234           return LS_UNKNOWN;
 235         }
 236     }
 237   return this->liveliness_;
 238 }
 239
 240 void
 241 LiveEntry::update_listeners (void)
 242 {
 243   Listen_Set remove;
 244   for (Listen_Set::ITERATOR i(this->listeners_);
 245        !i.done();
 246        i.advance ())
 247     {
 248
 249       if ((*i)->status_changed (this->liveliness_))
 250         {
 251           remove.insert (*i);
 252         }
 253     }
 254   {
 255     ACE_GUARD (TAO_SYNCH_MUTEX, mon, this->lock_);
 256     for (Listen_Set::ITERATOR i (remove);
 257          !i.done();
 258          i.advance ())
 259       {
 260         LiveListener_ptr llp (*i);
 261         int const result = this->listeners_.remove (llp);
 262         if (result == -1)
 263           {
 264           }
 265       }
 266     LiveListener_ptr dummy;
 267     this->listeners_.remove (dummy);
 268   }
 269 }
 270
 271 void
 272 LiveEntry::status (LiveStatus l)
 273 {
 274   {
 275     ACE_GUARD (TAO_SYNCH_MUTEX, mon, this->lock_);
 276     this->liveliness_ = l;
 277     if (l == LS_ALIVE)
 278       {
 279         ACE_Time_Value now (ACE_OS::gettimeofday());
 280         this->next_check_ = now + owner_->ping_interval();
 281       }
 282     if (l == LS_TRANSIENT && !this->reping_available())
 283       {
 284         this->liveliness_ = LS_LAST_TRANSIENT;
 285       }
 286   }
 287   this->update_listeners ();
 288
 289   if (!this->listeners_.is_empty ())
 290     {
 291       if (ImR_Locator_i::debug () > 2)
 292         {
 293           ORBSVCS_DEBUG ((LM_DEBUG,
 294                           ACE_TEXT ("(%P|%t) LiveEntry::status change, ")
 295                           ACE_TEXT ("server <%C> status <%C>\n"),
 296                           this->server_.c_str(),
 297                           status_name (this->liveliness_)));
 298         }
 299       this->owner_->schedule_ping (this);
 300     }
 301   else
 302     {
 303       if (this->owner_->remove_per_client_entry (this))
 304         {
 305           delete (this);
 306         }
 307     }
 308 }
 309
 310 const ACE_Time_Value &
 311 LiveEntry::next_check (void) const
 312 {
 313   return this->next_check_;
 314 }
 315
 316 const char *
 317 LiveEntry::server_name (void) const
 318 {
 319   return this->server_.c_str();
 320 }
 321
 322 void
 323 LiveEntry::set_pid (int pid)
 324 {
 325   this->pid_ = pid;
 326 }
 327
 328 int
 329 LiveEntry::pid (void) const
 330 {
 331   return this->pid_;
 332 }
 333
 334 bool
 335 LiveEntry::may_ping (void) const
 336 {
 337   return this->may_ping_;
 338 }
 339
 340 bool
 341 LiveEntry::has_pid (int pid) const
 342 {
 343   return this->pid_ == 0 || pid == 0 || pid == this->pid_;
 344 }
 345
 346 bool
 347 LiveEntry::validate_ping (bool &want_reping, ACE_Time_Value& next)
 348 {
 349   if (ImR_Locator_i::debug () > 4)
 350     {
 351       ORBSVCS_DEBUG ((LM_DEBUG,
 352                       ACE_TEXT ("(%P|%t) LiveEntry::validate_ping, status ")
 353                       ACE_TEXT ("<%C> listeners <%d> server <%C> pid <%d> want_reping <%d> may_ping <%d>\n"),
 354                       status_name (this->liveliness_), this->listeners_.size (),
 355                       this->server_.c_str(), this->pid_, want_reping, this->may_ping_));
 356     }
 357
 358   if (this->liveliness_ == LS_PING_AWAY ||
 359       this->liveliness_ == LS_DEAD ||
 360       this->listeners_.is_empty ())
 361     {
 362       return false;
 363     }
 364   ACE_Time_Value const now (ACE_OS::gettimeofday());
 365   ACE_Time_Value const diff = this->next_check_ - now;
 366   long const msec = diff.msec();
 367   if (msec > 0)
 368     {
 369       if (!want_reping || this->next_check_ < next)
 370         {
 371           want_reping = true;
 372           next = this->next_check_;
 373         }
 374       if (ImR_Locator_i::debug () > 2)
 375         {
 376           ORBSVCS_DEBUG ((LM_DEBUG,
 377                           ACE_TEXT ("(%P|%t) LiveEntry::validate_ping, ")
 378                           ACE_TEXT ("status <%C> listeners <%d> ")
 379                           ACE_TEXT ("msec <%d> server <%C> pid <%d>\n"),
 380                           status_name (this->liveliness_), this->listeners_.size (),
 381                           msec, this->server_.c_str(), this->pid_));
 382         }
 383       return false;
 384     }
 385   switch (this->liveliness_)
 386     {
 387     case LS_UNKNOWN:
 388     case LS_INIT:
 389       break;
 390     case LS_ALIVE:
 391     case LS_TIMEDOUT:
 392       {
 393         ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, mon, this->lock_, false);
 394         this->next_check_ = now + owner_->ping_interval();
 395       }
 396       break;
 397     case LS_TRANSIENT:
 398     case LS_LAST_TRANSIENT:
 399       {
 400         int const ms = this->next_reping ();
 401         if (ms != -1)
 402           {
 403             ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, mon, this->lock_, false);
 404             if (this->liveliness_ == LS_LAST_TRANSIENT)
 405               {
 406                 this->liveliness_ = LS_TRANSIENT;
 407               }
 408             ACE_Time_Value const next (ms / 1000, (ms % 1000) * 1000);
 409             this->next_check_ = now + next;
 410             if (ImR_Locator_i::debug () > 4)
 411               {
 412                 ORBSVCS_DEBUG ((LM_DEBUG,
 413                                 ACE_TEXT ("(%P|%t) LiveEntry::validate_ping, ")
 414                                 ACE_TEXT ("transient, reping in <%d> ms, ")
 415                                 ACE_TEXT ("server <%C> pid <%d>\n"),
 416                                 ms, this->server_.c_str(), this->pid_));
 417               }
 418           }
 419         else
 420           {
 421             if (this->liveliness_ == LS_TRANSIENT)
 422               {
 423                 ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, mon, this->lock_, false);
 424                 this->liveliness_ = LS_LAST_TRANSIENT;
 425               }
 426             if (ImR_Locator_i::debug () > 2)
 427               {
 428                 ORBSVCS_DEBUG ((LM_DEBUG,
 429                                 ACE_TEXT ("(%P|%t) LiveEntry::validate_ping, ")
 430                                 ACE_TEXT ("transient, no more repings, ")
 431                                 ACE_TEXT ("server <%C> pid <%d>\n"),
 432                                 this->server_.c_str(), this->pid_));
 433               }
 434             if (!this->listeners_.is_empty ())
 435               {
 436                 this->update_listeners ();
 437               }
 438             return false;
 439           }
 440       }
 441       break;
 442     default:;
 443     }
 444   return true;
 445 }
 446
 447 void
 448 LiveEntry::do_ping (PortableServer::POA_ptr poa)
 449 {
 450   this->callback_ = new PingReceiver (this, poa);
 451   PortableServer::ObjectId_var oid = poa->activate_object (this->callback_.in());
 452   CORBA::Object_var obj = poa->id_to_reference (oid.in());
 453   ImplementationRepository::AMI_ServerObjectHandler_var cb =
 454     ImplementationRepository::AMI_ServerObjectHandler::_narrow (obj.in());
 455   {
 456     ACE_GUARD (TAO_SYNCH_MUTEX, mon, this->lock_);
 457     this->liveliness_ = LS_PING_AWAY;
 458   }
 459   try
 460     {
 461       if (ImR_Locator_i::debug () > 3)
 462         {
 463           ORBSVCS_DEBUG ((LM_DEBUG,
 464                           ACE_TEXT ("(%P|%t) LiveEntry::do_ping, ")
 465                           ACE_TEXT ("starting sendc_ping for server <%C>\n"),
 466                           this->server_.c_str()));
 467         }
 468       this->ref_->sendc_ping (cb.in());
 469       if (ImR_Locator_i::debug () > 3)
 470         {
 471           ORBSVCS_DEBUG ((LM_DEBUG,
 472                           ACE_TEXT ("(%P|%t) LiveEntry::do_ping, ")
 473                           ACE_TEXT ("sendc_ping for server <%C> returned OK\n"),
 474                           this->server_.c_str()));
 475         }
 476     }
 477   catch (const CORBA::Exception &ex)
 478     {
 479       if (ImR_Locator_i::debug () > 3)
 480         {
 481           ORBSVCS_DEBUG ((LM_DEBUG,
 482                           ACE_TEXT ("(%P|%t) LiveEntry::do_ping, ")
 483                           ACE_TEXT ("sendc_ping for server <%C> threw <%C> marking as dead\n"),
 484                           this->server_.c_str(), ex._info ().c_str ()));
 485         }
 486       this->release_callback ();
 487       this->status (LS_DEAD);
 488     }
 489 }
 490
 491 //---------------------------------------------------------------------------
 492 PingReceiver::PingReceiver (LiveEntry *entry, PortableServer::POA_ptr poa)
 493   :poa_ (PortableServer::POA::_duplicate(poa)),
 494    entry_ (entry)
 495 {
 496 }
 497
 498 PingReceiver::~PingReceiver (void)
 499 {
 500 }
 501
 502 void
 503 PingReceiver::cancel (void)
 504 {
 505   if (ImR_Locator_i::debug () > 4)
 506     {
 507       const char *server = "not available";
 508       if (this->entry_ != 0)
 509         {
 510           server = this->entry_->server_name ();
 511         }
 512       ORBSVCS_DEBUG ((LM_DEBUG,
 513                       ACE_TEXT ("(%P|%t) PingReceiver::cancel server <%C>\n"),
 514                       server));
 515     }
 516
 517   this->entry_ = 0;
 518   try
 519     {
 520       PortableServer::ObjectId_var oid = this->poa_->servant_to_id (this);
 521       poa_->deactivate_object (oid.in());
 522     }
 523   catch (const CORBA::Exception &ex)
 524     {
 525       if (ImR_Locator_i::debug () > 4)
 526         {
 527           ORBSVCS_DEBUG ((LM_DEBUG,
 528                           ACE_TEXT ("(%P|%t) PingReceiver::cancel caught <%C>\n"),
 529                           ex._info ().c_str ()));
 530         }
 531     }
 532 }
 533
 534 void
 535 PingReceiver::ping (void)
 536 {
 537   if (this->entry_ != 0)
 538     {
 539       if (ImR_Locator_i::debug () > 5)
 540         {
 541           ORBSVCS_DEBUG ((LM_DEBUG,
 542                           ACE_TEXT ("(%P|%t) PingReceiver::ping received from <%C>\n"),
 543                           this->entry_->server_name ()));
 544         }
 545       this->entry_->release_callback ();
 546       this->entry_->status (LS_ALIVE);
 547     }
 548   PortableServer::ObjectId_var oid = this->poa_->servant_to_id (this);
 549   poa_->deactivate_object (oid.in());
 550 }
 551
 552 void
 553 PingReceiver::ping_excep (Messaging::ExceptionHolder * excep_holder)
 554 {
 555   const CORBA::ULong TAO_MINOR_MASK = 0x00000f80;
 556   try
 557     {
 558       if (ImR_Locator_i::debug () > 5)
 559         {
 560           ORBSVCS_DEBUG ((LM_DEBUG,
 561                           ACE_TEXT ("(%P|%t) PingReceiver::ping_excep received from <%C>\n"),
 562                           this->entry_->server_name ()));
 563         }
 564       excep_holder->raise_exception ();
 565     }
 566   catch (const CORBA::TRANSIENT &ex)
 567     {
 568       switch (ex.minor () & TAO_MINOR_MASK)
 569         {
 570         case TAO_POA_DISCARDING:
 571         case TAO_POA_HOLDING:
 572           {
 573             if (this->entry_ != 0)
 574               {
 575                 this->entry_->release_callback ();
 576                 this->entry_->status (LS_TRANSIENT);
 577               }
 578             break;
 579           }
 580         default: //case TAO_INVOCATION_SEND_REQUEST_MINOR_CODE:
 581           {
 582             if (this->entry_ != 0)
 583               {
 584                 this->entry_->release_callback ();
 585                 this->entry_->status (LS_DEAD);
 586               }
 587           }
 588         }
 589     }
 590   catch (const CORBA::TIMEOUT &ex)
 591     {
 592       if (this->entry_ != 0)
 593         {
 594           this->entry_->release_callback ();
 595           if ((ex.minor () & TAO_MINOR_MASK) == TAO_TIMEOUT_CONNECT_MINOR_CODE)
 596             {
 597               this->entry_->status (LS_DEAD);
 598             }
 599           else
 600             {
 601               this->entry_->status (LS_TIMEDOUT);
 602             }
 603         }
 604     }
 605   catch (const CORBA::Exception &)
 606     {
 607       if (this->entry_ != 0)
 608         {
 609           this->entry_->release_callback ();
 610           this->entry_->status (LS_DEAD);
 611         }
 612     }
 613
 614   PortableServer::ObjectId_var oid = this->poa_->servant_to_id (this);
 615   poa_->deactivate_object (oid.in());
 616 }
 617
 618 //---------------------------------------------------------------------------
 619 //---------------------------------------------------------------------------
 620
 621 LC_TimeoutGuard::LC_TimeoutGuard (LiveCheck *owner, LC_token_type token)
 622   :owner_ (owner),
 623    token_ (token),
 624    blocked_ (owner->in_handle_timeout ())
 625 {
 626   if (ImR_Locator_i::debug () > 3)
 627     {
 628       ORBSVCS_DEBUG ((LM_DEBUG,
 629                       ACE_TEXT ("(%P|%t) LC_TimeoutGuard(%d)::ctor, ")
 630                       ACE_TEXT ("blocked <%d>\n"),
 631                       this->token_, this->blocked_));
 632     }
 633   owner_->enter_handle_timeout ();
 634 }
 635
 636 LC_TimeoutGuard::~LC_TimeoutGuard (void)
 637 {
 638   owner_->exit_handle_timeout ();
 639
 640   if (blocked_)
 641   {
 642     if (ImR_Locator_i::debug () > 3)
 643       {
 644         ORBSVCS_DEBUG ((LM_DEBUG,
 645                         ACE_TEXT ("(%P|%t) LC_TimeoutGuard(%d)::dtor, ")
 646                         ACE_TEXT ("doing nothing because our owner is blocked\n"),
 647                         this->token_));
 648       }
 649     return;
 650   }
 651
 652   owner_->remove_deferred_servers ();
 653
 654   if (owner_->want_timeout_)
 655     {
 656       ACE_Time_Value delay = ACE_Time_Value::zero;
 657       if (owner_->deferred_timeout_ != ACE_Time_Value::zero)
 658         {
 659           ACE_Time_Value const now (ACE_OS::gettimeofday());
 660           if (owner_->deferred_timeout_ > now)
 661             delay = owner_->deferred_timeout_ - now;
 662         }
 663       ++owner_->token_;
 664       if (ImR_Locator_i::debug () > 2)
 665         {
 666           ORBSVCS_DEBUG ((LM_DEBUG,
 667                           ACE_TEXT ("(%P|%t) LC_TimeoutGuard(%d)::dtor, ")
 668                           ACE_TEXT ("scheduling new timeout(%d), delay = %d,%d\n"),
 669                           this->token_, owner_->token_, delay.sec(), delay.usec()));
 670         }
 671       owner_->reactor()->schedule_timer (owner_,
 672                                          reinterpret_cast<void *>(owner_->token_),
 673                                          delay);
 674       owner_->want_timeout_ = false;
 675     }
 676   else
 677     {
 678       if (ImR_Locator_i::debug () > 3)
 679         {
 680           ORBSVCS_DEBUG ((LM_DEBUG,
 681                           ACE_TEXT ("(%P|%t) LC_TimeoutGuard(%d)::dtor, ")
 682                           ACE_TEXT ("no pending timeouts requested\n"),
 683                           this->token_));
 684         }
 685     }
 686 }
 687
 688 bool LC_TimeoutGuard::blocked (void) const
 689 {
 690   return this->blocked_;
 691 }
 692
 693 //---------------------------------------------------------------------------
 694 //---------------------------------------------------------------------------
 695
 696 LiveCheck::LiveCheck ()
 697   :ping_interval_(),
 698    running_ (false),
 699    token_ (100),
 700    handle_timeout_busy_ (0),
 701    want_timeout_ (false),
 702    deferred_timeout_ (ACE_Time_Value::zero)
 703 {
 704 }
 705
 706 LiveCheck::~LiveCheck (void)
 707 {
 708   for (LiveEntryMap::iterator em (this->entry_map_); !em.done(); em++)
 709     {
 710       delete em->int_id_;
 711     }
 712   this->entry_map_.unbind_all();
 713
 714   for (PerClientStack::iterator pc (this->per_client_); !pc.done(); pc++)
 715     {
 716       delete *pc;
 717     }
 718   this->per_client_.reset ();
 719   this->removed_entries_.reset ();
 720 }
 721
 722 void
 723 LiveCheck::enter_handle_timeout (void)
 724 {
 725   ++this->handle_timeout_busy_;
 726 }
 727
 728 void
 729 LiveCheck::exit_handle_timeout (void)
 730 {
 731   --this->handle_timeout_busy_;
 732 }
 733
 734 bool
 735 LiveCheck::in_handle_timeout (void)
 736 {
 737   return this->handle_timeout_busy_ != 0;
 738 }
 739
 740 void
 741 LiveCheck::init (CORBA::ORB_ptr orb,
 742                  const ACE_Time_Value &pi)
 743 {
 744   this->ping_interval_ = pi;
 745   ACE_Reactor *r = orb->orb_core()->reactor();
 746   this->reactor (r);
 747   CORBA::Object_var obj = orb->resolve_initial_references ("RootPOA");
 748   this->poa_ = PortableServer::POA::_narrow (obj.in());
 749   this->running_ = true;
 750 }
 751
 752 void
 753 LiveCheck::shutdown (void)
 754 {
 755   this->running_ = false;
 756   this->reactor()->cancel_timer (this);
 757 }
 758
 759 const ACE_Time_Value &
 760 LiveCheck::ping_interval (void) const
 761 {
 762   return this->ping_interval_;
 763 }
 764
 765 int
 766 LiveCheck::handle_timeout (const ACE_Time_Value &,
 767                            const void * tok)
 768 {
 769   LC_token_type token = reinterpret_cast<LC_token_type>(tok);
 770   if (ImR_Locator_i::debug () > 2)
 771     {
 772       ORBSVCS_DEBUG ((LM_DEBUG,
 773                       ACE_TEXT ("(%P|%t) LiveCheck::handle_timeout(%d), ")
 774                       ACE_TEXT ("running <%d>\n"),
 775                       token, this->running_));
 776     }
 777   if (!this->running_)
 778     return -1;
 779
 780   LC_TimeoutGuard tg (this, token);
 781   if (tg.blocked ())
 782     return 0;
 783
 784   LiveEntryMap::iterator le_end = this->entry_map_.end();
 785   for (LiveEntryMap::iterator le = this->entry_map_.begin();
 786        le != le_end;
 787        ++le)
 788     {
 789       LiveEntry *entry = le->item ();
 790       if (entry->validate_ping (this->want_timeout_, this->deferred_timeout_))
 791         {
 792           entry->do_ping (poa_.in ());
 793           if (ImR_Locator_i::debug () > 2)
 794             {
 795               ORBSVCS_DEBUG ((LM_DEBUG,
 796                               ACE_TEXT ("(%P|%t) LiveCheck::handle_timeout(%d)")
 797                               ACE_TEXT (", ping sent to server <%C>\n"),
 798                               token, entry->server_name ()));
 799             }
 800         }
 801       else
 802         {
 803           if (ImR_Locator_i::debug () > 4)
 804             {
 805               ORBSVCS_DEBUG ((LM_DEBUG,
 806                               ACE_TEXT ("(%P|%t) LiveCheck::handle_timeout(%d)")
 807                               ACE_TEXT (", ping skipped for server <%C> may_ping <%d>\n"),
 808                               token, entry->server_name (), entry->may_ping ()));
 809             }
 810         }
 811     }
 812
 813   PerClientStack::iterator pe_end = this->per_client_.end();
 814   for (PerClientStack::iterator pe = this->per_client_.begin();
 815        pe != pe_end;
 816        ++pe)
 817     {
 818       LiveEntry *entry = *pe;
 819       if (entry != 0)
 820         {
 821           if (entry->validate_ping (this->want_timeout_, this->deferred_timeout_))
 822             {
 823               entry->do_ping (poa_.in ());
 824             }
 825           LiveStatus const status = entry->status ();
 826           if (status != LS_PING_AWAY && status != LS_TRANSIENT)
 827             {
 828               this->per_client_.remove (entry);
 829               delete entry;
 830             }
 831         }
 832     }
 833
 834   return 0;
 835 }
 836
 837 bool
 838 LiveCheck::has_server (const char *server)
 839 {
 840   ACE_CString s (server);
 841   LiveEntry *entry = 0;
 842   int const result = entry_map_.find (s, entry);
 843   return (result == 0 && entry != 0);
 844 }
 845
 846 void
 847 LiveCheck::add_server (const char *server,
 848                        bool may_ping,
 849                        ImplementationRepository::ServerObject_ptr ref,
 850                        int pid)
 851 {
 852   if (ImR_Locator_i::debug () > 2)
 853     {
 854       ORBSVCS_DEBUG ((LM_DEBUG,
 855                       ACE_TEXT ("(%P|%t) LiveCheck::add_server <%C> ")
 856                       ACE_TEXT ("may_ping <%d> running <%d> pid <%d>\n"),
 857                       server, may_ping, this->running_, pid));
 858     }
 859
 860   if (!this->running_)
 861     return;
 862
 863   ACE_CString s (server);
 864   LiveEntry *entry = 0;
 865   ACE_NEW (entry, LiveEntry (this, server, may_ping, ref, pid));
 866   int result = entry_map_.bind (s, entry);
 867   if (result != 0)
 868     {
 869       LiveEntry *old = 0;
 870       result = entry_map_.rebind (s, entry, old);
 871       if (old)
 872         {
 873           old->status (LS_CANCELED);
 874         }
 875       delete old;
 876     }
 877 }
 878
 879 void
 880 LiveCheck::set_pid (const char *server, int pid)
 881 {
 882   if (ImR_Locator_i::debug () > 0)
 883     {
 884       ORBSVCS_DEBUG ((LM_DEBUG,
 885                       ACE_TEXT ("(%P|%t) LiveCheck::set_pid <%C> pid <%d>\n"),
 886                       server, pid));
 887     }
 888   ACE_CString s(server);
 889   LiveEntry *entry = 0;
 890   int const result = entry_map_.find (s, entry);
 891   if (result != -1 && entry != 0)
 892     {
 893       entry->set_pid (pid);
 894     }
 895   else
 896     {
 897       if (ImR_Locator_i::debug () > 0)
 898         {
 899           ORBSVCS_DEBUG ((LM_DEBUG,
 900                           ACE_TEXT ("(%P|%t) LiveCheck::set_pid <%C> pid <%d> cannot find entry\n"),
 901                           server, pid));
 902         }
 903     }
 904 }
 905
 906 void
 907 LiveCheck::remove_server (const char *server, int pid)
 908 {
 909   if (ImR_Locator_i::debug () > 0)
 910     {
 911       ORBSVCS_DEBUG ((LM_DEBUG,
 912                       ACE_TEXT ("(%P|%t) LiveCheck::remove_server <%C> pid <%d>\n"),
 913                       server, pid));
 914     }
 915   ACE_CString s(server);
 916   LiveEntry *entry = 0;
 917   int const result = entry_map_.find (s, entry);
 918   if (result != -1 && entry != 0 && entry->has_pid (pid))
 919     {
 920       if (!this->in_handle_timeout ())
 921         {
 922           if (ImR_Locator_i::debug () > 0)
 923             {
 924               ORBSVCS_DEBUG ((LM_DEBUG,
 925                               ACE_TEXT ("(%P|%t) LiveCheck::remove_server removing <%C> pid <%d> entry pid <%d> status <%C>\n"),
 926                               server, pid, entry->pid (), LiveEntry::status_name (entry->status ())));
 927             }
 928           if (entry_map_.unbind (s, entry) == 0)
 929             {
 930               delete entry;
 931             }
 932         }
 933       else
 934         {
 935           // We got a request to remove the server but we are in handle timeout, so we have to postpone
 936           // the remove. We do set the status to dead so that we make sure that we only remove later
 937           // on the dead server and not a possible restart
 938           entry->status (LS_DEAD);
 939
 940           if (ImR_Locator_i::debug () > 0)
 941             {
 942               ORBSVCS_DEBUG ((LM_DEBUG,
 943                               ACE_TEXT ("(%P|%t) LiveCheck::remove_server <%C> pid <%d> entry pid <%d> status <%C> ")
 944                               ACE_TEXT ("called during handle_timeout\n"), server, pid, entry->pid (), LiveEntry::status_name (entry->status ())));
 945             }
 946           this->removed_entries_.insert_tail (std::make_pair (s, pid));
 947         }
 948     }
 949   else
 950     {
 951       if (ImR_Locator_i::debug () > 0)
 952         {
 953           if (entry == 0)
 954             {
 955               ORBSVCS_DEBUG ((LM_DEBUG,
 956                               ACE_TEXT ("(%P|%t) LiveCheck::remove_server <%C> ")
 957                               ACE_TEXT ("Can't find server entry, server probably already removed earlier\n"),
 958                               server));
 959             }
 960           else
 961             {
 962               ORBSVCS_DEBUG ((LM_DEBUG,
 963                               ACE_TEXT ("(%P|%t) LiveCheck::remove_server <%C> ")
 964                               ACE_TEXT ("pid <%d> does not match entry pid <%d>\n"),
 965                               server, pid, entry->pid ()));
 966             }
 967         }
 968     }
 969 }
 970
 971 void
 972 LiveCheck::remove_deferred_servers (void)
 973 {
 974   if (!this->removed_entries_.is_empty ())
 975     {
 976       // When we are in handle_timeout we can't remove deferred servers
 977       if (!this->in_handle_timeout ())
 978         {
 979           NamePidStack::iterator re_end = this->removed_entries_.end();
 980           for (NamePidStack::iterator re = this->removed_entries_.begin();
 981               re != re_end;
 982               ++re)
 983             {
 984               NamePidPair const & name_pid_pair = (*re);
 985               if (ImR_Locator_i::debug () > 4)
 986                 {
 987                   ORBSVCS_DEBUG ((LM_DEBUG,
 988                                   ACE_TEXT ("(%P|%t) LiveCheck::remove_deferred_servers ")
 989                                   ACE_TEXT ("removing <%C> pid <%d>\n"),
 990                                   name_pid_pair.first.c_str(), name_pid_pair.second));
 991                 }
 992               // Now try to remove the server, we have to make sure
 993               // that we only remove the server when the
 994               // name and pid match. These could potentially not
 995               // match when the server has already been restarted between the
 996               // moment it got in the removed_entries_ stack and this point
 997               // where we remove it from the internal administration
 998               LiveEntry *entry = 0;
 999               int const result = entry_map_.find (name_pid_pair.first, entry);
1000               if (result != -1 && entry != 0)
1001                 {
1002                   if (entry->pid () == name_pid_pair.second)
1003                     {
1004                       if (entry->status () == LS_DEAD)
1005                         {
1006                           // We have a matched process id
1007                           if (ImR_Locator_i::debug () > 4)
1008                             {
1009                               ORBSVCS_DEBUG ((LM_DEBUG,
1010                                               ACE_TEXT ("(%P|%t) LiveCheck::remove_deferred_servers <%C> ")
1011                                               ACE_TEXT ("removing dead server using matched pid <%d>\n"),
1012                                               name_pid_pair.first.c_str(), name_pid_pair.second));
1013                             }
1014                           if (entry_map_.unbind (name_pid_pair.first, entry) == 0)
1015                             {
1016                               delete entry;
1017                             }
1018                         }
1019                       else
1020                         {
1021                           ORBSVCS_DEBUG ((LM_DEBUG,
1022                                           ACE_TEXT ("(%P|%t) LiveCheck::remove_deferred_servers <%C> ")
1023                                           ACE_TEXT ("matched pid <%d> but is not dead but <%C>\n"),
1024                                           name_pid_pair.first.c_str(), name_pid_pair.second, LiveEntry::status_name (entry->status ())));
1025                         }
1026                     }
1027                   else
1028                     {
1029                       ORBSVCS_DEBUG ((LM_DEBUG,
1030                                       ACE_TEXT ("(%P|%t) LiveCheck::remove_deferred_servers <%C> ")
1031                                       ACE_TEXT ("pid <%d> does not match entry pid <%d>\n"),
1032                                       name_pid_pair.first.c_str(), name_pid_pair.second, entry->pid ()));
1033                     }
1034                 }
1035               else
1036                 {
1037                   if (ImR_Locator_i::debug () > 0)
1038                     {
1039                       ORBSVCS_DEBUG ((LM_DEBUG,
1040                                       ACE_TEXT ("(%P|%t) LiveCheck::remove_deferred_servers <%C> ")
1041                                       ACE_TEXT ("Can't find server entry, server probably already removed earlier\n"),
1042                                       name_pid_pair.first.c_str()));
1043                     }
1044                 }
1045             }
1046           this->removed_entries_.reset ();
1047         }
1048       else
1049         {
1050           if (ImR_Locator_i::debug () > 0)
1051             {
1052               ORBSVCS_DEBUG ((LM_DEBUG,
1053                               ACE_TEXT ("(%P|%t) LiveCheck::remove_deferred_servers ")
1054                               ACE_TEXT ("Can't remove <%d> servers because we are still in handle timeout\n"),
1055                               this->removed_entries_.size ()));
1056             }
1057         }
1058     }
1059 }
1060
1061 bool
1062 LiveCheck::remove_per_client_entry (LiveEntry *e)
1063 {
1064   return (this->per_client_.remove (e) == 0);
1065 }
1066
1067 bool
1068 LiveCheck::add_per_client_listener (LiveListener *l,
1069                                     ImplementationRepository::ServerObject_ptr ref)
1070 {
1071   if (!this->running_)
1072     return false;
1073
1074   LiveEntry *entry = 0;
1075   ACE_NEW_RETURN (entry, LiveEntry (this, l->server (), true, ref, 0), false);
1076
1077   if (this->per_client_.insert_tail(entry) == 0)
1078     {
1079       entry->add_listener (l);
1080
1081       if (!this->in_handle_timeout ())
1082         {
1083           ++this->token_;
1084           this->reactor()->schedule_timer (this,
1085                                            reinterpret_cast<void *>(this->token_),
1086                                            ACE_Time_Value::zero);
1087         }
1088       else
1089         {
1090           this->want_timeout_ = true;
1091           this->deferred_timeout_ = ACE_Time_Value::zero;
1092         }
1093       return true;
1094     }
1095   return false;
1096 }
1097
1098 bool
1099 LiveCheck::add_poll_listener (LiveListener *l)
1100 {
1101   if (!this->running_)
1102     return false;
1103
1104   LiveEntry *entry = 0;
1105   ACE_CString key (l->server());
1106   int const result = entry_map_.find (key, entry);
1107   if (result == -1 || entry == 0)
1108     {
1109       return false;
1110     }
1111
1112   entry->add_listener (l);
1113   entry->reset_status ();
1114   l->status_changed (entry->status());
1115   return this->schedule_ping (entry);
1116 }
1117
1118 bool
1119 LiveCheck::add_listener (LiveListener *l)
1120 {
1121   if (!this->running_)
1122     return false;
1123
1124   LiveEntry *entry = 0;
1125   ACE_CString key (l->server());
1126   int const result = entry_map_.find (key, entry);
1127   if (result == -1 || entry == 0)
1128     {
1129       return false;
1130     }
1131
1132   entry->add_listener (l);
1133   return this->schedule_ping (entry);
1134 }
1135
1136 void
1137 LiveCheck::remove_listener (LiveListener *l)
1138 {
1139   if (!this->running_)
1140     return;
1141
1142   LiveEntry *entry = 0;
1143   ACE_CString key (l->server());
1144   int const result = entry_map_.find (key, entry);
1145   if (result != -1 && entry != 0)
1146     {
1147       entry->remove_listener (l);
1148     }
1149 }
1150
1151 bool
1152 LiveCheck::schedule_ping (LiveEntry *entry)
1153 {
1154   if (!this->running_)
1155     return false;
1156
1157   LiveStatus const status = entry->status();
1158   if (status == LS_PING_AWAY || status == LS_DEAD)
1159     {
1160       return status != LS_DEAD;
1161     }
1162
1163   ACE_Time_Value const now (ACE_OS::gettimeofday());
1164   ACE_Time_Value const next = entry->next_check ();
1165
1166   if (!this->in_handle_timeout ())
1167     {
1168       ACE_Time_Value delay = ACE_Time_Value::zero;
1169       if (next > now)
1170         {
1171           delay = next - now;
1172         }
1173
1174       ACE_Timer_Queue *tq = this->reactor ()->timer_queue ();
1175       if (!tq->is_empty ())
1176         {
1177           for (ACE_Timer_Queue_Iterator_T<ACE_Event_Handler*> &i = tq->iter ();
1178                !i.isdone (); i.next())
1179             {
1180               if (i.item ()->get_type () == this)
1181                 {
1182                   if (next >= tq->earliest_time ())
1183                     {
1184                       if (ImR_Locator_i::debug () > 2)
1185                         {
1186                           ORBSVCS_DEBUG ((LM_DEBUG,
1187                                           ACE_TEXT ("(%P|%t) LiveCheck::schedule_ping ")
1188                                           ACE_TEXT ("already scheduled\n")));
1189                         }
1190                       return true;
1191                     }
1192                   break;
1193                 }
1194             }
1195         }
1196       ++this->token_;
1197       if (ImR_Locator_i::debug () > 2)
1198         {
1199           ORBSVCS_DEBUG ((LM_DEBUG,
1200                           ACE_TEXT ("(%P|%t) LiveCheck::schedule_ping (%d),")
1201                           ACE_TEXT (" delay <%d,%d>\n"),
1202                           this->token_, delay.sec(), delay.usec()));
1203         }
1204       this->reactor()->schedule_timer (this,
1205                                        reinterpret_cast<void *>(this->token_),
1206                                        delay);
1207     }
1208   else
1209     {
1210       if (ImR_Locator_i::debug () > 2)
1211         {
1212           ORBSVCS_DEBUG ((LM_DEBUG,
1213                           ACE_TEXT ("(%P|%t) LiveCheck::schedule_ping deferred because we are in handle timeout\n")));
1214         }
1215       if (!this->want_timeout_ || next < this->deferred_timeout_)
1216         {
1217           this->want_timeout_ = true;
1218           this->deferred_timeout_ = next;
1219         }
1220     }
1221   return true;
1222 }
1223
1224 LiveStatus
1225 LiveCheck::is_alive (const char *server)
1226 {
1227   if (!this->running_)
1228     return LS_DEAD;
1229
1230   if (this->ping_interval_ == ACE_Time_Value::zero)
1231     {
1232       return LS_ALIVE;
1233     }
1234
1235   ACE_CString s(server);
1236   LiveEntry *entry = 0;
1237   int const result = entry_map_.find (s, entry);
1238   if (result == 0 && entry != 0)
1239     {
1240       return entry->status ();
1241     }
1242   return LS_DEAD;
1243 }