All the action for a synchronous twoway invocation happen here. More...
#include <Synch_Invocation.h>
Public Member Functions | |
| Synch_Twoway_Invocation (CORBA::Object_ptr otarget, Profile_Transport_Resolver &resolver, TAO_Operation_Details &detail, bool response_expected=true) | |
| Constructor used by TAO::Invocation_Adapter. | |
| Invocation_Status | remote_twoway (ACE_Time_Value *max_wait_time) |
Protected Member Functions | |
| virtual Invocation_Status | handle_user_exception (TAO_InputCDR &cdr) |
| Invocation_Status | location_forward (TAO_InputCDR &cdr) |
| Helper method used to handle location forwarded replies. | |
| Invocation_Status | handle_system_exception (TAO_InputCDR &cdr) |
| Invocation_Status | wait_for_reply (ACE_Time_Value *max_wait_time, TAO_Synch_Reply_Dispatcher &rd, TAO_Bind_Dispatcher_Guard &bd) |
| As the name suggests waits for a reply from the remote ORB. | |
Private Member Functions | |
| Invocation_Status | check_reply_status (TAO_Synch_Reply_Dispatcher &rd) |
All the action for a synchronous twoway invocation happen here.
An object of this type is created by TAO::Invocation_Adapter and invokes a method on this class. The method takes care of creating and sending a request, waiting for a reply and demarshalling the reply for the client.
Definition at line 52 of file Synch_Invocation.h.
| TAO::Synch_Twoway_Invocation::Synch_Twoway_Invocation | ( | CORBA::Object_ptr | otarget, | |
| Profile_Transport_Resolver & | resolver, | |||
| TAO_Operation_Details & | detail, | |||
| bool | response_expected = true | |||
| ) |
Constructor used by TAO::Invocation_Adapter.
| otarget | The original target on which this invocation was started. This is there to be passed up to its parent class. | |
| resolver | The profile and transport holder. | |
| detail | Operation details of the invocation on the target | |
| response_expected | Flag to indicate whether the operation encapsulated by op returns a response or not. |
Definition at line 42 of file Synch_Invocation.cpp.
: Remote_Invocation (otarget, resolver, detail, response_expected) { }
| Invocation_Status TAO::Synch_Twoway_Invocation::check_reply_status | ( | TAO_Synch_Reply_Dispatcher & | rd | ) | [private] |
Helper method that checks the reply status of the replies and takes appropriate action. This method returns an exception when there is an error.
Definition at line 341 of file Synch_Invocation.cpp.
{
/*
* Precondition: We probably got a reply. <ACE_Thread::self> is
* checking the status of the reply
*
* Postcondition: Any error while reading the reply is marked by
* raising an exception. LOCATION_FORWARDED replies are marked by
* returning a restart since that is what needed to be done by the
* callee.
*/
TAO_InputCDR &cdr = rd.reply_cdr ();
// Set the translators
this->resolver_.transport ()->assign_translators (&cdr, 0);
// At this point it can be assumed that the GIOP/whatever protocol
// header and the reply header are already handled. Further it
// can be assumed that the reply body contains the details
// required for further processing. All the other details should
// have been handled in the reply dispatcher/protocol framework.
switch (rd.reply_status ())
{
case GIOP::NO_EXCEPTION:
{
Reply_Guard mon (this, TAO_INVOKE_FAILURE);
if (this->details_.demarshal_args (cdr) == false)
{
throw ::CORBA::MARSHAL ();
}
mon.set_status (TAO_INVOKE_SUCCESS);
}
break;
case GIOP::LOCATION_FORWARD:
return this->location_forward (cdr);
case GIOP::LOCATION_FORWARD_PERM:
{
// Unmarshal the location forward object and set the
// variable this->forward_to_.
Invocation_Status const s = this->location_forward (cdr);
if (s != TAO_INVOKE_FAILURE)
{
// de-marshalling of permanent object reference was successfull
CORBA::Boolean const permanent_forward_condition =
this->stub ()->orb_core ()->is_permanent_forward_condition
(this->forwarded_to_.in (),
this->request_service_context ());
if (!permanent_forward_condition)
{
// permanent condition not given
if (TAO_debug_level > 3)
ACE_DEBUG ((LM_DEBUG,
"TAO (%P|%t) - Synch_Twoway_Invocation::"
"check_reply_status: unexpected LOCATION_FORWARD_PERM reply\n"));
throw ::CORBA::INTERNAL (0, CORBA::COMPLETED_NO);
}
// This is the only place where we set in permanent forward state.
this->reply_status (GIOP::LOCATION_FORWARD_PERM);
}
return s;
}
case GIOP::USER_EXCEPTION:
return this->handle_user_exception (cdr);
case GIOP::SYSTEM_EXCEPTION:
return this->handle_system_exception (cdr);
case GIOP::NEEDS_ADDRESSING_MODE:
{
Reply_Guard mon (this, TAO_INVOKE_FAILURE);
// We have received a message with a request to change the
// addressing mode. First let us read the mode that the
// server/agent asks for.
CORBA::Short addr_mode = 0;
if (cdr.read_short (addr_mode) == 0)
{
// Could not demarshal the addressing disposition, raise an local
// CORBA::MARSHAL
throw ::CORBA::MARSHAL (0, CORBA::COMPLETED_MAYBE);
}
// Now set this addressing mode in the profile, so that
// the next invocation need not go through this.
this->resolver_.profile ()->addressing_mode (addr_mode);
mon.set_status (TAO_INVOKE_RESTART);
// Now restart the invocation
return TAO_INVOKE_RESTART;
}
}
return TAO_INVOKE_SUCCESS;
}
| Invocation_Status TAO::Synch_Twoway_Invocation::handle_system_exception | ( | TAO_InputCDR & | cdr | ) | [protected] |
Helper method used to handle system exceptions from the remote objects.
There has been a unanimous view that this is not the right way to do things. But a need to be compliant is forcing us into this.
Definition at line 516 of file Synch_Invocation.cpp.
{
Reply_Guard mon (this, TAO_INVOKE_FAILURE);
if (TAO_debug_level > 3)
ACE_DEBUG ((LM_DEBUG,
"TAO (%P|%t) - Synch_Twoway_Invocation::"
"handle_system_exception\n"));
CORBA::String_var type_id;
if (!(cdr >> type_id.inout ()))
{
// Could not demarshal the exception id, raise an local
// CORBA::MARSHAL
throw ::CORBA::MARSHAL (0, CORBA::COMPLETED_MAYBE);
}
CORBA::ULong minor = 0;
CORBA::ULong completion = 0;
if (!(cdr >> minor) || !(cdr >> completion))
{
throw ::CORBA::MARSHAL (0, CORBA::COMPLETED_MAYBE);
}
bool do_forward = false;
int foe_kind = this->stub ()->orb_core ()->orb_params ()->forward_once_exception();
if ((CORBA::CompletionStatus) completion != CORBA::COMPLETED_YES
&& (((foe_kind & TAO::FOE_TRANSIENT) == 0
&& ACE_OS_String::strcmp (type_id.in (),
"IDL:omg.org/CORBA/TRANSIENT:1.0") == 0) ||
ACE_OS_String::strcmp (type_id.in (),
"IDL:omg.org/CORBA/OBJ_ADAPTER:1.0") == 0 ||
ACE_OS_String::strcmp (type_id.in (),
"IDL:omg.org/CORBA/NO_RESPONSE:1.0") == 0 ||
((foe_kind & TAO::FOE_COMM_FAILURE) == 0
&& ACE_OS_String::strcmp (type_id.in (),
"IDL:omg.org/CORBA/COMM_FAILURE:1.0") == 0) ||
(this->stub ()->orb_core ()->orb_params ()->forward_invocation_on_object_not_exist ()
&& ACE_OS_String::strcmp (type_id.in (),
"IDL:omg.org/CORBA/OBJECT_NOT_EXIST:1.0") == 0) ||
(do_forward = ! this->stub ()->forwarded_on_exception ()
&& ((((foe_kind & TAO::FOE_OBJECT_NOT_EXIST) == TAO::FOE_OBJECT_NOT_EXIST)
&& (ACE_OS_String::strcmp (type_id.in (),
"IDL:omg.org/CORBA/OBJECT_NOT_EXIST:1.0") == 0)) ||
(((foe_kind & TAO::FOE_COMM_FAILURE) == TAO::FOE_COMM_FAILURE)
&& (ACE_OS_String::strcmp (type_id.in (),
"IDL:omg.org/CORBA/COMM_FAILURE:1.0") == 0)) ||
(((foe_kind & TAO::FOE_TRANSIENT) == TAO::FOE_TRANSIENT)
&& (ACE_OS_String::strcmp (type_id.in (),
"IDL:omg.org/CORBA/TRANSIENT:1.0") == 0)) ||
(((foe_kind & TAO::FOE_INV_OBJREF) == TAO::FOE_INV_OBJREF)
&& (ACE_OS_String::strcmp (type_id.in (),
"IDL:omg.org/CORBA/INV_OBJREF:1.0") == 0))))))
{
if (do_forward)
this->stub ()->forwarded_on_exception (true);
// Start the special case for FTCORBA.
/**
* There has been a unanimous view that this is not the
* right way to do things. But a need to be compliant is
* forcing us into this.
*/
Invocation_Status const s =
this->stub ()->orb_core ()->service_raise_transient_failure (
this->details_.request_service_context ().service_info (),
this->resolver_.profile ());
if (s == TAO_INVOKE_RESTART)
return s;
// Attempt profile retry.
/**
* @note A location forwarding loop may occur where a client
* is bounced from the original target to the forwarded
* target and back if the application is not equipped to
* handle retries of previously called targets. TAO may
* be able to help in this case but it ultimately ends
* up being an application issue.
*/
if (completion != CORBA::COMPLETED_MAYBE &&
this->resolver_.stub ()->next_profile_retry ())
{
return TAO_INVOKE_RESTART;
}
// Fall through and raise an exception.
}
CORBA::SystemException *ex = TAO::create_system_exception (type_id.in ());
if (ex == 0)
{
// @@ We should raise a CORBA::NO_MEMORY, but we ran out
// of memory already. We need a pre-allocated, TSS,
// CORBA::NO_MEMORY instance
ACE_NEW_RETURN (ex,
CORBA::UNKNOWN,
TAO_INVOKE_FAILURE);
}
// Without this, the call to create_system_exception() above
// causes a memory leak. On platforms without native exceptions,
// the CORBA::Environment class manages the memory.
auto_ptr<CORBA::SystemException> safety (ex);
ex->minor (minor);
ex->completed (CORBA::CompletionStatus (completion));
if (TAO_debug_level > 4)
ACE_DEBUG ((LM_DEBUG,
"TAO (%P|%t) - Synch_Twoway_Invocation::"
"handle_system_exception, about to raise\n"));
mon.set_status (TAO_INVOKE_SYSTEM_EXCEPTION);
// Raise the exception.
ex->_raise ();
return TAO_INVOKE_SYSTEM_EXCEPTION;
}
| Invocation_Status TAO::Synch_Twoway_Invocation::handle_user_exception | ( | TAO_InputCDR & | cdr | ) | [protected, virtual] |
This method is selectively made virtual, so that inherited classes can overload the user exception handling type. For example the DII needs a totally different method of user exception exception handling
Reimplemented in TAO::DII_Invocation.
Definition at line 471 of file Synch_Invocation.cpp.
{
Reply_Guard mon (this,
TAO_INVOKE_FAILURE);
if (TAO_debug_level > 3)
ACE_DEBUG ((LM_DEBUG,
"TAO (%P|%t) - Synch_Twoway_Invocation::"
"handle_user_exception\n"));
// Pull the exception from the stream.
CORBA::String_var buf;
if (!(cdr >> buf.inout ()))
{
// Could not demarshal the exception id, raise an local
// CORBA::MARSHAL
throw ::CORBA::MARSHAL (0, CORBA::COMPLETED_MAYBE);
}
CORBA::Exception *exception = this->details_.corba_exception (buf.in ());
exception->_tao_decode (cdr);
if (TAO_debug_level > 5)
{
ACE_DEBUG ((LM_DEBUG,
ACE_TEXT ("TAO (%P|%t) - Synch_Twoway_Invocation::")
ACE_TEXT ("handle_user_exception - ")
ACE_TEXT ("raising exception %C\n"),
buf.in ()));
}
mon.set_status (TAO_INVOKE_USER_EXCEPTION);
// We must manage the memory allocated
// by the call above to alloc().
auto_ptr<CORBA::Exception> safety (exception);
exception->_raise ();
return TAO_INVOKE_USER_EXCEPTION;
}
| Invocation_Status TAO::Synch_Twoway_Invocation::location_forward | ( | TAO_InputCDR & | cdr | ) | [protected] |
Helper method used to handle location forwarded replies.
Definition at line 441 of file Synch_Invocation.cpp.
{
Reply_Guard mon (this, TAO_INVOKE_FAILURE);
if (TAO_debug_level > 3)
{
ACE_DEBUG ((LM_DEBUG,
ACE_TEXT ("TAO (%P|%t) - Synch_Twoway_Invocation::location_forward ")
ACE_TEXT ("being handled\n")));
}
CORBA::Object_var fwd;
if (!(inp_stream >> fwd))
{
throw ::CORBA::MARSHAL (
CORBA::SystemException::_tao_minor_code (
TAO_INVOCATION_LOCATION_FORWARD_MINOR_CODE,
errno),
CORBA::COMPLETED_NO);
}
this->forwarded_reference (fwd.in ());
mon.set_status (TAO_INVOKE_RESTART);
return TAO_INVOKE_RESTART;
}
| Invocation_Status TAO::Synch_Twoway_Invocation::remote_twoway | ( | ACE_Time_Value * | max_wait_time | ) |
Method used by the adapter to kickstart an invocation to the remote object. There is a exception declaration in this method which ensures that the exceptions propogated by the remote objects are converted a CORBA exceptions for the clients. This method does a bunch of things necessary to create and send the invocation. This method is also nerve centre for the interceptor invocation points.
Definition at line 55 of file Synch_Invocation.cpp.
{
ACE_Countdown_Time countdown (max_wait_time);
TAO_Synch_Reply_Dispatcher *rd_p = 0;
ACE_NEW_NORETURN (rd_p, TAO_Synch_Reply_Dispatcher (this->resolver_.stub ()->orb_core (),
this->details_.reply_service_info ()));
if (!rd_p)
{
throw ::CORBA::NO_MEMORY ();
}
ACE_Intrusive_Auto_Ptr<TAO_Synch_Reply_Dispatcher> rd(rd_p, false);
Invocation_Status s = TAO_INVOKE_FAILURE;
#if TAO_HAS_INTERCEPTORS == 1
// Start the interception point here..
s = this->send_request_interception ();
if (s != TAO_INVOKE_SUCCESS)
return s;
// We have started the interception flow. We need to call the
// ending interception flow if things go wrong. The purpose of the
// try block is to do just this.
try
{
#endif /*TAO_HAS_INTERCEPTORS */
TAO_Transport* const transport = this->resolver_.transport ();
if (!transport)
{
// Way back, we failed to find a profile we could connect to.
// We've come this far only so we reach the interception points
// in case they can fix things. Time to bail....
throw CORBA::TRANSIENT (CORBA::OMGVMCID | 2, CORBA::COMPLETED_NO);
}
ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, ace_mon,
transport->output_cdr_lock (), TAO_INVOKE_FAILURE);
TAO_OutputCDR &cdr = transport->out_stream ();
cdr.message_attributes (this->details_.request_id (),
this->resolver_.stub (),
TAO_TWOWAY_REQUEST,
max_wait_time);
this->write_header (cdr);
this->marshal_data (cdr);
// Register a reply dispatcher for this invocation. Use the
// preallocated reply dispatcher.
TAO_Bind_Dispatcher_Guard dispatch_guard (
this->details_.request_id (),
rd.get (),
transport->tms ());
if (dispatch_guard.status () != 0)
{
// @@ What is the right way to handle this error? Why should
// we close the connection?
transport->close_connection ();
throw ::CORBA::INTERNAL (0, CORBA::COMPLETED_NO);
}
countdown.update ();
s = this->send_message (cdr,
TAO_TWOWAY_REQUEST,
max_wait_time);
ace_mon.release();
#if TAO_HAS_INTERCEPTORS == 1
// @@NOTE: Too much code repetition.
// If the above call returns a restart due to connection
// failure then call the receive_other interception point
// before we leave.
if (s == TAO_INVOKE_RESTART)
{
Invocation_Status const tmp = this->receive_other_interception ();
if (tmp != TAO_INVOKE_SUCCESS)
s = tmp;
}
#endif /* TAO_HAS_INTERCEPTORS */
if (s != TAO_INVOKE_SUCCESS)
return s;
countdown.update ();
// For some strategies one may want to release the transport
// back to cache. If the idling is successfull let the
// resolver about that.
if (transport->idle_after_send ())
this->resolver_.transport_released ();
// @@ In all MT environments, there's a cancellation point lurking
// here; need to investigate. Client threads would frequently be
// canceled sometime during recv_request ... the correct action to
// take on being canceled is to issue a CancelRequest message to the
// server and then imediately let other client-side cancellation
// handlers do their jobs.
//
// In C++, that basically means to unwind the stack using almost
// normal procedures: all destructors should fire, and some "catch"
// blocks should probably be able to handle things like releasing
// pointers. (Without unwinding the C++ stack, resources that must
// be freed by thread cancellation won't be freed, and the process
// won't continue to function correctly.) The tricky part is that
// according to POSIX, all C stack frames must also have their
// (explicitly coded) handlers called. We assume a POSIX.1c/C/C++
// environment.
s = this->wait_for_reply (max_wait_time, *rd.get (), dispatch_guard);
#if TAO_HAS_INTERCEPTORS == 1
if (s == TAO_INVOKE_RESTART)
{
Invocation_Status const tmp = this->receive_other_interception ();
// Push the latest values for the return..
if (tmp != TAO_INVOKE_SUCCESS)
s = tmp;
}
#endif /* TAO_HAS_INTERCEPTORS */
if (s != TAO_INVOKE_SUCCESS)
return s;
// What happens when the above call returns an error through
// the return value? That would be bogus as per the contract
// in the interface. The call violated the contract
s = this->check_reply_status (*rd.get ());
// For some strategies one may want to release the transport
// back to cache after receiving the reply.
if (transport->idle_after_reply ())
this->resolver_.transport_released ();
#if TAO_HAS_INTERCEPTORS == 1
Invocation_Status tmp = TAO_INVOKE_FAILURE;
if (s == TAO_INVOKE_RESTART)
{
tmp = this->receive_other_interception ();
}
else if (s == TAO_INVOKE_SUCCESS)
{
tmp = this->receive_reply_interception ();
}
if (tmp != TAO_INVOKE_SUCCESS)
s = tmp;
}
catch ( ::CORBA::Exception& ex)
{
PortableInterceptor::ReplyStatus const status =
this->handle_any_exception (&ex);
if (status == PortableInterceptor::LOCATION_FORWARD ||
status == PortableInterceptor::TRANSPORT_RETRY)
s = TAO_INVOKE_RESTART;
else if (status == PortableInterceptor::SYSTEM_EXCEPTION
|| status == PortableInterceptor::USER_EXCEPTION)
throw;
}
catch (...)
{
// Notify interceptors of non-CORBA exception, and propagate
// that exception to the caller.
PortableInterceptor::ReplyStatus const st =
this->handle_all_exception ();
if (st == PortableInterceptor::LOCATION_FORWARD ||
st == PortableInterceptor::TRANSPORT_RETRY)
s = TAO_INVOKE_RESTART;
else
throw;
}
#endif /* TAO_HAS_INTERCEPTORS */
return s;
}
| Invocation_Status TAO::Synch_Twoway_Invocation::wait_for_reply | ( | ACE_Time_Value * | max_wait_time, | |
| TAO_Synch_Reply_Dispatcher & | rd, | |||
| TAO_Bind_Dispatcher_Guard & | bd | |||
| ) | [protected] |
As the name suggests waits for a reply from the remote ORB.
This method returns an exception when there is an error.
Definition at line 245 of file Synch_Invocation.cpp.
{
/*
* Precondition: The call went to the remote
* peer. <ACE_Thread::self> is waiting for the reply.
*
* Postcondition: Any error during a wait is marked by raising an
* exception. Success alone is returned through the return value.
*/
bool const
expired= (max_wait_time && ACE_Time_Value::zero == *max_wait_time);
if (expired)
errno= ETIME;
int const
reply_error = expired ? -1 :
this->resolver_.transport ()->wait_strategy ()->wait (max_wait_time, rd);
if (TAO_debug_level > 0 && max_wait_time)
{
ACE_DEBUG ((LM_DEBUG,
"TAO (%P|%t) - Synch_Twoway_Invocation::wait_for_reply, "
"timeout after recv is <%u> status <%d>\n",
max_wait_time->msec (),
reply_error));
}
// Check the reply error.
if (reply_error == -1)
{
// Unbind the dispatcher, since its of no use at this point of
// time
if (TAO_debug_level > 3)
{
ACE_DEBUG ((LM_DEBUG,
"TAO (%P|%t) - Synch_Twoway_Invocation::wait_for_reply, "
"recovering after an error\n"));
}
// You the smarty, don't try to moving the unbind_dispatcher
// () call since it looks like it is repeated twice. That
// could land you in trouble. If you don't believe this
// warning go ahead and try. Try running tests to see what is
// going on ;)
if (errno == ETIME)
{
// If the unbind succeeds then thrown an exception to the
// application, else just collect the reply and dispatch
// that to the application.
//
// NOTE: A fragile synchronization is provided when using
// the Muxed Transport strategy. We could infact be a
// follower thread getting timedout in the LF whereas the
// dispatching thread could be on the reply_dispatcher
// that we created. This would lead bad crashes. To get
// around that, the call to unbind_dispatcher () will wait
// on the lock on the Muxed_Transport_Strategy if
// dispatching has started. This is fragile.
if (bd.unbind_dispatcher () == 0)
{
// Just a timeout with completed_maybe, don't close
// the connection or anything
throw ::CORBA::TIMEOUT (
CORBA::SystemException::_tao_minor_code (
TAO_TIMEOUT_RECV_MINOR_CODE,
errno),
CORBA::COMPLETED_MAYBE);
}
}
else
{
(void) bd.unbind_dispatcher ();
this->resolver_.transport ()->close_connection ();
try
{
return
this->stub()->orb_core ()->service_raise_comm_failure (
this->details_.request_service_context ().service_info (),
this->resolver_.profile ());
}
catch (const ::CORBA::Exception&)
{
this->resolver_.stub ()->reset_profiles ();
throw;
}
}
}
return TAO_INVOKE_SUCCESS;
}
1.7.0