Mon Aug 11 17:56:15 UTC 2008 Johnny Willemsen <jwillemsen@remedy.nl>

16 files changed, 172 insertions, 75 deletions

diff --git a/TAO/ChangeLog b/TAO/ChangeLog
index 64ed95ea7f4..c88b73986b8 100644
--- a/TAO/ChangeLog
+++ b/TAO/ChangeLog
@@ -1,3 +1,37 @@
+Mon Aug 11 17:56:15 UTC 2008  Johnny Willemsen  <jwillemsen@remedy.nl>
+
+          Some changes related to AMI timeouts. The tms stores the relationship
+          between the request id and the reply dispatcher. For AMI timeouts the
+          Asynch_Timeout_Handler did the same, so reworked that. The ATH now just
+          calls a method on the TMS, then also we can callback to the rd with the
+          tms lock hold just as we do for regular replies.
+
+        * tao/Asynch_Reply_Dispatcher_Base.cpp:
+        * tao/Asynch_Reply_Dispatcher_Base.h:
+          No need to deliver an implementation for pure virtual methods
+
+        * tao/Exclusive_TMS.cpp:
+        * tao/Exclusive_TMS.h:
+        * tao/Muxed_TMS.cpp:
+        * tao/Muxed_TMS.h:
+        * tao/Transport_Mux_Strategy.h:
+          Added reply_timed_out method
+
+        * tao/Messaging/Asynch_Reply_Dispatcher.cpp:
+          Make sure we always cancel the timeout handler
+
+        * tao/Messaging/Asynch_Timeout_Handler.cpp:
+        * tao/Messaging/Asynch_Timeout_Handler.h:
+          No need to store the reply_dispatcher, it is stored in the tms
+
+        * tao/Reply_Dispatcher.h:
+        * tao/Synch_Reply_Dispatcher.cpp
+        * tao/Synch_Reply_Dispatcher.h
+          Added reply_timed_out method
+
+        * tao/ORB.cpp:
+          Layout change
+
 Mon Aug 11 16:23:52 UTC 2008  Adam Mitz  <mitza@ociweb.com>
 
         * MPC/config/bidir_giop.mpb:
diff --git a/TAO/tao/Asynch_Reply_Dispatcher_Base.cpp b/TAO/tao/Asynch_Reply_Dispatcher_Base.cpp
index ae6ea86cfca..f2d00f91526 100644
--- a/TAO/tao/Asynch_Reply_Dispatcher_Base.cpp
+++ b/TAO/tao/Asynch_Reply_Dispatcher_Base.cpp
@@ -67,24 +67,6 @@ TAO_Asynch_Reply_Dispatcher_Base::transport (TAO_Transport *t)
   this->transport_->add_reference ();
 }
 
-// Must override pure virtual method in TAO_Reply_Dispatcher.
-int
-TAO_Asynch_Reply_Dispatcher_Base::dispatch_reply (
-    TAO_Pluggable_Reply_Params & /*params*/)
-{
-  return 0;
-}
-
-void
-TAO_Asynch_Reply_Dispatcher_Base::connection_closed (void)
-{
-}
-
-void
-TAO_Asynch_Reply_Dispatcher_Base::reply_timed_out (void)
-{
-}
-
 void
 TAO_Asynch_Reply_Dispatcher_Base::incr_refcount (void)
 {
diff --git a/TAO/tao/Asynch_Reply_Dispatcher_Base.h b/TAO/tao/Asynch_Reply_Dispatcher_Base.h
index 270db1d6067..c6a8d635d6e 100644
--- a/TAO/tao/Asynch_Reply_Dispatcher_Base.h
+++ b/TAO/tao/Asynch_Reply_Dispatcher_Base.h
@@ -53,16 +53,6 @@ public:
   /// Sets the transport for this invocation.
   void transport (TAO_Transport *t);
 
-  /// @name The Reply Dispatcher methods
-  //@{
-  virtual int dispatch_reply (TAO_Pluggable_Reply_Params &) = 0;
-
-  virtual void connection_closed (void) = 0;
-  //@}
-
-  /// Inform that the reply timed out
-  virtual void reply_timed_out (void) = 0;
-
   /// Install the timeout handler
   virtual long schedule_timer (CORBA::ULong , const ACE_Time_Value &) = 0;
 
diff --git a/TAO/tao/Exclusive_TMS.cpp b/TAO/tao/Exclusive_TMS.cpp
index 8a3c303802d..4305d7ed79b 100644
--- a/TAO/tao/Exclusive_TMS.cpp
+++ b/TAO/tao/Exclusive_TMS.cpp
@@ -103,6 +103,32 @@ TAO_Exclusive_TMS::dispatch_reply (TAO_Pluggable_Reply_Params &params)
   return rd->dispatch_reply (params);
 }
 
+int
+TAO_Exclusive_TMS::reply_timed_out (CORBA::ULong request_id)
+{
+  // Check the ids.
+  if (!this->has_request_ || this->request_id_ != request_id)
+    {
+      if (TAO_debug_level > 0)
+        ACE_DEBUG ((LM_DEBUG,
+                    ACE_TEXT ("(%P|%t) TAO_Exclusive_TMS::reply_timed_out - <%d != %d>\n"),
+                    this->request_id_, request_id));
+
+      // The return value 0 informs the transport that the mux strategy
+      // did not find the right reply handler.
+      return 0;
+    }
+
+  TAO_Reply_Dispatcher *rd = this->rd_;
+  this->has_request_ = false;
+  this->request_id_ = 0; // @@ What is a good value???
+  this->rd_ = 0;
+
+  rd->reply_timed_out ();
+
+  return 0;
+}
+
 bool
 TAO_Exclusive_TMS::idle_after_send (void)
 {
diff --git a/TAO/tao/Exclusive_TMS.h b/TAO/tao/Exclusive_TMS.h
index 6d244b861fa..fd2d9a23178 100644
--- a/TAO/tao/Exclusive_TMS.h
+++ b/TAO/tao/Exclusive_TMS.h
@@ -59,6 +59,7 @@ public:
   virtual int unbind_dispatcher (CORBA::ULong request_id);
 
   virtual int dispatch_reply (TAO_Pluggable_Reply_Params &params);
+  virtual int reply_timed_out (CORBA::ULong request_id);
 
   virtual bool idle_after_send (void);
   virtual bool idle_after_reply (void);
diff --git a/TAO/tao/Messaging/Asynch_Invocation_Adapter.cpp b/TAO/tao/Messaging/Asynch_Invocation_Adapter.cpp
index 71e6f8c1b4f..920493700b0 100644
--- a/TAO/tao/Messaging/Asynch_Invocation_Adapter.cpp
+++ b/TAO/tao/Messaging/Asynch_Invocation_Adapter.cpp
@@ -68,7 +68,6 @@ namespace TAO
         // New reply dispatcher on the heap or allocator, because
         // we will go out of scope and hand over the reply dispatcher
         // to the ORB.
-
         TAO_Asynch_Reply_Dispatcher *rd = 0;
 
         // Get the allocator we could use.
diff --git a/TAO/tao/Messaging/Asynch_Reply_Dispatcher.cpp b/TAO/tao/Messaging/Asynch_Reply_Dispatcher.cpp
index d17bdcb6289..2fbb376e27c 100644
--- a/TAO/tao/Messaging/Asynch_Reply_Dispatcher.cpp
+++ b/TAO/tao/Messaging/Asynch_Reply_Dispatcher.cpp
@@ -35,12 +35,6 @@ TAO_Asynch_Reply_Dispatcher::~TAO_Asynch_Reply_Dispatcher (void)
 int
 TAO_Asynch_Reply_Dispatcher::dispatch_reply (TAO_Pluggable_Reply_Params &params)
 {
-  if (params.input_cdr_ == 0)
-    return -1;
-
-  if (!this->try_dispatch_reply ())
-    return 0;
-
   if (this->timeout_handler_)
     {
       // If we had registered timeout handlers just cancel them and
@@ -51,6 +45,12 @@ TAO_Asynch_Reply_Dispatcher::dispatch_reply (TAO_Pluggable_Reply_Params &params)
       // AMI Timeout Handling End
     }
 
+  if (!params.input_cdr_)
+    return -1;
+
+  if (!this->try_dispatch_reply ())
+    return 0;
+
   this->reply_status_ = params.reply_status ();
   this->locate_reply_status_ = params.locate_reply_status ();
 
@@ -148,9 +148,6 @@ TAO_Asynch_Reply_Dispatcher::connection_closed (void)
 {
   try
     {
-      if (!this->try_dispatch_reply ())
-        return;
-
       if (this->timeout_handler_)
         {
           // If we had registered timeout handlers just cancel them and
@@ -160,6 +157,9 @@ TAO_Asynch_Reply_Dispatcher::connection_closed (void)
           this->timeout_handler_ = 0;
         }
 
+      if (!this->try_dispatch_reply ())
+        return;
+
       if (!CORBA::is_nil (this->reply_handler_.in ()))
         {
           // Generate a fake exception....
@@ -197,12 +197,6 @@ TAO_Asynch_Reply_Dispatcher::reply_timed_out (void)
 {
   try
     {
-      // This is okay here... Everything relies on our refcount being
-      // held by the timeout handler, whose refcount in turn is held
-      // by the reactor.
-      if (!this->try_dispatch_reply ())
-        return;
-
       // @@ This check probably is unnecessary..
       if (this->timeout_handler_)
         {
@@ -212,6 +206,12 @@ TAO_Asynch_Reply_Dispatcher::reply_timed_out (void)
           this->timeout_handler_ = 0;
         }
 
+      // This is okay here... Everything relies on our refcount being
+      // held by the timeout handler, whose refcount in turn is held
+      // by the reactor.
+      if (!this->try_dispatch_reply ())
+        return;
+
       if (!CORBA::is_nil (this->reply_handler_.in ()))
         {
           // Generate a fake exception....
@@ -239,7 +239,6 @@ TAO_Asynch_Reply_Dispatcher::reply_timed_out (void)
         {
           ex._tao_print_exception ("Asynch_Reply_Dispacher::reply_timed_out");
         }
-
     }
 
   (void) this->decr_refcount ();
@@ -254,7 +253,6 @@ TAO_Asynch_Reply_Dispatcher::schedule_timer (CORBA::ULong request_id,
       // @@ Need to use the pool for this..
       ACE_NEW_THROW_EX (this->timeout_handler_,
                         TAO_Asynch_Timeout_Handler (
-                          this,
                           this->transport_->orb_core ()->reactor ()),
                         CORBA::NO_MEMORY ());
     }
diff --git a/TAO/tao/Messaging/Asynch_Timeout_Handler.cpp b/TAO/tao/Messaging/Asynch_Timeout_Handler.cpp
index 8aef50719e1..7b66babda99 100644
--- a/TAO/tao/Messaging/Asynch_Timeout_Handler.cpp
+++ b/TAO/tao/Messaging/Asynch_Timeout_Handler.cpp
@@ -14,25 +14,18 @@ ACE_RCSID (Messaging,
 TAO_BEGIN_VERSIONED_NAMESPACE_DECL
 
 TAO_Asynch_Timeout_Handler::TAO_Asynch_Timeout_Handler (
-  TAO_Asynch_Reply_Dispatcher_Base *rd,
   ACE_Reactor *reactor)
-  : rd_ (rd),
-    tms_ (0),
+  : tms_ (0),
     request_id_ (0),
     reactor_ (reactor)
 {
   // Enable reference counting on the event handler.
   this->reference_counting_policy ().value (
     ACE_Event_Handler::Reference_Counting_Policy::ENABLED);
-
-  // We own a reference
-  (void) this->rd_->incr_refcount ();
 }
 
 TAO_Asynch_Timeout_Handler::~TAO_Asynch_Timeout_Handler ()
 {
-  // Forget rd's reference
-  (void) this->rd_->decr_refcount ();
 }
 
 
@@ -54,17 +47,30 @@ int
 TAO_Asynch_Timeout_Handler::handle_timeout (const ACE_Time_Value &,
                                             const void *)
 {
-  if (TAO_debug_level >= 4)
+  TAO_Reply_Dispatcher* rb_base = 0;
+
+  // Check if there was a reply dispatcher registered in the tms, if not
+  // the reply already got dispatched by another thread
+  if (this->tms_->reply_timed_out (this->request_id_) == 0)
     {
-      ACE_DEBUG ((LM_DEBUG,
-                  ACE_TEXT ("TAO_Messaging (%P|%t) - Asynch_Timeout_Handler")
-                  ACE_TEXT ("::handle_timeout, request %d timed out\n"),
-                            this->request_id_));
+      if (TAO_debug_level >= 4)
+        {
+          ACE_DEBUG ((LM_DEBUG,
+                      ACE_TEXT ("TAO_Messaging (%P|%t) - Asynch_Timeout_Handler")
+                      ACE_TEXT ("::handle_timeout, request %d timed out\n"),
+                                this->request_id_));
+        }
+    }
+  else
+    {
+      if (TAO_debug_level >= 1)
+        {
+          ACE_ERROR ((LM_ERROR,
+                      ACE_TEXT ("TAO_Messaging (%P|%t) - Asynch_Timeout_Handler")
+                      ACE_TEXT ("::handle_timeout, unable to dispatch timed out request %d\n"),
+                                this->request_id_));
+        }
     }
-
-  this->tms_->unbind_dispatcher (this->request_id_);
-
-  this->rd_->reply_timed_out ();
 
   // reset any possible timeout errno
   errno = 0;
diff --git a/TAO/tao/Messaging/Asynch_Timeout_Handler.h b/TAO/tao/Messaging/Asynch_Timeout_Handler.h
index 9b8a55eb8c2..8283bdca490 100644
--- a/TAO/tao/Messaging/Asynch_Timeout_Handler.h
+++ b/TAO/tao/Messaging/Asynch_Timeout_Handler.h
@@ -32,7 +32,6 @@
 
 TAO_BEGIN_VERSIONED_NAMESPACE_DECL
 
-class TAO_Asynch_Reply_Dispatcher_Base;
 class TAO_Transport_Mux_Strategy;
 
 /**
@@ -44,8 +43,7 @@ class TAO_Asynch_Timeout_Handler
   : public ACE_Event_Handler
 {
 public:
-  TAO_Asynch_Timeout_Handler (TAO_Asynch_Reply_Dispatcher_Base *rd,
-                              ACE_Reactor *reactor);
+  TAO_Asynch_Timeout_Handler (ACE_Reactor *reactor);
 
   ~TAO_Asynch_Timeout_Handler ();
 
@@ -62,9 +60,6 @@ public:
   virtual void cancel ();
 
 public:
-  /// The reply dispatcher expecting the reply
-  TAO_Asynch_Reply_Dispatcher_Base *rd_;
-
   /// The transport mux strategy dispatching the reply
   TAO_Transport_Mux_Strategy *tms_;
 
diff --git a/TAO/tao/Muxed_TMS.cpp b/TAO/tao/Muxed_TMS.cpp
index 0d2de607685..01415d363d9 100644
--- a/TAO/tao/Muxed_TMS.cpp
+++ b/TAO/tao/Muxed_TMS.cpp
@@ -162,6 +162,57 @@ TAO_Muxed_TMS::dispatch_reply (TAO_Pluggable_Reply_Params &params)
   return result;
 }
 
+int
+TAO_Muxed_TMS::reply_timed_out (CORBA::ULong request_id)
+{
+  int result = 0;
+  // Grab the reply dispatcher for this id.
+  {
+    ACE_GUARD_RETURN (ACE_Lock,
+                      ace_mon,
+                      *this->lock_,
+                      -1);
+
+    TAO_Reply_Dispatcher *rd = 0;
+    result = this->dispatcher_table_.unbind (request_id, rd);
+
+    if (result == 0 && rd)
+      {
+        if (TAO_debug_level > 8)
+          ACE_DEBUG ((LM_DEBUG,
+                      ACE_TEXT ("TAO (%P|%t) - TAO_Muxed_TMS::reply_timed_out, ")
+                      ACE_TEXT ("id = %d\n"),
+                      request_id));
+
+        // Do not move it outside the scope of the lock. A follower thread
+        // could have timedout unwinding the stack and the reply
+        // dispatcher, and that would mean the present thread could be left
+        // with a dangling pointer and may crash. To safeguard againt such
+        // cases we dispatch with the lock held.
+        // Dispatch the reply.
+        rd->reply_timed_out ();
+      }
+    else
+      {
+        if (TAO_debug_level > 0)
+          ACE_DEBUG ((LM_DEBUG,
+                      ACE_TEXT ("TAO (%P|%t) - TAO_Muxed_TMS::reply_timed_out, ")
+                      ACE_TEXT ("unbind dispatcher failed, id %d: result = %d\n"),
+                      request_id,
+                      result));
+
+        // Result = 0 means that the mux strategy was not able
+        // to find a registered reply handler, either because the reply
+        // was not our reply - just forget about it - or it was ours, but
+        // the reply timed out - just forget about the reply.
+        result = 0;
+      }
+  }
+
+  return result;
+}
+
+
 bool
 TAO_Muxed_TMS::idle_after_send (void)
 {
@@ -210,7 +261,9 @@ TAO_Muxed_TMS::clear_cache_i (void)
          this->dispatcher_table_.begin ();
        i != end;
        ++i)
+    {
       ubs.push ((*i).int_id_);
+    }
 
   this->dispatcher_table_.unbind_all ();
   size_t const sz = ubs.size ();
diff --git a/TAO/tao/Muxed_TMS.h b/TAO/tao/Muxed_TMS.h
index aa7c79e401a..ac0864a6510 100644
--- a/TAO/tao/Muxed_TMS.h
+++ b/TAO/tao/Muxed_TMS.h
@@ -63,6 +63,7 @@ public:
   virtual int unbind_dispatcher (CORBA::ULong request_id);
 
   virtual int dispatch_reply (TAO_Pluggable_Reply_Params &params);
+  virtual int reply_timed_out (CORBA::ULong request_id);
 
   virtual bool idle_after_send (void);
   virtual bool idle_after_reply (void);
diff --git a/TAO/tao/ORB.cpp b/TAO/tao/ORB.cpp
index 416aedf9f81..106186495e7 100644
--- a/TAO/tao/ORB.cpp
+++ b/TAO/tao/ORB.cpp
@@ -1,4 +1,4 @@
-  // "$Id$"
+// "$Id$"
 
 #include "tao/ORB.h"
 
@@ -927,8 +927,7 @@ CORBA::ORB::resolve_initial_references (const char *name,
       // Search the object reference table.  This search must occur before
       // the InitRef table search, since it may contain local objects.
       result =
-        this->orb_core ()->object_ref_table ().resolve_initial_reference (
-          name);
+        this->orb_core ()->object_ref_table ().resolve_initial_reference (name);
     }
 
   if (!CORBA::is_nil (result.in ()))
diff --git a/TAO/tao/Reply_Dispatcher.h b/TAO/tao/Reply_Dispatcher.h
index 7ce7fd4903e..e291148edfe 100644
--- a/TAO/tao/Reply_Dispatcher.h
+++ b/TAO/tao/Reply_Dispatcher.h
@@ -68,6 +68,9 @@ public:
    */
   virtual int dispatch_reply (TAO_Pluggable_Reply_Params &params) = 0;
 
+  /// Inform that the reply timed out
+  virtual void reply_timed_out (void) = 0;
+
   /**
    * The used for the pending reply has been closed.
    * No reply is expected.
diff --git a/TAO/tao/Synch_Reply_Dispatcher.cpp b/TAO/tao/Synch_Reply_Dispatcher.cpp
index 571239f2e05..34539ca3581 100644
--- a/TAO/tao/Synch_Reply_Dispatcher.cpp
+++ b/TAO/tao/Synch_Reply_Dispatcher.cpp
@@ -48,6 +48,12 @@ TAO_Synch_Reply_Dispatcher::reply_cdr (void)
   return this->reply_cdr_;
 }
 
+void
+TAO_Synch_Reply_Dispatcher::reply_timed_out (void)
+{
+  // noop
+}
+
 int
 TAO_Synch_Reply_Dispatcher::dispatch_reply (
     TAO_Pluggable_Reply_Params &params)
diff --git a/TAO/tao/Synch_Reply_Dispatcher.h b/TAO/tao/Synch_Reply_Dispatcher.h
index c22e515cd47..1a464fa7559 100644
--- a/TAO/tao/Synch_Reply_Dispatcher.h
+++ b/TAO/tao/Synch_Reply_Dispatcher.h
@@ -61,6 +61,8 @@ public:
 
   virtual void connection_closed (void);
 
+  virtual void reply_timed_out (void);
+
 protected:
   /// The service context list
   IOP::ServiceContextList &reply_service_info_;
diff --git a/TAO/tao/Transport_Mux_Strategy.h b/TAO/tao/Transport_Mux_Strategy.h
index 6f729a1db23..3fa8ba4b5dc 100644
--- a/TAO/tao/Transport_Mux_Strategy.h
+++ b/TAO/tao/Transport_Mux_Strategy.h
@@ -55,9 +55,6 @@ public:
   /// invocation.
   virtual CORBA::ULong request_id (void) = 0;
 
-  // = Bind and Find methods for the <Request ID, ReplyDispatcher>
-  //   pairs.
-
   /// Bind the dispatcher with the request id. Commonalities in the
   /// derived class implementations is kept here.
   virtual int bind_dispatcher (CORBA::ULong request_id,
@@ -72,6 +69,11 @@ public:
    */
   virtual int unbind_dispatcher (CORBA::ULong request_id) = 0;
 
+  /**
+   * Dispatch a reply timeout for request @a request_id
+   **/
+  virtual int reply_timed_out (CORBA::ULong request_id) = 0;
+
   /// Dispatch the reply for <request_id>, cleanup any resources
   /// allocated for that request.
   virtual int dispatch_reply (TAO_Pluggable_Reply_Params &params) = 0;