TAO/orbsvcs/ImplRepo_Service/LiveCheck.h

   1 // -*- C++ -*-
   2 /*
   3  * @file LiveCheck.h
   4  *
   5  * @author Phil Mesnier <mesnier_p@ociweb.com>
   6  */
   7
   8 #ifndef IMR_LIVECHECK_H_
   9 #define IMR_LIVECHECK_H_
  10
  11 #include "locator_export.h"
  12
  13 #include "ServerObjectS.h" // ServerObject_AMIS.h
  14
  15 #include "ace/Unbounded_Set.h"
  16 #include "ace/Hash_Map_Manager.h"
  17 #include "ace/SString.h"
  18 #include "ace/Event_Handler.h"
  19
  20 #if !defined (ACE_LACKS_PRAGMA_ONCE)
  21 #pragma once
  22 #endif /* ACE_LACKS_PRAGMA_ONCE */
  23
  24 #include "tao/Intrusive_Ref_Count_Handle_T.h"
  25 #include <atomic>
  26
  27 class LiveCheck;
  28 class LiveEntry;
  29 class PingReceiver;
  30
  31 //---------------------------------------------------------------------------
  32 /*
  33  * @enum LiveStatus
  34  *
  35  * @brief indication of the known condition of a target server
  36  *
  37  *  LS_INIT      - This is a new entry
  38  *  LS_UNKNOWN   - The server status was reset
  39  *  LS_PING_AWAY - A ping request has been issued, waiting for reply
  40  *  LS_DEAD      - The ping failed for reasons other than POA Activation
  41  *  LS_ALIVE     - The server positively acknowledged a ping
  42  *  LS_TRANSIENT - The server connected, but actively raised a transient
  43  *  LS_LAST_TRANSIENT - The maximum number of retries is reached
  44  *  LS_TIMEDOUT  - The server connected, but never returned any result.
  45  *  LS_CANCELED  - The ping was canceled by the owner
  46  */
  47 enum LiveStatus {
  48   LS_INIT,
  49   LS_UNKNOWN,
  50   LS_PING_AWAY,
  51   LS_DEAD,
  52   LS_ALIVE,
  53   LS_TRANSIENT,
  54   LS_LAST_TRANSIENT,
  55   LS_TIMEDOUT,
  56   LS_CANCELED
  57 };
  58
  59 //---------------------------------------------------------------------------
  60 /*
  61  * @class LiveListener
  62  *
  63  * @brief An interface for receiving asynch liveness status updates
  64  *
  65  * The code waiting on a confirmation of liveness status creates an instance
  66  * of a LiveListener and registers it with the LiveCheck object.
  67  * When the desired ping occurs, the status_changed method is called and the
  68  * listener is unregistered. It is up to the owner of the listener to re-
  69  * register if the ping result was inconclusive, such as a status of TRANSIENT
  70  * or TIMEDOUT. Such a decision is based on configuration settings.
  71  */
  72 class Locator_Export LiveListener
  73 {
  74  public:
  75   /// Construct a new listener. The server name supplied is used to
  76   /// look up a listener entry in the LiveCheck map.
  77   LiveListener (const char *server);
  78
  79   virtual ~LiveListener ();
  80
  81   /// called by the asynch ping receiver when a reply or an exception
  82   /// is received. Returns true if finished listening
  83   virtual bool status_changed (LiveStatus status) = 0;
  84
  85   /// Accessor for the server name. Used by the LiveCheck to associate a listener
  86   const char *server () const;
  87
  88   LiveListener *_add_ref ();
  89   void _remove_ref ();
  90
  91  protected:
  92   ACE_CString server_;
  93
  94  private:
  95   std::atomic<int> refcount_;
  96 };
  97
  98 typedef TAO_Intrusive_Ref_Count_Handle<LiveListener> LiveListener_ptr;
  99
 100 //---------------------------------------------------------------------------
 101 /*
 102  * @class LiveEntry
 103  *
 104  * @brief Contains a list of interested listeners for a server
 105  *
 106  * Each server the Locator is interested in has a live entry instance.
 107  * This holds the liveliness status and determines the next allowed time
 108  * for a ping. Instances of the LiveEntry class are retained until the
 109  * locator is no longer interested in the target server.
 110  */
 111 class Locator_Export LiveEntry
 112 {
 113  public:
 114   LiveEntry (LiveCheck *owner,
 115              const char *server,
 116              bool may_ping,
 117              ImplementationRepository::ServerObject_ptr ref,
 118              int pid);
 119   ~LiveEntry ();
 120
 121   void release_callback ();
 122   void add_listener (LiveListener *ll);
 123   void remove_listener (LiveListener *ll);
 124   LiveStatus status () const;
 125   void status (LiveStatus l);
 126   void reset_status ();
 127
 128   /// the current state value as text
 129   static const char *status_name (LiveStatus s);
 130
 131   void update_listeners ();
 132   bool validate_ping (bool &want_reping, ACE_Time_Value &next);
 133   void do_ping (PortableServer::POA_ptr poa);
 134   const ACE_Time_Value &next_check () const;
 135   static void set_reping_limit (int max);
 136   bool reping_available () const;
 137   int next_reping ();
 138   void max_retry_msec (int max);
 139   const char *server_name () const;
 140   void set_pid (int pid);
 141   bool has_pid (int pid) const;
 142   int pid () const;
 143   bool may_ping () const;
 144
 145  private:
 146   LiveCheck *owner_;
 147   ACE_CString server_;
 148   ImplementationRepository::ServerObject_var ref_;
 149   LiveStatus liveliness_;
 150   ACE_Time_Value next_check_;
 151   int repings_;
 152   int max_retry_;
 153   bool may_ping_;
 154
 155   typedef ACE_Unbounded_Set<LiveListener_ptr> Listen_Set;
 156   Listen_Set listeners_;
 157   TAO_SYNCH_MUTEX lock_;
 158   PortableServer::ServantBase_var callback_;
 159   int pid_;
 160
 161   static const int reping_msec_ [];
 162   static int reping_limit_;
 163 };
 164
 165 //---------------------------------------------------------------------------
 166 /*
 167  * @class PingReceiver
 168  *
 169  * @brief callback handler for asynch ping requests
 170  *
 171  * An instance of the ping receiver is used to handle the reply from a ping
 172  * request. Instances are created for the ping, then destroyed.
 173  `*/
 174 class Locator_Export PingReceiver :
 175   public virtual POA_ImplementationRepository::AMI_ServerObjectHandler
 176 {
 177  public:
 178   PingReceiver (LiveEntry * entry, PortableServer::POA_ptr poa);
 179   virtual ~PingReceiver ();
 180
 181   /// Called by the entry if it is no longer interested in the result of
 182   /// a ping.
 183   void cancel ();
 184
 185   /// Called when an anticipated ping reply is received
 186   void ping ();
 187
 188   /// Called when an anticipated ping raises an exception
 189   void ping_excep (Messaging::ExceptionHolder * excep_holder);
 190
 191  private:
 192   PortableServer::POA_var poa_;
 193   LiveEntry * entry_;
 194 };
 195
 196
 197 //---------------------------------------------------------------------------
 198 /*
 199  * @class LC_TimeoutGuard
 200  *
 201  * @brief A helper object to avoid reentrancy in the handle_timout method
 202  *
 203  * The LiveCheck::handle_timeout may be called reentrantly on a single thread
 204  * if the sending of a ping uses non-blocking connection establishment. If a
 205  * connection must be established before the ping can be sent, that may involve
 206  * waiting in the reactor, possibly handing other requests, and possibly even
 207  * subsequent timeouts.
 208  * */
 209
 210 #if (ACE_SIZEOF_VOID_P == 8)
 211 typedef ACE_INT64 LC_token_type;
 212 #else
 213 typedef ACE_INT32 LC_token_type;
 214 #endif
 215
 216 class Locator_Export LC_TimeoutGuard
 217 {
 218  public:
 219   /// Construct a new stack-based guard. This sets a flag in the owner that will
 220   /// be cleared on destruction.
 221   LC_TimeoutGuard (LiveCheck *owner, LC_token_type token);
 222
 223   /// Releases the flag. If the LiveCheck received any requests for an immediate
 224   /// or deferred ping during this time, schedule it now.
 225   ~LC_TimeoutGuard ();
 226
 227   /// Returns true if the in handle timeout in the owner was already set.
 228   bool blocked () const;
 229
 230  private:
 231   LiveCheck *owner_;
 232   LC_token_type token_;
 233   bool blocked_;
 234 };
 235
 236 //---------------------------------------------------------------------------
 237 /*
 238  * @class LiveCheck
 239  *
 240  * @brief The manager class used for pinging servers as needed.
 241  *
 242  * The LiveCheck class maintains a map of named LiveEntries. When the locator
 243  * needs to determine the liveliness of a server, registers a LiveListener
 244  * for the desired server. A ping to the server is then scheduled, based on the
 245  * limits determined by the entry's state.
 246  */
 247 class Locator_Export LiveCheck : public ACE_Event_Handler
 248 {
 249  public:
 250   friend class LC_TimeoutGuard;
 251
 252   LiveCheck ();
 253   ~LiveCheck ();
 254
 255   void init (CORBA::ORB_ptr orb,
 256              const ACE_Time_Value &interval);
 257   void shutdown ();
 258   int handle_timeout (const ACE_Time_Value &current_time,
 259                       const void *act = 0);
 260   bool has_server (const char *server);
 261   void add_server (const char *server,
 262                    bool may_ping,
 263                    ImplementationRepository::ServerObject_ptr ref,
 264                    int pid);
 265   void set_pid (const char *server, int pid);
 266   void remove_server (const char *server, int pid);
 267   bool remove_per_client_entry (LiveEntry *entry);
 268   bool add_listener (LiveListener *listener);
 269   bool add_poll_listener (LiveListener *listener);
 270   bool add_per_client_listener (LiveListener *listener,
 271                                 ImplementationRepository::ServerObject_ptr ref);
 272   void remove_listener (LiveListener *listener);
 273   bool schedule_ping (LiveEntry *entry);
 274   LiveStatus is_alive (const char *server);
 275   const ACE_Time_Value &ping_interval () const;
 276
 277  private:
 278   void enter_handle_timeout ();
 279   void exit_handle_timeout ();
 280   bool in_handle_timeout ();
 281   void remove_deferred_servers ();
 282
 283   typedef ACE_Hash_Map_Manager_Ex<ACE_CString,
 284                                   LiveEntry *,
 285                                   ACE_Hash<ACE_CString>,
 286                                   ACE_Equal_To<ACE_CString>,
 287                                   ACE_Null_Mutex> LiveEntryMap;
 288   typedef ACE_Unbounded_Set<LiveEntry *> PerClientStack;
 289   typedef std::pair<ACE_CString, int> NamePidPair;
 290   typedef ACE_Unbounded_Set<NamePidPair> NamePidStack;
 291
 292   LiveEntryMap entry_map_;
 293   PerClientStack per_client_;
 294   PortableServer::POA_var poa_;
 295   ACE_Time_Value ping_interval_;
 296   bool running_;
 297   LC_token_type token_;
 298   /// Flag to check whether we are in handle timeout. Because this can be
 299   /// called re-entrant it is zero when we are not in handle timeout
 300   int handle_timeout_busy_;
 301   bool want_timeout_;
 302   ACE_Time_Value deferred_timeout_;
 303   /// Contains a list of servers which got removed during the handle_timeout,
 304   /// these will be removed at the end of the handle_timeout.
 305   NamePidStack removed_entries_;
 306 };
 307
 308 #endif /* IMR_LIVECHECK_H_  */