ChangeLogTag:Thu Nov 6 22:32:19 2003 Balachandran Natarajan <bala@dre.vanderbilt.edu>

12 files changed, 175 insertions, 93 deletions

diff --git a/TAO/ChangeLog b/TAO/ChangeLog
index be5af308086..d045917b483 100644
--- a/TAO/ChangeLog
+++ b/TAO/ChangeLog
@@ -1,3 +1,83 @@
+Thu Nov  6 22:32:19 2003  Balachandran Natarajan  <bala@dre.vanderbilt.edu>
+
+	This checkin is about an optimization and not definitely for the
+	weak hearted :-)! All this jugglery is for getting rid of one
+        allocation along the critical path. Not sure how much would this
+	fetch us, but never the less an useful optimization. 
+
+	A quick brief. In older versions of TAO, the client thread used to
+        create a output CDR on the TSS and use that for marshalling data
+        and sending data which was neat. But during the reply, we used to
+        create an InputCDR on the global memory since the thread reading
+        the reply message could be different from the thread to whom the
+        reply belongs to. If you want to know why, look for documentation
+        of TAO's LF. After we worked on BUG 1369, we got a few advantages
+        and this  optimization cashed in on that. 
+
+	Refactoring the generated code allowed us to move the CDR stream
+        inside the transport, to be more specific inside the GIOP
+        objects. This was also right, since the CDR's are tied to
+        GIOP. But this gave rise to a problem. We could not use the TSS
+        anymore since multiple threads could share the same transport (one
+        after another). Hence we used the global memory pool (different
+	from the global heap) to allocate the CDR and this is a one time
+	activity cached inside the transport. All threads can infact use
+	the same chunk of memory. That was done when we merged the branch
+        for fixes from 1369. Since the transports are anyway on a per-lane
+        basis RTCORBA wouldn't be affected at all.
+
+	This checkin broke another barrier, ie. the allocation required
+        while getting the replies. The thread collecting the reply can
+        read the messages on its stack and copy the messages to the stack
+        of another thread which is expecting a reply. Therefore we now
+        require no allocations but for a copy. 
+
+	All the above are valid for messages of decent size. If the
+        messages are big we have to hit the global memory. What are the
+        end results?
+
+	- TAO now does not do any allocation along the outgoing request
+	  path. 
+
+	- It doesn't do any allocation along the reply collection path
+	  even in MT scenarios for decent sized messages and when the
+	  network is behaving well (ie. when there is no IP
+	  fragmentation).  
+
+	We were also helped by an optimization that was added in ACE
+        yesterday to the CDR streams. We have to wait and see how much
+        does this checkin buys us in terms of performance. Anyway, this 
+        documentation should be useful for others.
+	
+	* tao/GIOP_Message_Lite.cpp:
+	* tao/GIOP_Message_Base.cpp (process_reply_message):
+
+	  Construct the CDR stream and use that CDR to dispatch the
+	  reply. Now the reply dispatch is done by the GIOP objects. 
+
+	* tao/GIOP_Message_Generator_Parser.cpp:
+	* tao/GIOP_Message_Generator_Parser_10.cpp:
+	* tao/GIOP_Message_Generator_Parser_12.cpp:
+
+	  No need to do any exchange CDR blocks since things are being
+	  taken care elsewhere. 
+
+	* tao/Pluggable_Messaging_Utils.cpp:
+	* tao/Pluggable_Messaging_Utils.h:
+
+	  Added a transport pointer to the Reply_Params object. It is
+	  used by the GIOP layer to dispatch the reply. Removed the
+	  temporary CDR that was created here. 
+
+	* tao/Synch_Reply_Dispatcher.cpp:
+	* tao/DynamicInterface/DII_Reply_Dispatcher.cpp:
+	* tao/Messaging/Asynch_Reply_Dispatcher.cpp:
+
+	  Added a check for null CDR's in the Reply_Param's object. 
+
+	* tao/Transport.cpp: Construct the Reply_Params using the
+	  transport instead of the ORB_Core. 
+	
 Fri Nov  7 02:08:17 UTC 2003  Don Hinton  <dhinton@dresystems.com>
 
         * orbsvcs/ImplRepo_Service/Server_Repository.cpp:
diff --git a/TAO/tao/DynamicInterface/DII_Reply_Dispatcher.cpp b/TAO/tao/DynamicInterface/DII_Reply_Dispatcher.cpp
index 1ef07265194..bf713b3a628 100644
--- a/TAO/tao/DynamicInterface/DII_Reply_Dispatcher.cpp
+++ b/TAO/tao/DynamicInterface/DII_Reply_Dispatcher.cpp
@@ -49,11 +49,14 @@ TAO_DII_Deferred_Reply_Dispatcher::dispatch_reply (
     TAO_Pluggable_Reply_Params &params
   )
 {
+  if (params.input_cdr_ == 0)
+    return -1;
+
   this->reply_status_ = params.reply_status_;
 
   // Transfer the <params.input_cdr_>'s content to this->reply_cdr_
   ACE_Data_Block *db =
-    this->reply_cdr_.clone_from (params.input_cdr_);
+    this->reply_cdr_.clone_from (*params.input_cdr_);
 
 
   if (db == 0)
diff --git a/TAO/tao/GIOP_Message_Base.cpp b/TAO/tao/GIOP_Message_Base.cpp
index 4a0e8bc4c4f..76aeeaee51e 100644
--- a/TAO/tao/GIOP_Message_Base.cpp
+++ b/TAO/tao/GIOP_Message_Base.cpp
@@ -7,6 +7,7 @@
 #include "TAO_Server_Request.h"
 #include "GIOP_Message_Locate_Header.h"
 #include "Transport.h"
+#include "Transport_Mux_Strategy.h"
 #include "LF_Strategy.h"
 #include "Request_Dispatcher.h"
 #include "Codeset_Manager.h"
@@ -15,8 +16,8 @@
 # include "GIOP_Message_Base.i"
 #endif /* __ACE_INLINE__ */
 
-ACE_RCSID (tao, 
-           GIOP_Message_Base, 
+ACE_RCSID (tao,
+           GIOP_Message_Base,
            "$Id$")
 
 TAO_GIOP_Message_Base::TAO_GIOP_Message_Base (TAO_ORB_Core *orb_core,
@@ -739,7 +740,7 @@ TAO_GIOP_Message_Base::process_reply_message (
   // Create a empty buffer on stack
   // NOTE: We use the same data block in which we read the message and
   // we pass it on to the higher layers of the ORB. So we dont to any
-  // copies at all here. The same is alos done in the higher layers.
+  // copies at all here.
   TAO_InputCDR input_cdr (qd->msg_block_->data_block (),
                           ACE_Message_Block::DONT_DELETE,
                           rd_pos,
@@ -749,30 +750,52 @@ TAO_GIOP_Message_Base::process_reply_message (
                           qd->minor_version_,
                           this->orb_core_);
 
-  // Reset the message state. Now, we are ready for the next nested
-  // upcall if any.
-  // this->message_handler_.reset (0);
-
   // We know we have some reply message. Check whether it is a
   // GIOP_REPLY or GIOP_LOCATE_REPLY to take action.
 
   // Once we send the InputCDR stream we need to just forget about
   // the stream and never touch that again for anything. We basically
   // loose ownership of the data_block.
+  int retval = 0;
 
   switch (qd->msg_type_)
     {
     case TAO_PLUGGABLE_MESSAGE_REPLY:
       // Should be taken care by the state specific parsing
-      return generator_parser->parse_reply (input_cdr,
-                                            params);
+      retval =
+        generator_parser->parse_reply (input_cdr,
+                                       params);
 
+      break;
     case TAO_PLUGGABLE_MESSAGE_LOCATEREPLY:
-      return generator_parser->parse_locate_reply (input_cdr,
-                                                   params);
+      retval =
+        generator_parser->parse_locate_reply (input_cdr,
+                                              params);
+      break;
     default:
-        return -1;
+      retval = -1;
     }
+
+  if (retval == -1)
+    return retval;
+
+  params.input_cdr_ = &input_cdr;
+
+  retval =
+    params.transport_->tms ()->dispatch_reply (params);
+
+  if (retval == -1)
+    {
+      // Something really critical happened, we will forget about
+      // every reply on this connection.
+      if (TAO_debug_level > 0)
+        ACE_ERROR ((LM_ERROR,
+                    "TAO (%P|%t) - GIOP_Message_Base[%d]::process_parsed_messages, "
+                    "dispatch reply failed\n",
+                    params.transport_->id ()));
+    }
+
+  return retval;
 }
 
 int
diff --git a/TAO/tao/GIOP_Message_Generator_Parser.cpp b/TAO/tao/GIOP_Message_Generator_Parser.cpp
index f7cad45ccc5..37c2497f9db 100644
--- a/TAO/tao/GIOP_Message_Generator_Parser.cpp
+++ b/TAO/tao/GIOP_Message_Generator_Parser.cpp
@@ -2,15 +2,14 @@
 #include "tao/Pluggable_Messaging_Utils.h"
 #include "tao/GIOP_Utils.h"
 #include "tao/debug.h"
-
 #include "ace/Log_Msg.h"
 
 #if !defined (__ACE_INLINE__)
 # include "tao/GIOP_Message_Generator_Parser.inl"
 #endif /* __ACE_INLINE__ */
 
-ACE_RCSID (tao, 
-           GIOP_Message_Gen_Parser, 
+ACE_RCSID (tao,
+           GIOP_Message_Gen_Parser,
            "$Id$")
 
 int
diff --git a/TAO/tao/GIOP_Message_Generator_Parser_10.cpp b/TAO/tao/GIOP_Message_Generator_Parser_10.cpp
index a5e6fc9ad10..414f38da0c9 100644
--- a/TAO/tao/GIOP_Message_Generator_Parser_10.cpp
+++ b/TAO/tao/GIOP_Message_Generator_Parser_10.cpp
@@ -418,10 +418,6 @@ TAO_GIOP_Message_Generator_Parser_10::parse_reply (
                                                       params) == -1)
     return -1;
 
-  // Steal rest of the contents in to the reply params and loose
-  // ownership of the  data block.
-  params.input_cdr_.exchange_data_blocks (cdr);
-
   return 0;
 }
 
@@ -437,10 +433,6 @@ TAO_GIOP_Message_Generator_Parser_10::parse_locate_reply (
 
     return -1;
 
-  // Steal the contents in to the reply CDR and loose ownership of the
-  // data block.
-  params.input_cdr_.exchange_data_blocks (cdr);
-
   return 0;
 }
 
diff --git a/TAO/tao/GIOP_Message_Generator_Parser_12.cpp b/TAO/tao/GIOP_Message_Generator_Parser_12.cpp
index 8ab8a468e1d..fb83f49e386 100644
--- a/TAO/tao/GIOP_Message_Generator_Parser_12.cpp
+++ b/TAO/tao/GIOP_Message_Generator_Parser_12.cpp
@@ -366,7 +366,7 @@ TAO_GIOP_Message_Generator_Parser_12::parse_reply (
 
   if ((cdr >> params.svc_ctx_) == 0)
     {
-      if (TAO_debug_level > 0)
+      // if (TAO_debug_level > 0)
         ACE_ERROR ((LM_ERROR,
                     ACE_TEXT ("TAO (%P|%t) parse_reply, ")
                     ACE_TEXT ("extracting context\n")));
@@ -380,13 +380,6 @@ TAO_GIOP_Message_Generator_Parser_12::parse_reply (
       cdr.align_read_ptr (TAO_GIOP_MESSAGE_ALIGN_PTR);
     }
 
-
-
-
-  // Steal the contents in to the reply CDR and loose ownership of the
-  // data block.
-  params.input_cdr_.exchange_data_blocks (cdr);
-
   return 0;
 }
 
@@ -411,10 +404,6 @@ TAO_GIOP_Message_Generator_Parser_12::parse_locate_reply (
       cdr.align_read_ptr (TAO_GIOP_MESSAGE_ALIGN_PTR);
       }*/
 
-  // Steal the contents in to the reply CDR and loose ownership of the
-  // data block.
-  params.input_cdr_.exchange_data_blocks (cdr);
-
   return 0;
 }
 
diff --git a/TAO/tao/GIOP_Message_Lite.cpp b/TAO/tao/GIOP_Message_Lite.cpp
index ea5d8a47e13..462266dd9bf 100644
--- a/TAO/tao/GIOP_Message_Lite.cpp
+++ b/TAO/tao/GIOP_Message_Lite.cpp
@@ -2,16 +2,17 @@
 //
 //$Id$
 
-#include "tao/GIOP_Message_Lite.h"
-#include "tao/debug.h"
-#include "tao/TAOC.h"
-#include "tao/ORB_Core.h"
-#include "tao/operation_details.h"
-#include "tao/TAO_Server_Request.h"
-#include "tao/GIOP_Message_Locate_Header.h"
-#include "tao/LF_Strategy.h"
-#include "tao/Transport.h"
-#include "tao/Codeset_Manager.h"
+#include "GIOP_Message_Lite.h"
+#include "debug.h"
+#include "TAOC.h"
+#include "ORB_Core.h"
+#include "operation_details.h"
+#include "TAO_Server_Request.h"
+#include "GIOP_Message_Locate_Header.h"
+#include "LF_Strategy.h"
+#include "Transport.h"
+#include "Transport_Mux_Strategy.h"
+#include "Codeset_Manager.h"
 
 #if !defined (__ACE_INLINE__)
 # include "tao/GIOP_Message_Lite.i"
@@ -750,7 +751,7 @@ TAO_GIOP_Message_Lite::process_request (TAO_Transport *transport,
       if (!CORBA::is_nil (forward_to.in ()))
         {
           // We should forward to another object...
-          TAO_Pluggable_Reply_Params reply_params (this->orb_core_);
+          TAO_Pluggable_Reply_Params reply_params (transport);
           reply_params.request_id_ = request_id;
           reply_params.reply_status_ = TAO_GIOP_LOCATION_FORWARD;
           reply_params.svc_ctx_.length (0);
@@ -1130,9 +1131,20 @@ TAO_GIOP_Message_Lite::parse_reply (TAO_InputCDR &cdr,
         }
     }
 
-  // Steal the contents in to the reply CDR and loose ownership of the
-  // data block.
-  params.input_cdr_.exchange_data_blocks (cdr);
+  params.input_cdr_= &cdr;
+
+  if ( params.transport_->tms ()->dispatch_reply (params) == -1)
+    {
+      // Something really critical happened, we will forget about
+      // every reply on this connection.
+      if (TAO_debug_level > 0)
+        ACE_ERROR ((LM_ERROR,
+                    "TAO (%P|%t) - GIOP_Message_Lite[%d]::process_parsed_messages, "
+                    "dispatch reply failed\n",
+                    params.transport_->id ()));
+
+      return -1;
+    }
 
   return 0;
 }
@@ -1420,7 +1432,7 @@ TAO_GIOP_Message_Lite::send_reply_exception (
   transport->assign_translators(0,&output);
 
   // Make the GIOP & reply header. They are version specific.
-  TAO_Pluggable_Reply_Params reply_params (orb_core);
+  TAO_Pluggable_Reply_Params reply_params (transport);
   reply_params.request_id_ = request_id;
   reply_params.svc_ctx_.length (0);
 
diff --git a/TAO/tao/Messaging/Asynch_Reply_Dispatcher.cpp b/TAO/tao/Messaging/Asynch_Reply_Dispatcher.cpp
index 6150142a87b..786ad3d90c8 100644
--- a/TAO/tao/Messaging/Asynch_Reply_Dispatcher.cpp
+++ b/TAO/tao/Messaging/Asynch_Reply_Dispatcher.cpp
@@ -39,6 +39,9 @@ TAO_Asynch_Reply_Dispatcher::dispatch_reply (
     TAO_Pluggable_Reply_Params &params
   )
 {
+  if (params.input_cdr_ == 0)
+    return -1;
+
   // AMI Timeout Handling Begin
   timeout_handler_.cancel ();
 
@@ -48,7 +51,7 @@ TAO_Asynch_Reply_Dispatcher::dispatch_reply (
 
   // Transfer the <params.input_cdr_>'s content to this->reply_cdr_
   ACE_Data_Block *db =
-    this->reply_cdr_.clone_from (params.input_cdr_);
+    this->reply_cdr_.clone_from (*params.input_cdr_);
 
   if (db == 0)
     {
diff --git a/TAO/tao/Pluggable_Messaging_Utils.cpp b/TAO/tao/Pluggable_Messaging_Utils.cpp
index f4ab4646cc0..d7f79cecec5 100644
--- a/TAO/tao/Pluggable_Messaging_Utils.cpp
+++ b/TAO/tao/Pluggable_Messaging_Utils.cpp
@@ -7,21 +7,15 @@
 #include "tao/Pluggable_Messaging_Utils.i"
 #endif /* __ACE_INLINE__ */
 
-ACE_RCSID (tao, 
-           Pluggable_Messaging_Utils, 
+ACE_RCSID (tao,
+           Pluggable_Messaging_Utils,
            "$Id$")
 
 
 TAO_Pluggable_Reply_Params::TAO_Pluggable_Reply_Params (
-    TAO_ORB_Core *orb_core
+    TAO_Transport *t
   )
-  : input_cdr_ (orb_core->create_input_cdr_data_block (
-                              ACE_CDR::DEFAULT_BUFSIZE
-                            ),
-                0,
-                TAO_ENCAP_BYTE_ORDER,
-                TAO_DEF_GIOP_MAJOR,
-                TAO_DEF_GIOP_MINOR,
-                orb_core)
+  : input_cdr_ (0)
+  ,  transport_ (t)
 {
 }
diff --git a/TAO/tao/Pluggable_Messaging_Utils.h b/TAO/tao/Pluggable_Messaging_Utils.h
index 193d8c70753..d85c44770bd 100644
--- a/TAO/tao/Pluggable_Messaging_Utils.h
+++ b/TAO/tao/Pluggable_Messaging_Utils.h
@@ -8,7 +8,6 @@
  *
  *   Utility classes for the TAO pluggable messaging framework.
  *
- *
  *  @author  Balachandran Natarajan <bala@cs.wustl.edu>
  */
 //=============================================================================
@@ -25,6 +24,9 @@
 # pragma once
 #endif /* ACE_LACKS_PRAGMA_ONCE */
 
+
+class TAO_Transport;
+
 /**
  * @class TAO_Pluggable_Reply_Params_Base
  *
@@ -88,23 +90,19 @@ protected:
  *
  * @brief TAO_Pluggable_Connector_Params
  *
- * This represents a set of data that would be received by the
- * connector from the acceptor.
  */
 class TAO_Export TAO_Pluggable_Reply_Params
   : public TAO_Pluggable_Reply_Params_Base
 {
 public:
   /// Constructor.
-  TAO_Pluggable_Reply_Params (TAO_ORB_Core *orb_core);
-
-  /* @todo: There is a way out clear this off from stack. Need to look
-     into that after 1.2
-  */
+  TAO_Pluggable_Reply_Params (TAO_Transport *t);
 
   /// The stream with the non-demarshaled reply. This stream will be
   /// passed up to the stubs to demarshal the parameter values.
-  TAO_InputCDR input_cdr_;
+  TAO_InputCDR *input_cdr_;
+
+  TAO_Transport *transport_;
 };
 
 // @@ Bala: this is a GIOPism too, there is no such thing as locate
diff --git a/TAO/tao/Synch_Reply_Dispatcher.cpp b/TAO/tao/Synch_Reply_Dispatcher.cpp
index 47b0b31cc70..1233fdcd3cf 100644
--- a/TAO/tao/Synch_Reply_Dispatcher.cpp
+++ b/TAO/tao/Synch_Reply_Dispatcher.cpp
@@ -49,6 +49,13 @@ int
 TAO_Synch_Reply_Dispatcher::dispatch_reply (
     TAO_Pluggable_Reply_Params &params)
 {
+  if (params.input_cdr_ == 0)
+    return -1;
+
+  if (TAO_debug_level > 2)
+    ACE_ERROR ((LM_ERROR,
+                "TAO (%P|%t) - Synch_Reply_Dispatcher::dispatch_reply ",
+                "going on \n"));
   this->reply_status_ = params.reply_status_;
 
   // Steal the buffer, that way we don't do any unnecesary copies of
@@ -64,7 +71,7 @@ TAO_Synch_Reply_Dispatcher::dispatch_reply (
 
   // Transfer the <params.input_cdr_>'s content to this->reply_cdr_
   ACE_Data_Block *db =
-    this->reply_cdr_.clone_from (params.input_cdr_);
+    this->reply_cdr_.clone_from (*params.input_cdr_);
 
   if (db == 0)
     {
diff --git a/TAO/tao/Transport.cpp b/TAO/tao/Transport.cpp
index 59885f09ef4..9923ff1f895 100644
--- a/TAO/tao/Transport.cpp
+++ b/TAO/tao/Transport.cpp
@@ -1762,7 +1762,7 @@ TAO_Transport::process_parsed_messages (TAO_Queued_Data *qd,
   // Get the <message_type> that we have received
   TAO_Pluggable_Message_Type t =  qd->msg_type_;
 
-  int result = 0;
+  // int result = 0;
 
   if (t == TAO_PLUGGABLE_MESSAGE_CLOSECONNECTION)
     {
@@ -1797,10 +1797,7 @@ TAO_Transport::process_parsed_messages (TAO_Queued_Data *qd,
     {
       rh.resume_handle ();
 
-      // @@todo: Maybe the input_cdr can be constructed from the
-      // message_block
-      TAO_Pluggable_Reply_Params params (this->orb_core ());
-
+      TAO_Pluggable_Reply_Params params (this);
 
       if (this->messaging_object ()->process_reply_message (params,
                                                             qd) == -1)
@@ -1814,21 +1811,6 @@ TAO_Transport::process_parsed_messages (TAO_Queued_Data *qd,
           return -1;
         }
 
-      result = this->tms ()->dispatch_reply (params);
-
-      if (result == -1)
-        {
-          // Something really critical happened, we will forget about
-          // every reply on this connection.
-          if (TAO_debug_level > 0)
-            ACE_ERROR ((LM_ERROR,
-                        "TAO (%P|%t) - Transport[%d]::process_parsed_messages, "
-                        "dispatch reply failed\n",
-                        this->id ()));
-
-          return -1;
-        }
-
     }
   else if (t == TAO_PLUGGABLE_MESSAGE_MESSAGERROR)
     {