ChangeLogTag:Sun Dec 14 09:56:37 2003 Balachandran Natarajan <bala@dre.vanderbilt.edu>

1 files changed, 56 insertions, 403 deletions

diff --git a/TAO/tao/Incoming_Message_Queue.cpp b/TAO/tao/Incoming_Message_Queue.cpp
index a510be13f4b..14970106b50 100644
--- a/TAO/tao/Incoming_Message_Queue.cpp
+++ b/TAO/tao/Incoming_Message_Queue.cpp
@@ -1,8 +1,7 @@
 #include "Incoming_Message_Queue.h"
-#include "Pluggable_Messaging.h"
 #include "debug.h"
-#include "ace/Malloc_T.h"
-#include "ace/Message_Block.h"
+
+#include "ace/Log_Msg.h"
 
 #if !defined (__ACE_INLINE__)
 # include "Incoming_Message_Queue.inl"
@@ -13,7 +12,7 @@ ACE_RCSID (tao,
            "$Id$")
 
 TAO_Incoming_Message_Queue::TAO_Incoming_Message_Queue (TAO_ORB_Core *orb_core)
-  : last_added_ (0),
+  : queued_data_ (0),
     size_ (0),
     orb_core_ (orb_core)
 {
@@ -43,21 +42,21 @@ TAO_Incoming_Message_Queue::copy_tail (ACE_Message_Block &block)
     {
       // Check to see if the length of the incoming block is less than
       // that of the <missing_data_> of the tail.
-      if (block.length () <= this->last_added_->missing_data_bytes_)
+      if ((CORBA::Long)block.length () <= this->queued_data_->missing_data_)
         {
           n = block.length ();
         }
       else
         {
-          n = this->last_added_->missing_data_bytes_;
+          n = this->queued_data_->missing_data_;
         }
 
       // Do the copy
-      this->last_added_->msg_block_->copy (block.rd_ptr (),
+      this->queued_data_->msg_block_->copy (block.rd_ptr (),
                                             n);
 
       // Decerement the missing data
-      this->last_added_->missing_data_bytes_ -= n;
+      this->queued_data_->missing_data_ -= n;
     }
 
   return n;
@@ -66,20 +65,17 @@ TAO_Incoming_Message_Queue::copy_tail (ACE_Message_Block &block)
 TAO_Queued_Data *
 TAO_Incoming_Message_Queue::dequeue_head (void)
 {
-  if (this->size_ == 0)
-    return 0;
-
   // Get the node on the head of the queue...
-  TAO_Queued_Data *head = this->last_added_->next_;
+  TAO_Queued_Data *tmp =
+    this->queued_data_->next_;
 
   // Reset the head node..
-  this->last_added_->next_ = head->next_;
-  
-  // Decrease the size and reset last_added_ if empty
-  if (--this->size_ == 0)
-    this->last_added_ = 0;
+  this->queued_data_->next_ = tmp->next_;
+
+  // Decrease the size
+  --this->size_;
 
-  return head;
+ return tmp;
 }
 
 TAO_Queued_Data *
@@ -90,412 +86,95 @@ TAO_Incoming_Message_Queue::dequeue_tail (void)
     return 0;
 
   // Get the node on the head of the queue...
-  TAO_Queued_Data *head =
-    this->last_added_->next_;
+  TAO_Queued_Data *tmp =
+    this->queued_data_->next_;
 
-  while (head->next_ != this->last_added_)
+  while (tmp->next_ != this->queued_data_)
     {
-      head = head->next_;
+      tmp = tmp->next_;
     }
 
   // Put the head in tmp.
-  head->next_ = this->last_added_->next_;
+  tmp->next_ = this->queued_data_->next_;
 
-  TAO_Queued_Data *ret_qd = this->last_added_;
+  TAO_Queued_Data *ret_qd = this->queued_data_;
 
-  this->last_added_ = head;
+  this->queued_data_ = tmp;
 
   // Decrease the size
-  if (--this->size_ == 0)
-    this->last_added_ = 0;
+  --this->size_;
 
  return ret_qd;
 }
 
+
 int
 TAO_Incoming_Message_Queue::enqueue_tail (TAO_Queued_Data *nd)
 {
   if (this->size_ == 0)
     {
-      this->last_added_ = nd;
-      this->last_added_->next_ = this->last_added_;
+      this->queued_data_ = nd;
+      this->queued_data_->next_ = this->queued_data_;
     }
   else
     {
-      nd->next_ = this->last_added_->next_;
-      this->last_added_->next_ = nd;
-      this->last_added_ = nd;
+      nd->next_ = this->queued_data_->next_;
+      this->queued_data_->next_ = nd;
+      this->queued_data_ = nd;
     }
 
   ++ this->size_;
   return 0;
 }
 
-TAO_Queued_Data *
-TAO_Incoming_Message_Queue::find_fragment (CORBA::Octet major,
-                                           CORBA::Octet minor) const
-{
-  TAO_Queued_Data *found = 0;
-  if (this->last_added_ != 0)
-    {
-      TAO_Queued_Data *qd = this->last_added_->next_;
-
-      do {
-        if (qd->more_fragments_ &&
-            qd->major_version_ == major && qd->minor_version_ == minor)
-          {
-            found = qd;
-          }
-        else
-          {
-            qd = qd->next_;
-          }
-      } while (found == 0 && qd != this->last_added_->next_);
-    }
-
-  return found;
-}
-
-TAO_Queued_Data *
-TAO_Incoming_Message_Queue::find_fragment (CORBA::ULong request_id) const
-{
-  TAO_Queued_Data *found = 0;
-  if (this->last_added_ != 0)
-    {
-      TAO_Queued_Data *qd = this->last_added_->next_;
-
-      do {
-        if (qd->more_fragments_ && qd->request_id_ == request_id)
-          {
-            found = qd;
-          }
-        else
-          {
-            qd = qd->next_;
-          }
-      } while (found == 0 && qd != this->last_added_->next_);
-    }
-
-  return found;
-}
-
 
 /************************************************************************/
 // Methods  for TAO_Queued_Data
 /************************************************************************/
 
 TAO_Queued_Data::TAO_Queued_Data (ACE_Allocator *alloc)
-  : msg_block_ (0)
-  , current_state_ (INVALID)
-  , missing_data_bytes_ (0)
-  , byte_order_ (0)
-  , major_version_ (0)
-  , minor_version_ (0)
-  , more_fragments_ (0)
-  , request_id_ (0)
-  , msg_type_ (TAO_PLUGGABLE_MESSAGE_MESSAGERROR)
-  , next_ (0)
-  , allocator_ (alloc)
+  : msg_block_ (0),
+    missing_data_ (0),
+    byte_order_ (0),
+    major_version_ (0),
+    minor_version_ (0),
+    more_fragments_ (0),
+    msg_type_ (TAO_PLUGGABLE_MESSAGE_MESSAGERROR),
+    next_ (0),
+    allocator_ (alloc)
 {
 }
 
 TAO_Queued_Data::TAO_Queued_Data (ACE_Message_Block *mb,
                                   ACE_Allocator *alloc)
-  : msg_block_ (mb)
-  , current_state_ (INVALID)
-  , missing_data_bytes_ (0)
-  , byte_order_ (0)
-  , major_version_ (0)
-  , minor_version_ (0)
-  , more_fragments_ (0)
-  , request_id_ (0)
-  , msg_type_ (TAO_PLUGGABLE_MESSAGE_MESSAGERROR)
-  , next_ (0)
-  , allocator_ (alloc)
+  : msg_block_ (mb),
+    missing_data_ (0),
+    byte_order_ (0),
+    major_version_ (0),
+    minor_version_ (0),
+    more_fragments_ (0),
+    msg_type_ (TAO_PLUGGABLE_MESSAGE_MESSAGERROR),
+    next_ (0),
+    allocator_ (alloc)
 {
 }
 
 TAO_Queued_Data::TAO_Queued_Data (const TAO_Queued_Data &qd)
-  : msg_block_ (qd.msg_block_->duplicate ())
-  , current_state_ (qd.current_state_)
-  , missing_data_bytes_ (qd.missing_data_bytes_)
-  , byte_order_ (qd.byte_order_)
-  , major_version_ (qd.major_version_)
-  , minor_version_ (qd.minor_version_)
-  , more_fragments_ (qd.more_fragments_)
-  , request_id_ (qd.request_id_)
-  , msg_type_ (qd.msg_type_)
-  , next_ (0)
-  , allocator_ (qd.allocator_)
+  : msg_block_ (qd.msg_block_->duplicate ()),
+    missing_data_ (qd.missing_data_),
+    byte_order_ (qd.byte_order_),
+    major_version_ (qd.major_version_),
+    minor_version_ (qd.minor_version_),
+    more_fragments_ (qd.more_fragments_),
+    msg_type_ (qd.msg_type_),
+    next_ (0),
+    allocator_ (qd.allocator_)
 {
 }
 
-
-/*!
-  \brief Allocate and return a new empty message block of size \a new_size mimicking parameters of \a mb.
-
-  This function allocates a new aligned message block using the same
-  allocators and flags as found in \a mb.  The size of the new message
-  block is at least \a new_size; the size may be adjusted up in order
-  to accomodate alignment requirements and still fit \a new_size bytes
-  into the aligned buffer.
-  
-  \param mb message block whose parameters should be mimicked
-  \param new_size size of the new message block (will be adjusted for proper alignment)
-  \return an aligned message block with rd_ptr sitting at correct alignment spot, 0 on failure
-
-  \author Thanks to Rich Seibel for helping implement with the public API for ACE_Message_Block!
- */
-static ACE_Message_Block*
-clone_mb_nocopy_size (ACE_Message_Block *mb, size_t span_size)
-{
-  // Calculate the required size of the cloned block with alignment
-  size_t aligned_size = ACE_CDR::first_size (span_size + ACE_CDR::MAX_ALIGNMENT);
-
-  // Get the allocators
-  ACE_Allocator *data_allocator;
-  ACE_Allocator *data_block_allocator;
-  ACE_Allocator *message_block_allocator;
-  mb->access_allocators (data_allocator, 
-                         data_block_allocator, 
-                         message_block_allocator);
-
-  // Create a new Message Block
-  ACE_Message_Block *nb;
-  ACE_NEW_MALLOC_RETURN (nb,
-                         ACE_static_cast(ACE_Message_Block*,
-                                         message_block_allocator->malloc (
-                                           sizeof (ACE_Message_Block))),
-                         ACE_Message_Block(aligned_size,
-                                           mb->msg_type(),
-                                           mb->cont(),
-                                           0, //we want the data block created
-                                           data_allocator,
-                                           mb->locking_strategy(),
-                                           mb->msg_priority(),
-                                           mb->msg_execution_time (),
-                                           mb->msg_deadline_time (),
-                                           data_block_allocator,
-                                           message_block_allocator),
-                         0);
-
-  ACE_CDR::mb_align (nb);
-
-  // Copy the flags over, but be SURE to clear the DONT_DELETE flag, since
-  // we just dynamically allocated the two things.
-  nb->set_flags (mb->flags());
-  nb->clr_flags (ACE_Message_Block::DONT_DELETE);
-
-  return nb;
-}
-
-/*!
-  \brief Copy data from \a src->rd_ptr to \a dst->wr_ptr, of at most \a span_size bytes.
-
-  (This is similar to memcpy, although with message blocks we can be a
-  little smarter.)  This function assumes that \a dst has enough space
-  for \a span_size bytes, and that \a src has at least \a span_size
-  bytes available to copy.  When everything is copied \a dst->wr_ptr
-  gets updated accordingly, but \a src->rd_ptr is left to the caller
-  to update.
-
-  \param dst the destination message block
-  \param src the source message block
-  \param span_size size of the maximum span of bytes to be copied
-  \return 0 on failure, otherwise \a dst
- */
-static ACE_Message_Block*
-copy_mb_span (ACE_Message_Block *dst, ACE_Message_Block *src, size_t span_size)
-{
-  // @todo check for enough space in dst, and src contains at least span_size
-
-  if (src == 0 || dst == 0)
-    return 0;
-
-  if (span_size == 0)
-    return dst;
-
-  dst->copy (src->rd_ptr (), span_size);
-  return dst;
-}
-
 /*static*/
 TAO_Queued_Data *
-TAO_Queued_Data::make_uncompleted_message (ACE_Message_Block *mb,
-                                           TAO_Pluggable_Messaging &msging_obj,
-                                           ACE_Allocator *alloc)
-{
-  register TAO_Queued_Data *new_qd = 0;
-  register const size_t HDR_LEN = msging_obj.header_length (); /* COMPUTE ONCE! */
-  register const size_t MB_LEN = mb->length (); /* COMPUTE ONCE! */
-
-  // Validate arguments.
-  if (mb == 0)
-    goto failure;
-
-  new_qd = make_queued_data (alloc);
-  if (new_qd == 0)
-    goto failure;
-
-  // do we have enough bytes to make a complete header?
-  if (MB_LEN >= HDR_LEN)
-    {
-      // Since we have enough bytes to make a complete header,
-      // the header needs to be valid.  Check that now, and punt
-      // if it's not valid.
-      if (! msging_obj.check_for_valid_header (*mb))
-        {
-          goto failure;
-        }
-      else
-        {
-          new_qd->current_state_ = WAITING_TO_COMPLETE_PAYLOAD;
-          msging_obj.set_queued_data_from_message_header (new_qd, *mb);
-          if (new_qd->current_state_ == INVALID)
-            goto failure;
-
-          // missing_data_bytes_ now has the full GIOP message size, so we allocate
-          // a new message block of that size, plus the header.
-          new_qd->msg_block_ = clone_mb_nocopy_size  (mb,
-                                                      new_qd->missing_data_bytes_ +
-                                                      HDR_LEN);
-          // Of course, we don't have the whole message (if we did, we
-          // wouldn't be here!), so we copy only what we've got, i.e., whatever's
-          // in the message block.
-          if (copy_mb_span (new_qd->msg_block_, mb, MB_LEN) == 0)
-            goto failure;
-
-          // missing_data_bytes_ now has the full GIOP message size, but
-          // there might still be stuff in mb.  Therefore, we have to adjust
-          // missing_data_bytes_, i.e., decrease it by the number of "actual
-          // payload bytes" in mb.
-          //
-          // "actual payload bytes" :== length of mb (which included the header) - header length
-          new_qd->missing_data_bytes_ -= (MB_LEN - HDR_LEN);
-          mb->rd_ptr (MB_LEN);
-        }
-    }
-  else
-    {
-      new_qd->current_state_ = WAITING_TO_COMPLETE_HEADER;
-      new_qd->msg_block_ = clone_mb_nocopy_size (mb, HDR_LEN);
-      if (new_qd->msg_block_ == 0 ||
-          copy_mb_span (new_qd->msg_block_, mb, MB_LEN) == 0)
-        goto failure;
-      new_qd->missing_data_bytes_ = HDR_LEN - MB_LEN;
-      mb->rd_ptr (MB_LEN);
-    }
-
-  ACE_ASSERT (new_qd->current_state_ != INVALID);
-  if (TAO_debug_level > 7)
-    {
-      const char* s = "?unk?";
-      switch (new_qd->current_state_)
-        {
-        case WAITING_TO_COMPLETE_HEADER: s = "WAITING_TO_COMPLETE_HEADER"; break;
-        case WAITING_TO_COMPLETE_PAYLOAD: s = "WAITING_TO_COMPLETE_PAYLOAD"; break;
-        case INVALID: s = "INVALID"; break;
-        case COMPLETED: s = "COMPLETED"; break;
-        }
-      ACE_DEBUG ((LM_DEBUG,
-                  ACE_TEXT ("TAO (%P|%t) Queued_Data::make_uncompleted_message: ")
-                  ACE_TEXT ("made uncompleted message from %u bytes into qd=%-08x:")
-                  ACE_TEXT ("state=%s,missing_data_bytes=%u\n"),
-                  new_qd->msg_block_->length(), new_qd, s, new_qd->missing_data_bytes_));
-    }
-  return new_qd;
-
-failure:
-  if (TAO_debug_level > 7)
-    {
-      ACE_DEBUG ((LM_DEBUG,
-                  ACE_TEXT ("TAO (%P|%t) Queued_Data::make_uncompleted_message: ")
-                  ACE_TEXT ("failed to make uncompleted message: mb=%-08x, qd=%-08x\n"),
-                  mb, new_qd));
-    }
-  TAO_Queued_Data::release (new_qd);
-  return 0;
-}
-
-
-/*static*/
-TAO_Queued_Data *
-TAO_Queued_Data::make_completed_message (ACE_Message_Block &mb,
-                                         TAO_Pluggable_Messaging &msging_obj,
-                                         ACE_Allocator *alloc)
-{
-  register const size_t HDR_LEN = msging_obj.header_length ();
-  register const size_t MB_LEN  = mb.length ();
-
-  // Validate arguments.
-  if (MB_LEN < HDR_LEN)
-    return 0;
-
-  size_t total_msg_len = 0;
-  register TAO_Queued_Data *new_qd = make_queued_data (alloc);
-  if (new_qd == 0)
-    goto failure;
-
-  // We can assume that there are enough bytes for a header, so
-  // extract the header data.  Don't assume that there's enough for
-  // the payload just yet.
-  new_qd->current_state_ = WAITING_TO_COMPLETE_PAYLOAD;
-  msging_obj.set_queued_data_from_message_header (new_qd, mb);
-  if (new_qd->current_state_ == INVALID)
-    goto failure;
-
-  // new_qd_->missing_data_bytes_ + protocol header length should be
-  // *at least* the length of the message.  Verify that we have that
-  // many bytes in the message block and, if we don't, release the new
-  // qd and fail.
-  total_msg_len = new_qd->missing_data_bytes_ + HDR_LEN;
-  if (total_msg_len > MB_LEN)
-    goto failure;
-
-  // Make a copy of the relevant portion of mb and hang on to it
-  if ((new_qd->msg_block_ = clone_mb_nocopy_size (&mb, total_msg_len)) == 0)
-    goto failure;
-
-  if (copy_mb_span (new_qd->msg_block_, &mb, total_msg_len) == 0)
-    goto failure;
-  
-  // Update missing data and the current state
-  new_qd->missing_data_bytes_ = 0;
-  new_qd->current_state_ = COMPLETED;
-
-  // Advance the rd_ptr on the message block
-  mb.rd_ptr (total_msg_len);
-
-  if (TAO_debug_level > 7)
-    {
-      ACE_DEBUG ((LM_DEBUG,
-                  ACE_TEXT ("TAO (%P|%t) Queued_Data::make_complete_message: ")
-                  ACE_TEXT ("extracted complete message (%u bytes incl hdr) from mblk=%-08x into qd=%-08x\n"),
-                  total_msg_len, &mb, new_qd));
-    }
-
-  return new_qd;
-
-failure:
-  if (TAO_debug_level > 7)
-    {
-      ACE_DEBUG ((LM_DEBUG,
-                  ACE_TEXT ("TAO (%P|%t) Queued_Data::make_complete_message: ")
-                  ACE_TEXT ("failed to find complete message in mblk=%-08x; leaving %u bytes in block\n"),
-                  &mb, MB_LEN));
-      if (TAO_debug_level >= 10)
-        ACE_HEX_DUMP ((LM_DEBUG,
-                       mb.rd_ptr (), MB_LEN,
-                       ACE_TEXT ("                residual bytes in buffer")));
-
-    }
-  TAO_Queued_Data::release (new_qd);
-  return 0;
-}
-
-/*static*/
-TAO_Queued_Data *
-TAO_Queued_Data::make_queued_data (ACE_Allocator *alloc)
+TAO_Queued_Data::get_queued_data (ACE_Allocator *alloc)
 {
   TAO_Queued_Data *qd = 0;
 
@@ -602,29 +281,3 @@ TAO_Queued_Data::duplicate (TAO_Queued_Data &sqd)
 
   return qd;
 }
-
-void
-TAO_Queued_Data::consolidate (void)
-{
-  // Is this a chain of fragments?
-  if (this->more_fragments_ && this->msg_block_->cont () != 0)
-    {
-      // Create a message block big enough to hold the entire chain
-      ACE_Message_Block *dest = clone_mb_nocopy_size (
-                                      this->msg_block_,
-                                      this->msg_block_->total_length ());
-      // Reset the cont() parameter
-      dest->cont (0);
-
-      // Use ACE_CDR to consolidate the chain for us
-      ACE_CDR::consolidate (dest, this->msg_block_);
-
-      // free the original message block chain
-      this->msg_block_->release ();
-
-      // Set the message block to the new consolidated message block
-      this->msg_block_ = dest;
-      this->more_fragments_ = 0;
-    }
-}
-