TAO/tao/Connection_Handler.cpp

   1 // -*- C++ -*-
   2 #include "tao/Connection_Handler.h"
   3 #include "tao/ORB_Core.h"
   4 #include "tao/debug.h"
   5 #include "tao/Resume_Handle.h"
   6 #include "tao/Transport.h"
   7 #include "tao/Wait_Strategy.h"
   8
   9 #include "ace/SOCK.h"
  10 #include "ace/Reactor.h"
  11 #include "ace/os_include/sys/os_socket.h"
  12 #include "ace/Svc_Handler.h"
  13
  14 #if !defined (__ACE_INLINE__)
  15 #include "tao/Connection_Handler.inl"
  16 #endif /* __ACE_INLINE__ */
  17
  18 TAO_BEGIN_VERSIONED_NAMESPACE_DECL
  19
  20 TAO_Connection_Handler::TAO_Connection_Handler (TAO_ORB_Core *orb_core)
  21   : orb_core_ (orb_core),
  22     transport_ (nullptr),
  23     connection_pending_ (false),
  24     is_closed_ (false)
  25 {
  26   // Put ourselves in the connection wait state as soon as we get
  27   // created
  28   this->state_changed (TAO_LF_Event::LFS_CONNECTION_WAIT,
  29                        this->orb_core_->leader_follower ());
  30 }
  31
  32 TAO_Connection_Handler::~TAO_Connection_Handler ()
  33 {
  34 }
  35
  36 int
  37 TAO_Connection_Handler::shared_open ()
  38 {
  39   // This reference counting is related to asynch connections.  It
  40   // should probably be managed by the ACE_Strategy_Connector, since
  41   // that's really the reference being managed here.  also, whether
  42   // open ultimately succeeds or fails, the connection attempted is
  43   // ending, so the reference must be removed in any case.
  44   this->cancel_pending_connection();
  45
  46   return 0;
  47 }
  48
  49 int
  50 TAO_Connection_Handler::set_socket_option (ACE_SOCK &sock,
  51                                            int snd_size,
  52                                            int rcv_size)
  53 {
  54 #if !defined (ACE_LACKS_SO_SNDBUF)
  55   if (snd_size != 0
  56       && sock.set_option (SOL_SOCKET,
  57                           SO_SNDBUF,
  58                           (void *) &snd_size,
  59                           sizeof (snd_size)) == -1)
  60   {
  61     if (TAO_debug_level)
  62       TAOLIB_DEBUG ((LM_ERROR,
  63                   ACE_TEXT ("TAO (%P|%t) - Connection_Handler::")
  64                   ACE_TEXT ("set_socket_option, setting SO_SNDBUF failed ")
  65                   ACE_TEXT ("'%m'\n")));
  66
  67     if (errno != ENOTSUP)
  68       return -1;
  69   }
  70 #endif /* !ACE_LACKS_SO_SNDBUF */
  71
  72 #if !defined (ACE_LACKS_SO_RCVBUF)
  73   if (rcv_size != 0
  74       && sock.set_option (SOL_SOCKET,
  75                           SO_RCVBUF,
  76                           (void *) &rcv_size,
  77                           sizeof (int)) == -1)
  78   {
  79       if (TAO_debug_level)
  80         TAOLIB_ERROR ((LM_ERROR,
  81                     ACE_TEXT ("TAO (%P|%t) - Connection_Handler::")
  82                     ACE_TEXT ("set_socket_option, setting SO_RCVBUF failed ")
  83                     ACE_TEXT ("'%m'\n")));
  84
  85     if (errno != ENOTSUP)
  86       return -1;
  87   }
  88 #endif /* !ACE_LACKS_SO_RCVBUF */
  89
  90 #if defined (ACE_LACKS_SO_SNDBUF) && defined (ACE_LACKS_SO_RCVBUF)
  91   ACE_UNUSED_ARG (snd_size);
  92   ACE_UNUSED_ARG (rcv_size);
  93 #endif
  94
  95   // Set the close-on-exec flag for that file descriptor. If the
  96   // operation fails we are out of luck (some platforms do not support
  97   // it and return -1).
  98   (void) sock.enable (ACE_CLOEXEC);
  99
 100   return 0;
 101 }
 102
 103 int
 104 TAO_Connection_Handler::svc_i ()
 105 {
 106   int result = 0;
 107
 108   if (TAO_debug_level > 0)
 109     TAOLIB_DEBUG ((LM_DEBUG,
 110                 ACE_TEXT ("TAO (%P|%t) - Connection_Handler::svc_i begin\n")));
 111
 112   // Here we simply synthesize the "typical" event loop one might find
 113   // in a reactive handler, except that this can simply block waiting
 114   // for input.
 115
 116   ACE_Time_Value *max_wait_time = nullptr;
 117   ACE_Time_Value timeout;
 118   ACE_Time_Value current_timeout;
 119
 120   if (this->orb_core_->thread_per_connection_timeout (timeout))
 121     {
 122       current_timeout = timeout;
 123       max_wait_time = &current_timeout;
 124     }
 125
 126   TAO_Resume_Handle rh (this->orb_core_, ACE_INVALID_HANDLE);
 127
 128   // We exit of the loop if
 129   // - If the ORB core is shutdown by another thread
 130   // - Or if the transport is null. This could happen if an error
 131   // occurred.
 132   // - Or if during processing a return value of -1 is received.
 133   while (!this->orb_core_->has_shutdown () && result >= 0)
 134     {
 135       // Let the transport know that it is used
 136       (void) this->transport ()->update_transport ();
 137
 138       result = this->transport ()->handle_input (rh, max_wait_time);
 139
 140       if (result == -1 && errno == ETIME)
 141         {
 142           // Ignore timeouts, they are only used to wake up and
 143           // shutdown.
 144           result = 0;
 145
 146           // Reset errno to make sure we don't trip over an old value
 147           // of errno in case it is not reset when the recv() call
 148           // fails if the socket has been closed.
 149           errno = 0;
 150         }
 151       else if (result == -1)
 152         {
 153           // Something went wrong with the socket. Just quit
 154           return result;
 155         }
 156
 157       current_timeout = timeout;
 158
 159       if (TAO_debug_level > 0)
 160         TAOLIB_DEBUG ((LM_DEBUG,
 161                     "TAO (%P|%t) - Connection_Handler::svc_i - "
 162                        "loop <%d> shutdown = %d\n", current_timeout.msec (), this->is_closed_));
 163       if (this->is_closed_)
 164         {
 165           return result;
 166         }
 167     }
 168
 169   if (TAO_debug_level > 0)
 170     TAOLIB_DEBUG  ((LM_DEBUG,
 171                  "TAO (%P|%t) - Connection_Handler::svc_i - end\n"));
 172
 173   return result;
 174 }
 175
 176 void
 177 TAO_Connection_Handler::transport (TAO_Transport* transport)
 178 {
 179   this->transport_ = transport;
 180
 181   // Enable reference counting on the event handler.
 182   this->transport_->event_handler_i ()->reference_counting_policy ().value (
 183       ACE_Event_Handler::Reference_Counting_Policy::ENABLED);
 184 }
 185
 186 int
 187 TAO_Connection_Handler::handle_output_eh (ACE_HANDLE, ACE_Event_Handler * eh)
 188 {
 189   // Let the transport that it is going to be used
 190   (void) this->transport ()->update_transport ();
 191
 192   // Instantiate the resume handle here.. This will automatically
 193   // resume the handle once data is written..
 194   TAO_Resume_Handle resume_handle (this->orb_core (), eh->get_handle ());
 195
 196   int return_value = 0;
 197   this->pre_io_hook (return_value);
 198   if (return_value != 0)
 199     {
 200       resume_handle.set_flag (TAO_Resume_Handle::TAO_HANDLE_LEAVE_SUSPENDED);
 201       return return_value;
 202     }
 203
 204   // The default constraints are to never block.
 205   TAO::Transport::Drain_Constraints dc;
 206   if (this->transport ()->handle_output (dc) == TAO_Transport::DR_ERROR)
 207     {
 208       return_value = -1;
 209     }
 210
 211   this->pos_io_hook (return_value);
 212
 213   if (return_value != 0)
 214     {
 215       resume_handle.set_flag (TAO_Resume_Handle::TAO_HANDLE_LEAVE_SUSPENDED);
 216     }
 217
 218   return return_value;
 219 }
 220
 221 int
 222 TAO_Connection_Handler::handle_input_eh (ACE_HANDLE h, ACE_Event_Handler *eh)
 223 {
 224   // If we can't process upcalls just return
 225   if (!this->transport ()->wait_strategy ()->can_process_upcalls ())
 226     {
 227       if (TAO_debug_level > 6)
 228         TAOLIB_DEBUG ((LM_DEBUG,
 229                     "TAO (%P|%t) - Connection_Handler[%d]::handle_input_eh, "
 230                     "not going to handle_input on transport "
 231                     "because upcalls temporarily suspended on this thread\n",
 232                     this->transport()->id()));
 233
 234       // defer upcall at leader_follower
 235       if (this->transport ()->wait_strategy ()->defer_upcall (eh) != 0)
 236         {
 237           if (TAO_debug_level > 5)
 238             TAOLIB_ERROR ((LM_ERROR,
 239                       "TAO (%P|%t) - Connection_Handler[%d]::handle_input_eh, "
 240                       "Error deferring upcall handler[%d]\n",
 241                       this->transport ()->id (),
 242                       eh->get_handle ()));
 243           return -1;
 244         }
 245       // Returning 0 causes the wait strategy to exit and the leader thread
 246       // to enter the reactor's select() call.
 247       return 0;
 248     }
 249
 250   int const result = this->handle_input_internal (h, eh);
 251
 252   if (result == -1)
 253     {
 254       this->close_connection ();
 255       return 0;
 256     }
 257
 258   return result;
 259 }
 260
 261 int
 262 TAO_Connection_Handler::handle_input_internal (
 263     ACE_HANDLE h, ACE_Event_Handler * eh)
 264 {
 265   // Let the transport know that it is used
 266   (void) this->transport ()->update_transport ();
 267
 268   // Grab the transport id now and use the cached value for printing
 269   // since the  transport could disappear by the time the thread
 270   // returns.
 271   size_t const t_id = this->transport ()->id ();
 272
 273   if (TAO_debug_level > 6)
 274     {
 275       ACE_HANDLE const handle = eh->get_handle();
 276       TAOLIB_DEBUG ((LM_DEBUG,
 277                   "TAO (%P|%t) - Connection_Handler[%d]::handle_input_internal, "
 278                   "handle = %d/%d\n",
 279                   t_id, handle, h));
 280     }
 281
 282   TAO_Resume_Handle resume_handle (this->orb_core (), eh->get_handle ());
 283
 284   int return_value = 0;
 285
 286   this->pre_io_hook (return_value);
 287
 288   if (return_value != 0)
 289     return return_value;
 290
 291   return_value = this->transport ()->handle_input (resume_handle);
 292
 293   this->pos_io_hook (return_value);
 294
 295   // Bug 1647; might need to change resume_handle's flag or
 296   // change handle_input return value.
 297   resume_handle.handle_input_return_value_hook(return_value);
 298
 299   if (TAO_debug_level > 6)
 300     {
 301       ACE_HANDLE const handle = eh->get_handle ();
 302       TAOLIB_DEBUG ((LM_DEBUG,
 303                   "TAO (%P|%t) - Connection_Handler[%d]::handle_input_internal, "
 304                   "handle = %d/%d, retval = %d\n",
 305                   t_id, handle, h, return_value));
 306     }
 307
 308   if (return_value != 0)
 309     {
 310       resume_handle.set_flag (TAO_Resume_Handle::TAO_HANDLE_LEAVE_SUSPENDED);
 311     }
 312
 313   return return_value;
 314 }
 315
 316 int
 317 TAO_Connection_Handler::close_connection_eh (ACE_Event_Handler *eh)
 318 {
 319   if (this->is_closed_)
 320     {
 321       return 1;
 322     }
 323
 324   this->is_closed_ = true;
 325
 326   // Save the ID for debugging messages
 327   ACE_HANDLE const handle = eh->get_handle ();
 328
 329   size_t const id = this->transport ()->id ();
 330   if (TAO_debug_level)
 331     {
 332       TAOLIB_DEBUG  ((LM_DEBUG,
 333                    "TAO (%P|%t) - Connection_Handler[%d]::"
 334                    "close_connection_eh, purging entry from cache\n",
 335                    handle));
 336     }
 337
 338   this->transport ()->pre_close ();
 339
 340   // @@ This seems silly, but if we have no reason to be in the
 341   // reactor, then we dont remove ourselves.
 342   if (this->transport ()->wait_strategy ()->is_registered ())
 343     {
 344       ACE_Reactor *eh_reactor = eh->reactor ();
 345
 346       if (this->orb_core_->has_shutdown () == 0)
 347         {
 348           // If the ORB is nil, get the reactor from orb_core which gets it
 349           // from LF.
 350           if (eh_reactor == nullptr)
 351             eh_reactor = this->transport()->orb_core()->reactor ();
 352         }
 353
 354       // The Reactor must not be null, otherwise something else is
 355       // horribly broken.
 356       ACE_ASSERT (eh_reactor != nullptr);
 357
 358       if (TAO_debug_level)
 359         {
 360           TAOLIB_DEBUG  ((LM_DEBUG,
 361                        "TAO (%P|%t) - Connection_Handler[%d]::"
 362                        "close_connection_eh, removing from the reactor\n",
 363                        handle));
 364         }
 365
 366       // Use id instead of handle. Why? "handle" may be invalid for RW
 367       // cases  when drop_reply_on_shutdown is on, and when the
 368       // orb_core is shutting down. This means that the handler will
 369       // be left behind in the reactor which would create problems
 370       // later. Just forcefully remove them. If none exists reactor
 371       // will make things safer.
 372       ACE_HANDLE tmp_handle = handle;
 373       if (this->orb_core_->has_shutdown ())
 374         tmp_handle = (ACE_HANDLE) id;
 375
 376       eh_reactor->remove_handler (tmp_handle,
 377                                   ACE_Event_Handler::ALL_EVENTS_MASK |
 378                                   ACE_Event_Handler::DONT_CALL);
 379
 380       // Also cancel any timers, we may create those for time-limited
 381       // buffering
 382       if (TAO_debug_level)
 383         {
 384           TAOLIB_DEBUG  ((LM_DEBUG,
 385                        "TAO (%P|%t) - Connection_Handler[%d]::"
 386                        "close_connection_eh, cancel all timers\n",
 387                        handle));
 388         }
 389
 390       eh_reactor->cancel_timer (eh);
 391
 392       // @@ This seems silly, the reactor is a much better authority to
 393       //    find out if a handle is registered...
 394       this->transport ()->wait_strategy ()->is_registered (false);
 395     }
 396
 397   // This call should be made only after the cache and reactor are
 398   // cleaned up. This call can make upcalls to the application which
 399   // in turn can make remote calls (Bug 1551 and Bug 1482). The remote
 400   // calls from the application can try to use this handler from the
 401   // cache or from the reactor. So clean them up before this is
 402   // called.
 403   this->transport ()->send_connection_closed_notifications ();
 404   this->state_changed (TAO_LF_Event::LFS_CONNECTION_CLOSED,
 405                        this->orb_core_->leader_follower ());
 406
 407   if (TAO_debug_level)
 408     {
 409       TAOLIB_DEBUG  ((LM_DEBUG,
 410                    "TAO (%P|%t) - Connection_Handler[%d]::"
 411                    "close_connection_eh end\n",
 412                    id));
 413     }
 414
 415   return 1;
 416 }
 417
 418 int
 419 TAO_Connection_Handler::set_dscp_codepoint (CORBA::Boolean)
 420 {
 421   return 0;
 422 }
 423
 424 int
 425 TAO_Connection_Handler::set_dscp_codepoint (CORBA::Long)
 426 {
 427   return 0;
 428 }
 429
 430 int
 431 TAO_Connection_Handler::release_os_resources ()
 432 {
 433   return 0;
 434 }
 435
 436 void
 437 TAO_Connection_Handler::pre_io_hook (int &)
 438 {
 439 }
 440
 441 void
 442 TAO_Connection_Handler::pos_io_hook (int &)
 443 {
 444 }
 445
 446 int
 447 TAO_Connection_Handler::close_handler (u_long)
 448 {
 449   if (!this->is_closed_)
 450     {
 451       this->is_closed_ = true;
 452
 453       this->state_changed (TAO_LF_Event::LFS_CONNECTION_CLOSED,
 454                             this->orb_core_->leader_follower ());
 455
 456       // If there was a pending connection cancel it.
 457       this->cancel_pending_connection ();
 458
 459       // Purge transport from cache if it's in cache.
 460       this->transport ()->purge_entry();
 461     }
 462
 463   return 0;
 464 }
 465
 466 TAO_END_VERSIONED_NAMESPACE_DECL