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