diff options
Diffstat (limited to 'TAO/tao/UIOP_Connect.cpp')
| -rw-r--r-- | TAO/tao/UIOP_Connect.cpp | 501 |
1 files changed, 4 insertions, 497 deletions
diff --git a/TAO/tao/UIOP_Connect.cpp b/TAO/tao/UIOP_Connect.cpp index fceff511557..e1bc7a1ab75 100644 --- a/TAO/tao/UIOP_Connect.cpp +++ b/TAO/tao/UIOP_Connect.cpp @@ -35,6 +35,8 @@ TAO_UIOP_Handler_Base::resume_handler (ACE_Reactor *) return -1; } +// **************************************************************** + // @@ For pluggable protocols, added a reference to // the corresponding transport obj. TAO_UIOP_Server_Connection_Handler::TAO_UIOP_Server_Connection_Handler (ACE_Thread_Manager *t) @@ -743,6 +745,8 @@ TAO_UIOP_Server_Connection_Handler::handle_input (ACE_HANDLE) return result; } +// **************************************************************** + // @@ For pluggable protocols, added a reference to the corresponding // transport obj. TAO_UIOP_Client_Connection_Handler::TAO_UIOP_Client_Connection_Handler (ACE_Thread_Manager *t) @@ -818,51 +822,6 @@ TAO_UIOP_Client_Connection_Handler::handle_input (ACE_HANDLE) } int -TAO_UIOP_Client_Connection_Handler::check_unexpected_data (void) -{ - // We're a client, so we're not expecting to see input. Still we - // better check what it is! - char ignored; - ssize_t ret = this->peer().recv (&ignored, - sizeof ignored, - MSG_PEEK); - switch (ret) - { - case 0: - case -1: - // 0 is a graceful shutdown - // -1 is a somewhat ugly shutdown - // - // Both will result in us returning -1 and this connection - // getting closed - // - if (TAO_orbdebug) - ACE_DEBUG ((LM_WARNING, - "UIOP_Client_Connection_Handler::handle_input: " - "closing connection on fd %d\n", - this->peer().get_handle ())); - break; - - case 1: - // - // @@ Fix me!! - // - // This should be the close connection message. Since we don't - // handle this yet, log an error, and close the connection. - ACE_ERROR ((LM_WARNING, - "UIOP_Client_Connection_Handler::handle_input received " - "input while not expecting a response; " - "closing connection on fd %d\n", - this->peer().get_handle ())); - break; - } - - // We're not expecting input at this time, so we'll always - // return -1 for now. - return -1; -} - -int TAO_UIOP_Client_Connection_Handler::handle_close (ACE_HANDLE handle, ACE_Reactor_Mask rm) { @@ -904,458 +863,6 @@ TAO_UIOP_Client_Connection_Handler::close (u_long) // **************************************************************** -TAO_RW_Client_Connection_Handler::TAO_RW_Client_Connection_Handler (ACE_Thread_Manager *t) - : TAO_UIOP_Client_Connection_Handler (t) -{ -} - -TAO_RW_Client_Connection_Handler::~TAO_RW_Client_Connection_Handler (void) -{ -} - -int -TAO_RW_Client_Connection_Handler::send_request (TAO_ORB_Core* orb_core, - TAO_OutputCDR &stream, - int is_twoway) -{ - TAO_FUNCTION_PP_TIMEPROBE (TAO_CLIENT_CONNECTION_HANDLER_SEND_REQUEST_START); - - // NOTE: Here would also be a fine place to calculate a digital - // signature for the message and place it into a preallocated slot - // in the "ServiceContext". Similarly, this is a good spot to - // encrypt messages (or just the message bodies) if that's needed in - // this particular environment and that isn't handled by the - // networking infrastructure (e.g. IPSEC). - // - // We could call a template method to do all this stuff, and if the - // connection handler were obtained from a factory, then this could - // be dynamically linked in (wouldn't that be cool/freaky?) - - // Send the request - int success = - ACE_static_cast (int, TAO_GIOP::send_request (this->uiop_transport_, - stream, - orb_core)); - - TAO_MINIMAL_TIMEPROBE (GIOP_SEND_REQUEST_RETURN); - - if (!success) - return -1; - - return 0; -} - -int -TAO_RW_Client_Connection_Handler::resume_handler (ACE_Reactor *) -{ - // Since we don't suspend, we don't have to resume. - return 0; -} - -// **************************************************************** - -TAO_ST_Client_Connection_Handler::TAO_ST_Client_Connection_Handler (ACE_Thread_Manager *t) - : TAO_UIOP_Client_Connection_Handler (t) -{ -} - -TAO_ST_Client_Connection_Handler::~TAO_ST_Client_Connection_Handler (void) -{ -} - -int -TAO_ST_Client_Connection_Handler::open (void *something) -{ - int result = TAO_UIOP_Client_Connection_Handler::open (something); - - if (result != 0) - return result; - - // Now we must register ourselves with the reactor for input events - // which will detect GIOP Reply messages and EOF conditions. - ACE_Reactor *r = TAO_ORB_Core_instance ()->reactor (); - return r->register_handler (this, - ACE_Event_Handler::READ_MASK); -} - -// @@ this seems odd that the connection handler would call methods in the -// GIOP object. Some of this mothod's functionality should be moved -// to GIOP. fredk -int -TAO_ST_Client_Connection_Handler::send_request (TAO_ORB_Core* orb_core, - TAO_OutputCDR &stream, - int is_twoway) -{ - TAO_FUNCTION_PP_TIMEPROBE (TAO_CLIENT_CONNECTION_HANDLER_SEND_REQUEST_START); - - // NOTE: Here would also be a fine place to calculate a digital - // signature for the message and place it into a preallocated slot - // in the "ServiceContext". Similarly, this is a good spot to - // encrypt messages (or just the message bodies) if that's needed in - // this particular environment and that isn't handled by the - // networking infrastructure (e.g. IPSEC). - // - // We could call a template method to do all this stuff, and if the - // connection handler were obtained from a factory, then this could - // be dynamically linked in (wouldn't that be cool/freaky?) - - // Send the request - int success = (int) TAO_GIOP::send_request (this->uiop_transport_, - stream, - orb_core); - TAO_MINIMAL_TIMEPROBE (GIOP_SEND_REQUEST_RETURN); - - if (!success) - return -1; - - if (is_twoway) - { - // Set the state so that we know we're looking for a response. - this->expecting_response_ = 1; - - // Go into a loop, waiting until it's safe to try to read - // something on the socket. The handle_input() method doesn't - // actualy do the read, though, proper behavior based on what is - // read may be different if we're not using GIOP above here. - // So, we leave the reading of the response to the caller of - // this method, and simply insure that this method doesn't - // return until such time as doing a recv() on the socket would - // actually produce fruit. - ACE_Reactor *r = orb_core->reactor (); - - int ret = 0; - - while (ret != -1 && ! this->input_available_) - ret = r->handle_events (); - - this->input_available_ = 0; - // We can get events now, b/c we want them! - - // We're no longer expecting a response! - this->expecting_response_ = 0; - } - - return 0; -} - -int -TAO_ST_Client_Connection_Handler::handle_input (ACE_HANDLE) -{ - int retval = 0; - - if (this->expecting_response_) - { - this->input_available_ = 1; - // Temporarily remove ourself from notification so that if - // another sub event loop is in effect still waiting for its - // response, it doesn't spin tightly gobbling up CPU. - TAO_ORB_Core_instance ()->reactor ()->suspend_handler (this); - } - else - retval = this->check_unexpected_data (); - - return retval; -} - -int -TAO_ST_Client_Connection_Handler::resume_handler (ACE_Reactor *reactor) -{ - return reactor->resume_handler (this); -} - -// **************************************************************** - -TAO_MT_UIOP_Client_Connection_Handler::TAO_MT_UIOP_Client_Connection_Handler (ACE_Thread_Manager *t) - : TAO_UIOP_Client_Connection_Handler (t), - calling_thread_ (ACE_OS::NULL_thread), - cond_response_available_ (0), - orb_core_ (0) -{ -} - -TAO_MT_UIOP_Client_Connection_Handler::~TAO_MT_UIOP_Client_Connection_Handler (void) -{ - delete this->cond_response_available_; -} - -int -TAO_MT_UIOP_Client_Connection_Handler::open (void *something) -{ - int result = TAO_UIOP_Client_Connection_Handler::open (something); - - if (result != 0) - return result; - - // Now we must register ourselves with the reactor for input events - // which will detect GIOP Reply messages and EOF conditions. - ACE_Reactor *r = TAO_ORB_Core_instance ()->reactor (); - return r->register_handler (this, - ACE_Event_Handler::READ_MASK); -} - -ACE_SYNCH_CONDITION* -TAO_MT_UIOP_Client_Connection_Handler::cond_response_available (TAO_ORB_Core* orb_core) -{ - // @@ TODO This condition variable should per-ORB-per-thread, not - // per-connection, it is a waste to have more than one of this in - // the same thread. - if (this->cond_response_available_ == 0) - { - ACE_NEW_RETURN (this->cond_response_available_, - ACE_SYNCH_CONDITION (orb_core->leader_follower_lock ()), - 0); - } - return this->cond_response_available_; -} - -int -TAO_MT_UIOP_Client_Connection_Handler::send_request (TAO_ORB_Core *orb_core, - TAO_OutputCDR &stream, - int is_twoway) -{ - TAO_FUNCTION_PP_TIMEPROBE (TAO_CLIENT_CONNECTION_HANDLER_SEND_REQUEST_START); - - // Save the ORB_Core for the handle_input callback... - this->orb_core_ = orb_core; - - // NOTE: Here would also be a fine place to calculate a digital - // signature for the message and place it into a preallocated slot - // in the "ServiceContext". Similarly, this is a good spot to - // encrypt messages (or just the message bodies) if that's needed in - // this particular environment and that isn't handled by the - // networking infrastructure (e.g. IPSEC). - // - // We could call a template method to do all this stuff, and if the - // connection handler were obtained from a factory, then this could - // be dynamically linked in (wouldn't that be cool/freaky?) - - if (!is_twoway) - { - // Send the request - int success = (int) TAO_GIOP::send_request (this->uiop_transport_, - stream, - this->orb_core_); - - TAO_MINIMAL_TIMEPROBE (GIOP_SEND_REQUEST_RETURN); - - if (!success) - return -1; - } - else // is_twoway - { - ACE_Reactor *r = this->orb_core_->reactor (); - - if (this->reactor () != r) - { - ACE_Reactor_Mask mask = - ACE_Event_Handler::ALL_EVENTS_MASK | ACE_Event_Handler::DONT_CALL; - this->reactor ()->remove_handler (this, mask); - - r->register_handler (this, - ACE_Event_Handler::READ_MASK); - } - - if (this->orb_core_->leader_follower_lock ().acquire() == -1) - ACE_ERROR_RETURN ((LM_ERROR, - "(%P|%t) TAO_UIOP_Client_Connection_Handler::send_request: " - "Failed to get the lock.\n"), - -1); - - // Set the state so that we know we're looking for a response. - this->expecting_response_ = 1; - // remember in which thread the client connection handler was running - this->calling_thread_ = ACE_Thread::self (); - - // Send the request - int success = (int) TAO_GIOP::send_request (this->uiop_transport_, - stream, - orb_core); - - TAO_MINIMAL_TIMEPROBE (GIOP_SEND_REQUEST_RETURN); - - if (!success) - { - this->orb_core_->leader_follower_lock ().release (); - return -1; - } - - // check if there is a leader, but the leader is not us - if (this->orb_core_->leader_available () && - !this->orb_core_->I_am_the_leader_thread ()) - { - // wait as long as no input is available and/or - // no leader is available - while (!this->input_available_ && - this->orb_core_->leader_available ()) - { - ACE_SYNCH_CONDITION* cond = - this->cond_response_available (orb_core); - if (this->orb_core_->add_follower (cond) == -1) - ACE_ERROR ((LM_ERROR, - "(%P|%t) TAO_UIOP_Client_Connection_Handler::send_request: " - "Failed to add a follower thread\n")); - cond->wait (); - } - // now somebody woke us up to become a leader or to handle - // our input. We are already removed from the follower queue - if (this->input_available_) - { - // there is input waiting for me - if (this->orb_core_->leader_follower_lock ().release () == -1) - ACE_ERROR_RETURN ((LM_ERROR, - "(%P|%t) TAO_UIOP_Client_Connection_Handler::send_request: " - "Failed to release the lock.\n"), - -1); - // The following variables are safe, because we are not - // registered with the reactor any more. - this->input_available_ = 0; - this->expecting_response_ = 0; - this->calling_thread_ = ACE_OS::NULL_thread; - return 0; - } - } - - // Become a leader, because there is no leader or we have to - // update to a leader or we are doing nested upcalls in this - // case we do increase the refcount on the leader in - // TAO_ORB_Core. - - this->orb_core_->set_leader_thread (); - // this might increase the recount of the leader - - if (this->orb_core_->leader_follower_lock ().release () == -1) - ACE_ERROR_RETURN ((LM_ERROR, - "(%P|%t) TAO_UIOP_Client_Connection_Handler::send_request: " - "Failed to release the lock.\n"), - -1); - - r->owner (ACE_Thread::self ()); - - int ret = 0; - - while (ret != -1 && !this->input_available_) - ret = r->handle_events (); - - if (ret == -1) - ACE_ERROR_RETURN ((LM_ERROR, - "(%P|%t) TAO_UIOP_Client_Connection_Handler::send_request: " - "handle_events failed.\n"), - -1); - - // Wake up the next leader, we cannot do that in handle_input, - // because the woken up thread would try to get into - // handle_events, which is at the time in handle_input still - // occupied. - - if (this->orb_core_->unset_leader_wake_up_follower () == -1) - ACE_ERROR_RETURN ((LM_ERROR, - "(%P|%t) TAO_UIOP_Client_Connection_Handler::send_request: " - "Failed to unset the leader and wake up a new follower.\n"), - -1); - // Make use reusable - this->input_available_ = 0; - this->expecting_response_ = 0; - this->calling_thread_ = ACE_OS::NULL_thread; - } - - return 0; -} - -int -TAO_MT_UIOP_Client_Connection_Handler::handle_input (ACE_HANDLE) -{ - if (this->orb_core_ == 0) - this->orb_core_ = TAO_ORB_Core_instance (); - - if (this->orb_core_->leader_follower_lock ().acquire () == -1) - ACE_ERROR_RETURN ((LM_ERROR, - "(%P|%t) TAO_UIOP_Client_Connection_Handler::handle_input: " - "Failed to get the lock.\n"), - -1); - - if (!this->expecting_response_) - { - // we got something, but did not want - // @@ wake up an other thread, we are lost - - if (this->orb_core_->leader_follower_lock ().release () == -1) - ACE_ERROR_RETURN ((LM_ERROR, - "(%P|%t) TAO_UIOP_Client_Connection_Handler::handle_input: " - "Failed to release the lock.\n"), - -1); - return this->check_unexpected_data (); - } - - if (ACE_OS::thr_equal (this->calling_thread_, - ACE_Thread::self ())) - { - // We are now a leader getting its response. - this->input_available_ = 1; - - if (this->orb_core_->leader_follower_lock ().release () == -1) - ACE_ERROR_RETURN ((LM_ERROR, - "(%P|%t) TAO_UIOP_Client_Connection_Handler::handle_input: " - "Failed to release the lock.\n"), - -1); - this->orb_core_->reactor ()->suspend_handler (this); - // resume_handler is called in TAO_GIOP_Invocation::invoke - return 0; - } - else - { - // We are a leader, which got a response for one of the - // followers, which means we are now a thread running the wrong - // Client_Connection_Handler - - // At this point we might fail to remove the follower, because - // it has been already chosen to become the leader, so it is - // awake and will get this too. - ACE_SYNCH_CONDITION* cond = - this->cond_response_available (this->orb_core_); - - this->orb_core_->remove_follower (cond); - - if (this->orb_core_->leader_follower_lock ().release () == -1) - ACE_ERROR_RETURN ((LM_ERROR, - "(%P|%t) TAO_UIOP_Client_Connection_Handler::handle_input: " - "Failed to release the lock.\n"), - -1); - - this->orb_core_->reactor ()->suspend_handler (this); - // We should wake suspend the thread before we wake him up. - // resume_handler is called in TAO_GIOP_Invocation::invoke - - // @@ TODO (Michael): We might be able to optimize this in - // doing the suspend_handler as last thing, but I am not sure - // if a race condition would occur. - - if (this->orb_core_->leader_follower_lock ().acquire () == -1) - ACE_ERROR_RETURN ((LM_ERROR, - "(%P|%t) TAO_UIOP_Client_Connection_Handler::handle_input: " - "Failed to acquire the lock.\n"), - -1); - // The thread was already selected to become a leader, so we - // will be called again. - this->input_available_ = 1; - cond->signal (); - - if (this->orb_core_->leader_follower_lock ().release () == -1) - ACE_ERROR_RETURN ((LM_ERROR, - "(%P|%t) TAO_UIOP_Client_Connection_Handler::handle_input: " - "Failed to release the lock.\n"), - -1); - return 0; - } -} - -int -TAO_MT_UIOP_Client_Connection_Handler::resume_handler (ACE_Reactor *reactor) -{ - return reactor->resume_handler (this); -} - -// **************************************************************** - #define TAO_UIOP_SVC_TUPLE ACE_Svc_Tuple<TAO_UIOP_Client_Connection_Handler> #define UIOP_CACHED_CONNECT_STRATEGY ACE_Cached_Connect_Strategy<TAO_UIOP_Client_Connection_Handler, ACE_LSOCK_CONNECTOR, TAO_Cached_Connector_Lock> #define UIOP_REFCOUNTED_HASH_RECYCLABLE_ADDR ACE_Refcounted_Hash_Recyclable<ACE_UNIX_Addr> |