ChangeLogTag:Sun Dec 22 11:26:30 2002 Balachandran Natarajan <bala@isis-server.isis.vanderbilt.edu>

14 files changed, 397 insertions, 103 deletions

diff --git a/TAO/ChangeLog b/TAO/ChangeLog
index b63b295a6b3..ca97b683a60 100644
--- a/TAO/ChangeLog
+++ b/TAO/ChangeLog
@@ -1,3 +1,59 @@
+Sun Dec 22 11:26:30 2002  Balachandran Natarajan  <bala@isis-server.isis.vanderbilt.edu>
+
+	Fix for Bug 1276. Please see 
+
+	http://deuce.doc.wustl.edu/bugzilla/show_bug.cgi?id=1276 for more
+        details on this bug. The gist of the fix is this
+
+	  - If a follower thread timesout, it first tries to unbind the
+	    dispatcher from the table
+	 
+	  - If unbinding is succesfull it throws a TIMEOUT exception to
+	    the client. 
+
+	  - If unbinding is unsuccesful, it waits on a condition variable
+	    in the reply dispatcher
+
+	  - Once the reply is available it dispatches the reply. 
+
+	  - This helps avoid the race conditions that are documented in
+	    1276. 
+
+	None of the above code should be executed when timeouts are not
+        enabled in the ORB ie. for the default path.
+	
+	* tao/Bind_Dispatcher_Guard.cpp:
+	* tao/Bind_Dispatcher_Guard.h:
+	* tao/Bind_Dispatcher_Guard.i: Added method unbind_dispatcher ()
+	  which unbinds the dispatcher from the Transport_Mux_Strategy and
+	  status (), which allows to set the status of the dispatcher
+	  guard. 
+
+	* tao/Transport_Mux_Strategy.h
+	* tao/Exclusive_TMS.cpp:
+	* tao/Exclusive_TMS.h:
+	* tao/Muxed_TMS.cpp:
+	* tao/Muxed_TMS.h: Changed the signature of the method
+	  unbind_dispatcher (). It now returns an int. Further, we mark
+	  the Reply_Dispatcher as busy when the dispatching thread is
+	  ready to call Reply_Dispatcher::dispatch_reply ().
+
+	* tao/Reply_Dispatcher.cpp:
+	* tao/Reply_Dispatcher.h:
+	* tao/Reply_Dispatcher.i: Added a mutex and a condition variable
+	  for the follower to wait on when a leader thread is within the
+	  Reply_Dispatcher. Methods to wait on the CV and to mark the 
+	  reply_dispatcher as being used and free have been added. These
+	  methods are essentially no-ops if timeouts are not set.
+	
+	* tao/Invocation.cpp: 
+	* tao/Invocation.h: If the invocation thread gets a timeout, it
+	  first unbinds its reply dispatcher from the Mux_Strategy. If the
+	  unbind fails (this means that another thread has aleardy
+	  collected the reply from the transport) , it waits on the
+	  reply_dispatcher to collect the reply instead of just throwing a
+	  TIMEOUT exception.
+	
 Fri Dec 20 13:47:33 2002  Balachandran Natarajan  <bala@isis-server.isis.vanderbilt.edu>
 
 	* orbsvcs/tests/Security/Big_Request/Makefile:
diff --git a/TAO/tao/Bind_Dispatcher_Guard.cpp b/TAO/tao/Bind_Dispatcher_Guard.cpp
index 38618f89d10..b9f83b726f2 100644
--- a/TAO/tao/Bind_Dispatcher_Guard.cpp
+++ b/TAO/tao/Bind_Dispatcher_Guard.cpp
@@ -1,15 +1,6 @@
 // This may look like C, but it's really -*- C++ -*-
 // $Id$
 
-// ===========================================================================
-//
-// = LIBRARY
-//     TAO
-//
-// = AUTHOR
-//     Chad Elliott <elliott_c@ociweb.com>
-//
-// ===========================================================================
 
 #include "tao/Bind_Dispatcher_Guard.h"
 
@@ -17,9 +8,32 @@
 # include "tao/Bind_Dispatcher_Guard.i"
 #endif /* ! __ACE_INLINE__ */
 
+ACE_RCSID (tao,
+           Bind_Dispatcher_Guard,
+           "$Id$")
+
+TAO_Bind_Dispatcher_Guard::TAO_Bind_Dispatcher_Guard (
+                                         CORBA::ULong request_id,
+                                         TAO_Reply_Dispatcher* rd,
+                                         TAO_Transport_Mux_Strategy* tms)
+ : status_(TAO_Bind_Dispatcher_Guard::UNBIND),
+   request_id_(request_id),
+   rd_(rd),
+   tms_(tms)
+{
+  int retval =
+    this->tms_->bind_dispatcher (this->request_id_,
+                                 this->rd_);
+
+  if (retval == -1 )
+    this->status_ =
+      TAO_Bind_Dispatcher_Guard::NO_UNBIND;
+}
+
 TAO_Bind_Dispatcher_Guard::~TAO_Bind_Dispatcher_Guard (void)
 {
-  if (this->status_ == 0) {
-    this->tms_->unbind_dispatcher (this->request_id_);
-  }
+  // We try unbinding. If it works it works, else cant do much about
+  // it..
+  if (this->status_ == TAO_Bind_Dispatcher_Guard::UNBIND)
+    (void) this->tms_->unbind_dispatcher (this->request_id_);
 }
diff --git a/TAO/tao/Bind_Dispatcher_Guard.h b/TAO/tao/Bind_Dispatcher_Guard.h
index 29b9da86feb..7e17d2e07a3 100644
--- a/TAO/tao/Bind_Dispatcher_Guard.h
+++ b/TAO/tao/Bind_Dispatcher_Guard.h
@@ -30,7 +30,21 @@ public:
 
   ~TAO_Bind_Dispatcher_Guard (void);
 
-  int& status(void);
+  /// Returns 0 if succesful and non-zero otherwise.
+  int unbind_dispatcher (void);
+
+  /// State information
+  enum
+    {
+      /// Unbind the dispatcher
+      UNBIND = 0,
+      /// Do not unbind dispatcher
+      NO_UNBIND
+    };
+
+  /// Set and get status
+  int status (void) const;
+  void status (int status);
 
 private:
   int                         status_;
diff --git a/TAO/tao/Bind_Dispatcher_Guard.i b/TAO/tao/Bind_Dispatcher_Guard.i
index 3c497e3c03f..64bce7bf6f6 100644
--- a/TAO/tao/Bind_Dispatcher_Guard.i
+++ b/TAO/tao/Bind_Dispatcher_Guard.i
@@ -16,22 +16,27 @@
 #include "tao/Transport_Mux_Strategy.h"
 
 
-ACE_INLINE
-TAO_Bind_Dispatcher_Guard::TAO_Bind_Dispatcher_Guard (
-                                         CORBA::ULong request_id,
-                                         TAO_Reply_Dispatcher* rd,
-                                         TAO_Transport_Mux_Strategy* tms)
- : status_(0),
-   request_id_(request_id),
-   rd_(rd),
-   tms_(tms)
+ACE_INLINE void
+TAO_Bind_Dispatcher_Guard::status (int s)
 {
-  this->status_ = this->tms_->bind_dispatcher (this->request_id_,
-                                               this->rd_);
+  this->status_ = s;
 }
 
-ACE_INLINE int &
-TAO_Bind_Dispatcher_Guard::status (void)
+ACE_INLINE int
+TAO_Bind_Dispatcher_Guard::status (void) const
 {
-  return status_;
+  return this->status_;
+}
+
+ACE_INLINE int
+TAO_Bind_Dispatcher_Guard::unbind_dispatcher (void)
+{
+  int retval =
+    this->tms_->unbind_dispatcher (this->request_id_);
+
+  // Already unbound and so do not try again during destruction.
+  this->status_ =
+    TAO_Bind_Dispatcher_Guard::NO_UNBIND;
+
+  return retval;
 }
diff --git a/TAO/tao/Exclusive_TMS.cpp b/TAO/tao/Exclusive_TMS.cpp
index 0f17f8eddca..17fa3990127 100644
--- a/TAO/tao/Exclusive_TMS.cpp
+++ b/TAO/tao/Exclusive_TMS.cpp
@@ -62,14 +62,16 @@ TAO_Exclusive_TMS::bind_dispatcher (CORBA::ULong request_id,
   return 0;
 }
 
-void
+int
 TAO_Exclusive_TMS::unbind_dispatcher (CORBA::ULong request_id)
 {
   if (!this->has_request_ || this->request_id_ != request_id)
-    return;
+    return -1;
   this->has_request_ = 0;
   this->request_id_ = 0;
   this->rd_ = 0;
+
+  return 0;
 }
 
 int
@@ -93,9 +95,17 @@ TAO_Exclusive_TMS::dispatch_reply (TAO_Pluggable_Reply_Params &params)
   this->request_id_ = 0; // @@ What is a good value???
   this->rd_ = 0;
 
+  // Starting dispatch
+  (void) rd->start_dispatch ();
+
   // Dispatch the reply.
   // Returns 1 on success, -1 on failure.
-  return rd->dispatch_reply (params);
+  int retval =
+    rd->dispatch_reply (params);
+
+  (void) rd->end_dispatch ();
+
+  return retval;
 }
 
 int
diff --git a/TAO/tao/Exclusive_TMS.h b/TAO/tao/Exclusive_TMS.h
index 5bd62e6a1b3..9a9094bf1cd 100644
--- a/TAO/tao/Exclusive_TMS.h
+++ b/TAO/tao/Exclusive_TMS.h
@@ -54,7 +54,7 @@ public:
   virtual CORBA::ULong request_id (void);
   virtual int bind_dispatcher (CORBA::ULong request_id,
                                TAO_Reply_Dispatcher *rh);
-  virtual void unbind_dispatcher (CORBA::ULong request_id);
+  virtual int unbind_dispatcher (CORBA::ULong request_id);
 
   virtual int dispatch_reply (TAO_Pluggable_Reply_Params &params);
 
diff --git a/TAO/tao/Invocation.cpp b/TAO/tao/Invocation.cpp
index caeb5237c1c..7d12897caa3 100644
--- a/TAO/tao/Invocation.cpp
+++ b/TAO/tao/Invocation.cpp
@@ -599,18 +599,12 @@ TAO_GIOP_Synch_Invocation::invoke_i (CORBA::Boolean is_locate_request
 {
   // Register a reply dispatcher for this invocation. Use the
   // preallocated reply dispatcher.
-
-  // Bind.
-  TAO_Transport_Mux_Strategy *tms =
-    this->transport_->tms ();
-
   TAO_Bind_Dispatcher_Guard dispatch_guard (
     this->op_details_.request_id (),
     &this->rd_,
-    tms);
-  int &status = dispatch_guard.status ();
+    this->transport_->tms ());
 
-  if (status == -1)
+  if (dispatch_guard.status () != 0)
     {
       // @@ What is the right way to handle this error?
       this->close_connection ();
@@ -620,6 +614,17 @@ TAO_GIOP_Synch_Invocation::invoke_i (CORBA::Boolean is_locate_request
                         TAO_INVOKE_EXCEPTION);
     }
 
+  // Do we have timeout se for the invocation?
+  if (this->max_wait_time_ != 0)
+    {
+      if (TAO_debug_level > 4)
+        ACE_DEBUG ((LM_DEBUG,
+                    "TAO (%P|%t) - Synch_Invocation::invoke_i, "
+                    "setting timeout in the reply dispatcher \n"));
+
+      this->rd_.has_timeout (TAO_Reply_Dispatcher::TIMEOUT);
+    }
+
   // Just send the request, without trying to wait for the reply.
   int retval = TAO_GIOP_Invocation::invoke (TAO_Transport::TAO_TWOWAY_REQUEST
                                             ACE_ENV_ARG_PARAMETER);
@@ -682,38 +687,29 @@ TAO_GIOP_Synch_Invocation::invoke_i (CORBA::Boolean is_locate_request
     }
 
   // Check the reply error.
-
   if (reply_error == -1)
     {
-      // The guard automatically unbinds the dispatcher.
-      if (errno == ETIME)
-        {
-          // Just a timeout, don't close the connection or
-          // anything...
-          ACE_THROW_RETURN (CORBA::TIMEOUT (
-              CORBA_SystemException::_tao_minor_code (
-                  TAO_TIMEOUT_SEND_MINOR_CODE,
-                  errno),
-              CORBA::COMPLETED_NO),
-            TAO_INVOKE_EXCEPTION);
-        }
+      // Check whether the error that occured is really true or not.
+      reply_error =
+        this->validate_error (dispatch_guard
+                              ACE_ENV_ARG_PARAMETER);
+      ACE_CHECK_RETURN (reply_error);
+    }
 
-      // As there is an error set the status flag to -1
-      // @@ This is a hack. The problem is -- when an error occurs we
-      // @@ try to close the connection. This just goes and destroys
-      // @@ the transport object and so the TMS. When the Bind
-      // @@ Dispatcher Guard goes out of scope the destructor is
-      // @@ called and it uses the TMS, which is "bad". Let us have
-      // @@ this for the time being -- Bala
-      status = -1;
-      // Call the ORB Core which would check whether we need to really
-      // raise an exception or are we going to base our decision on the
-      // loaded services.
+  // If this is still an error that needs special handing, call the
+  // ORB Core which would check whether we need to really
+  // raise an exception or are we going to base our decision on the
+  // loaded services.
+  if (reply_error == -1)
+    {
+      // An error has occured while waiting for the reply. So reset the
+      // state of the dispatcher guard so that no unbind happens when
+      // the destructor is called.
+      (void) dispatch_guard.status (TAO_Bind_Dispatcher_Guard::NO_UNBIND);
       return this->orb_core_->service_raise_comm_failure (this,
                                                           this->profile_
                                                           ACE_ENV_ARG_PARAMETER);
     }
-
   // @@ Alex: the old version of this had some error handling code,
   //    like:  this->profile_->reset_hint ()
   //    Can you make sure we don't forget to do that on exceptions
@@ -836,6 +832,53 @@ TAO_GIOP_Synch_Invocation::invoke_i (CORBA::Boolean is_locate_request
   return TAO_INVOKE_OK;
 }
 
+int
+TAO_GIOP_Synch_Invocation::validate_error (TAO_Bind_Dispatcher_Guard &guard
+                                           ACE_ENV_ARG_DECL)
+  ACE_THROW_SPEC ((CORBA::SystemException))
+{
+  // We can check the validity of only ETIME
+  if (errno != ETIME)
+    {
+      return -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_Invocation::invoke_i, "
+                "unbinding dispatcher after an error \n"));
+
+  int retval =
+    guard.unbind_dispatcher ();
+
+  if (retval == 0)
+    {
+      // Just a timeout, don't close the connection or
+      // anything...
+      ACE_THROW_RETURN (CORBA::TIMEOUT (
+              CORBA_SystemException::_tao_minor_code (
+                  TAO_TIMEOUT_SEND_MINOR_CODE,
+                  errno),
+              CORBA::COMPLETED_NO),
+            TAO_INVOKE_EXCEPTION);
+    }
+
+  if (TAO_debug_level > 0)
+    {
+      ACE_DEBUG ((LM_DEBUG,
+                  "TAO (%P|%t) - Synch_Invocation::validate_error "
+                  "waiting for dispatching to end \n"));
+    }
+
+  // Peek into the dispatcher to see whether we need to be waiting and
+  // if so wait
+  (void) this->rd_.wait_for_dispatch_completion ();
+
+  return 0;
+}
+
 // ****************************************************************
 
 void
diff --git a/TAO/tao/Invocation.h b/TAO/tao/Invocation.h
index 247efce5220..332b6cc8add 100644
--- a/TAO/tao/Invocation.h
+++ b/TAO/tao/Invocation.h
@@ -43,6 +43,7 @@ class TAO_Transport_Descriptor_Interface;
 class TAO_Endpoint;
 class TAO_Profile;
 class TAO_Invocation_Endpoint_Selector;
+class TAO_Bind_Dispatcher_Guard;
 
 enum TAO_Invoke_Status
 {
@@ -368,6 +369,16 @@ protected:
 
   /// Reply dispatcher for the current synchronous invocation.
   TAO_Synch_Reply_Dispatcher rd_;
+
+private:
+  /// Helper method for validating the error. Returns 0 to if
+  /// processing can be continued  or returns -1 to indicate a real
+  /// error with the invocation. Useful mostly for validating timeouts
+  /// that we got back from LF.
+  int validate_error (TAO_Bind_Dispatcher_Guard &
+                      ACE_ENV_ARG_DECL)
+    ACE_THROW_SPEC ((CORBA::SystemException));
+
 };
 
 // ****************************************************************
diff --git a/TAO/tao/Muxed_TMS.cpp b/TAO/tao/Muxed_TMS.cpp
index 5f9a12bff1e..f235e32cfe7 100644
--- a/TAO/tao/Muxed_TMS.cpp
+++ b/TAO/tao/Muxed_TMS.cpp
@@ -60,6 +60,7 @@ int
 TAO_Muxed_TMS::bind_dispatcher (CORBA::ULong request_id,
                                 TAO_Reply_Dispatcher *rd)
 {
+
   int result = this->dispatcher_table_.bind (request_id, rd);
 
   if (result != 0)
@@ -76,22 +77,23 @@ TAO_Muxed_TMS::bind_dispatcher (CORBA::ULong request_id,
   return 0;
 }
 
-void
+int
 TAO_Muxed_TMS::unbind_dispatcher (CORBA::ULong request_id)
 {
-  ACE_GUARD (TAO_SYNCH_MUTEX, ace_mon, this->lock_);
+  ACE_GUARD_RETURN (TAO_SYNCH_MUTEX,
+                    ace_mon,
+                    this->lock_,
+                    -1);
   TAO_Reply_Dispatcher *rd = 0;
-  (void) this->dispatcher_table_.unbind (request_id, rd);
+
+  // @@TODO: WTH are we sending the rd in? We can just unbind using
+  // the request_id
+  return this->dispatcher_table_.unbind (request_id, rd);
 }
 
 int
 TAO_Muxed_TMS::dispatch_reply (TAO_Pluggable_Reply_Params &params)
 {
-  // This message state should be the same as the one we have here,
-  // which we gave to the Transport to read the message. Just a sanity
-  // check here.
-  //  ACE_ASSERT (message_state == this->message_state_);
-
   int result = 0;
   TAO_Reply_Dispatcher *rd = 0;
 
@@ -99,32 +101,41 @@ TAO_Muxed_TMS::dispatch_reply (TAO_Pluggable_Reply_Params &params)
   {
     ACE_GUARD_RETURN (TAO_SYNCH_MUTEX, ace_mon, this->lock_, -1);
     result = this->dispatcher_table_.unbind (params.request_id_, rd);
-    //ACE_DEBUG ((LM_DEBUG,
-    //            "\n(%P|%t) TAO_Muxed_TMS::dispatch_reply: id = %d\n",
-    //            params.request_id_));
-  }
 
-  if (result != 0)
-    {
-      if (TAO_debug_level > 0)
-        ACE_DEBUG ((LM_DEBUG,
-                    ACE_TEXT ("(%P | %t):TAO_Muxed_TMS::dispatch_reply: ")
-                    ACE_TEXT ("unbind dispatcher failed: result = %d\n"),
-                    result));
-
-      // This return value 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.
-      return 0;
-    }
+    if (TAO_debug_level > 8)
+      ACE_DEBUG ((LM_DEBUG,
+                  "TAO (%P|%t)- TAO_Muxed_TMS::dispatch_reply, "
+                  "id = %d\n",
+                  params.request_id_));
+
+    if (result != 0)
+      {
+        if (TAO_debug_level > 0)
+          ACE_DEBUG ((LM_DEBUG,
+                      ACE_TEXT ("(%P | %t):TAO_Muxed_TMS::dispatch_reply: ")
+                      ACE_TEXT ("unbind dispatcher failed: result = %d\n"),
+                      result));
+
+        // This return value 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.
+        return 0;
+      }
+
+    // Just let the Reply_Dispatcher know that dispatching has
+    // started.
+    (void) rd->start_dispatch ();
+  }
 
   // Dispatch the reply.
   // They return 1 on success, and -1 on failure.
-  return rd->dispatch_reply (params);
+  int retval =  rd->dispatch_reply (params);
+
+  // Just let the Reply_Dispatcher know that dispatching is done.
+  (void) rd->end_dispatch ();
 
-  // No need for idling Transport, it would have got idle'd soon after
-  // sending the request.
+  return retval;
 }
 
 int
diff --git a/TAO/tao/Muxed_TMS.h b/TAO/tao/Muxed_TMS.h
index 124f5839ae2..fc942ddb610 100644
--- a/TAO/tao/Muxed_TMS.h
+++ b/TAO/tao/Muxed_TMS.h
@@ -55,7 +55,7 @@ public:
   //   class.
   virtual int bind_dispatcher (CORBA::ULong request_id,
                                TAO_Reply_Dispatcher *rh);
-  virtual void unbind_dispatcher (CORBA::ULong request_id);
+  virtual int unbind_dispatcher (CORBA::ULong request_id);
 
   virtual int dispatch_reply (TAO_Pluggable_Reply_Params &params);
 
diff --git a/TAO/tao/Reply_Dispatcher.cpp b/TAO/tao/Reply_Dispatcher.cpp
index 8e1c67591fe..86e0723344f 100644
--- a/TAO/tao/Reply_Dispatcher.cpp
+++ b/TAO/tao/Reply_Dispatcher.cpp
@@ -1,6 +1,7 @@
 // $Id$
 
 #include "tao/Reply_Dispatcher.h"
+#include "ace/Synch_T.h"
 
 ACE_RCSID(tao, Reply_Dispatcher, "$Id$")
 
@@ -10,7 +11,13 @@ ACE_RCSID(tao, Reply_Dispatcher, "$Id$")
 
 // Constructor.
 TAO_Reply_Dispatcher::TAO_Reply_Dispatcher (void)
-  : reply_status_ (100) // Just an invalid reply status.
+  // Just an invalid reply status.
+  : reply_status_ (100),
+    mutex_ (),
+    condition_ (this->mutex_),
+    timeout_ (TAO_Reply_Dispatcher::NO_TIMEOUT),
+    dispatching_ (0),
+    threads_waiting_ (0)
 {
 }
 
@@ -19,3 +26,65 @@ TAO_Reply_Dispatcher::~TAO_Reply_Dispatcher (void)
 {
 }
 
+void
+TAO_Reply_Dispatcher::start_dispatch (void)
+{
+  if (this->timeout_ != TAO_Reply_Dispatcher::TIMEOUT)
+    return;
+
+  {
+    ACE_MT (ACE_GUARD (ACE_SYNCH_MUTEX,
+                       guard,
+                       this->mutex_));
+
+    this->dispatching_ = 1;
+  }
+}
+
+
+void
+TAO_Reply_Dispatcher::end_dispatch (void)
+{
+  if (this->timeout_ != TAO_Reply_Dispatcher::TIMEOUT)
+    return;
+  {
+    ACE_MT (ACE_GUARD (ACE_SYNCH_MUTEX,
+                       ace_mon,
+                       this->mutex_));
+
+    this->dispatching_ = 0;
+
+    if (this->threads_waiting_)
+      this->condition_.signal ();
+  }
+}
+
+int
+TAO_Reply_Dispatcher::wait_for_dispatch_completion (void)
+{
+  if (this->timeout_ != TAO_Reply_Dispatcher::TIMEOUT)
+    return -1;
+
+  if (this->dispatching_)
+    {
+      ACE_MT (ACE_GUARD_RETURN (ACE_SYNCH_MUTEX,
+                                ace_mon,
+                                this->mutex_,
+                                -1));
+
+      // The dispatching could have ended by now, in which case just
+      // return.
+      if (this->dispatching_ == 0)
+        return 0;
+
+      // Mark the number of waiting threads
+      ++this->threads_waiting_;
+
+      this->condition_.wait ();
+
+      --this->threads_waiting_;
+      return 0;
+    }
+
+  return -1;
+}
diff --git a/TAO/tao/Reply_Dispatcher.h b/TAO/tao/Reply_Dispatcher.h
index 67df4e5d8df..2cb0069e72f 100644
--- a/TAO/tao/Reply_Dispatcher.h
+++ b/TAO/tao/Reply_Dispatcher.h
@@ -25,6 +25,8 @@
 # pragma once
 #endif /* ACE_LACKS_PRAGMA_ONCE */
 
+#include "Condition.h"
+
 // Forward Declarations.
 class TAO_GIOP_Message_State;
 class TAO_GIOP_Message_Version;
@@ -68,13 +70,6 @@ public:
    */
   virtual int dispatch_reply (TAO_Pluggable_Reply_Params &params) = 0;
 
-  /// Get the reply status.
-  CORBA::ULong reply_status (void) const;
-
-  // @@ Commented for the time being -  Bala
-  // virtual TAO_GIOP_Message_State *message_state (void) = 0;
-  // Get the Message State into which the reply has been read.
-
   /**
    * The used for the pending reply has been closed.
    * No reply is expected.
@@ -85,9 +80,70 @@ public:
    */
   virtual void connection_closed (void) = 0;
 
+  /// Get the reply status.
+  CORBA::ULong reply_status (void) const;
+
+  /// Following methods are useful only when the invocation has a
+  /// timeout.
+  /**
+   * MT invocations with timeouts are a different beast in
+   * itself. They need some special attention. Things get nasty when
+   * the leader thread collects the reply and at the same time the
+   * follower timesout waiting for the reply. There is a semantic
+   * problem here. Does the follower throw an exception back as a
+   * CORBA::TIMEOUT or just make one last check to see whether anyone
+   * (leader) has collected the reply. Some of these extra methods
+   * here help the invocation threads to make a decision one way or
+   * another. The code is designed to be on the positive side ie. the
+   * follower will collect the reply if a leader has collected its
+   * reply already, instead of throwing an exception.  Its a decision
+   * that can be argued for hours.
+   *
+   * The following methods should be no-ops if timeouts arent set.
+   */
+  enum
+    {
+      TIMEOUT = 0,
+      NO_TIMEOUT
+    };
+
+  /// Set whether the dispatcher should be prepared for a timeout. The
+  /// set operation is not synchronized.
+  void has_timeout (CORBA::Boolean t);
+
+  /// Methods used to change the state to indicate that whether
+  /// dispatching is started or not!
+  void start_dispatch (void);
+  void end_dispatch (void);
+
+  /// Wait on the condition variable for the dispatch to end. If no
+  /// threads are dispatching will just return immediately.
+  int wait_for_dispatch_completion (void);
+
 protected:
   /// Reply or LocateReply status.
   CORBA::ULong reply_status_;
+
+  /// Mutex and Synch condition for timeouts..
+  TAO_SYNCH_MUTEX mutex_;
+  TAO_Condition<TAO_SYNCH_MUTEX> condition_;
+
+  /// Flag to indicate whether the invocation for which this
+  /// dispatcher is used, has a timeout or not.
+  CORBA::Boolean timeout_;
+
+  /// Variable to indicate whether dispatching is happening in this
+  /// reply dispatcher.
+  /**
+   * Point to note is that, the value of the flag is not of much
+   * importance for the normal case, but only for the  case when
+   * timeouts are set.
+   */
+  CORBA::Boolean dispatching_;
+
+  /// Any threads waiting for end of dispatch?
+  CORBA::Boolean threads_waiting_;
+
 };
 
 #if defined (__ACE_INLINE__)
diff --git a/TAO/tao/Reply_Dispatcher.i b/TAO/tao/Reply_Dispatcher.i
index 9d8265bb761..f61cbee55ec 100644
--- a/TAO/tao/Reply_Dispatcher.i
+++ b/TAO/tao/Reply_Dispatcher.i
@@ -6,3 +6,8 @@ TAO_Reply_Dispatcher::reply_status (void) const
   return this->reply_status_;
 }
 
+ACE_INLINE void
+TAO_Reply_Dispatcher::has_timeout (CORBA::Boolean t)
+{
+  this->timeout_ = t;
+}
diff --git a/TAO/tao/Transport_Mux_Strategy.h b/TAO/tao/Transport_Mux_Strategy.h
index d58f8b0fd11..a7f27db98e4 100644
--- a/TAO/tao/Transport_Mux_Strategy.h
+++ b/TAO/tao/Transport_Mux_Strategy.h
@@ -62,7 +62,7 @@ public:
    * request.
    * A later reply for that request should be ignored.
    */
-  virtual void unbind_dispatcher (CORBA::ULong request_id) = 0;
+  virtual int unbind_dispatcher (CORBA::ULong request_id) = 0;
 
   /// Dispatch the reply for <request_id>, cleanup any resources
   /// allocated for that request.