TAO/tao/Asynch_Queued_Message.cpp

   1 // -*- C++ -*-
   2 #include "tao/Asynch_Queued_Message.h"
   3 #include "tao/debug.h"
   4 #include "tao/ORB_Core.h"
   5
   6 #include "ace/OS_Memory.h"
   7 #include "ace/OS_NS_string.h"
   8 #include "ace/os_include/sys/os_uio.h"
   9 #include "ace/Log_Msg.h"
  10 #include "ace/Message_Block.h"
  11 #include "ace/Malloc_Base.h"
  12 #include "ace/High_Res_Timer.h"
  13
  14 TAO_BEGIN_VERSIONED_NAMESPACE_DECL
  15
  16 TAO_Asynch_Queued_Message::TAO_Asynch_Queued_Message (
  17   const ACE_Message_Block *contents,
  18   TAO_ORB_Core *oc,
  19   ACE_Time_Value *timeout,
  20   ACE_Allocator *alloc,
  21   bool is_heap_allocated)
  22   : TAO_Queued_Message (oc, alloc, is_heap_allocated)
  23   , size_ (contents->total_length ())
  24   , offset_ (0)
  25   , abs_timeout_ (ACE_Time_Value::zero)
  26 {
  27   if (timeout != nullptr)// && *timeout != ACE_Time_Value::zero)
  28     {
  29       this->abs_timeout_ = ACE_High_Res_Timer::gettimeofday_hr () + *timeout;
  30     }
  31   // @@ Use a pool for these guys!!
  32   ACE_NEW (this->buffer_, char[this->size_]);
  33
  34   size_t copy_offset = 0;
  35   for (const ACE_Message_Block *i = contents;
  36        i != nullptr;
  37        i = i->cont ())
  38     {
  39       ACE_OS::memcpy (this->buffer_ + copy_offset,
  40                       i->rd_ptr (),
  41                       i->length ());
  42       copy_offset += i->length ();
  43     }
  44 }
  45
  46 TAO_Asynch_Queued_Message::TAO_Asynch_Queued_Message (char *buf,
  47                                                       TAO_ORB_Core *oc,
  48                                                       size_t size,
  49                                                       const ACE_Time_Value &abs_timeout,
  50                                                       ACE_Allocator *alloc,
  51                                                       bool is_heap_allocated)
  52   : TAO_Queued_Message (oc, alloc, is_heap_allocated)
  53   , size_ (size)
  54   , offset_ (0)
  55   , buffer_ (buf)
  56   , abs_timeout_ (abs_timeout)
  57 {
  58 }
  59
  60 TAO_Asynch_Queued_Message::~TAO_Asynch_Queued_Message ()
  61 {
  62   // @@ Use a pool for these guys!
  63   delete [] this->buffer_;
  64 }
  65
  66 size_t
  67 TAO_Asynch_Queued_Message::message_length () const
  68 {
  69   return this->size_ - this->offset_;
  70 }
  71
  72 int
  73 TAO_Asynch_Queued_Message::all_data_sent () const
  74 {
  75   return this->size_ == this->offset_;
  76 }
  77
  78 void
  79 TAO_Asynch_Queued_Message::fill_iov (int iovcnt_max,
  80                                      int &iovcnt,
  81                                      iovec iov[]) const
  82 {
  83   ACE_ASSERT (iovcnt_max > iovcnt);
  84   ACE_UNUSED_ARG (iovcnt_max); // not used if ACE_ASSERT() is empty
  85
  86   iov[iovcnt].iov_base = this->buffer_ + this->offset_;
  87   iov[iovcnt].iov_len  = static_cast<u_long> (this->size_ - this->offset_);
  88   ++iovcnt;
  89 }
  90
  91 void
  92 TAO_Asynch_Queued_Message::bytes_transferred (size_t &byte_count)
  93 {
  94   this->state_changed_i (TAO_LF_Event::LFS_ACTIVE);
  95
  96   size_t const remaining_bytes = this->size_ - this->offset_;
  97   if (byte_count > remaining_bytes)
  98     {
  99       this->offset_ = this->size_;
 100       byte_count -= remaining_bytes;
 101       return;
 102     }
 103   this->offset_ += byte_count;
 104   byte_count = 0;
 105
 106   if (this->all_data_sent ())
 107     this->state_changed (TAO_LF_Event::LFS_SUCCESS,
 108                          this->orb_core_->leader_follower ());
 109 }
 110
 111
 112 TAO_Queued_Message *
 113 TAO_Asynch_Queued_Message::clone (ACE_Allocator *alloc)
 114 {
 115   char *buf = nullptr;
 116
 117   // @todo: Need to use a memory pool. But certain things need to
 118   // change a bit in this class for that. Till then.
 119
 120   // Just allocate and copy data that needs to be sent, no point
 121   // copying the whole buffer.
 122   size_t const sz = this->size_ - this->offset_;
 123
 124   ACE_NEW_RETURN (buf,
 125                   char[sz],
 126                   nullptr);
 127
 128   ACE_OS::memcpy (buf,
 129                   this->buffer_ + this->offset_,
 130                   sz);
 131
 132   TAO_Asynch_Queued_Message *qm = nullptr;
 133
 134   if (alloc)
 135     {
 136       ACE_NEW_MALLOC_RETURN (qm,
 137                              static_cast<TAO_Asynch_Queued_Message *> (
 138                                  alloc->malloc (sizeof (TAO_Asynch_Queued_Message))),
 139                              TAO_Asynch_Queued_Message (buf,
 140                                                         this->orb_core_,
 141                                                         sz,
 142                                                         this->abs_timeout_,
 143                                                         alloc,
 144                                                         true),
 145                              nullptr);
 146     }
 147   else
 148     {
 149       // No allocator, so use the common heap!
 150       if (TAO_debug_level == 4)
 151         {
 152           // This debug is for testing purposes!
 153           TAOLIB_DEBUG ((LM_DEBUG,
 154                       "TAO (%P|%t) - Asynch_Queued_Message::clone\n"
 155                       "Using global pool for allocation\n"));
 156         }
 157
 158       ACE_NEW_RETURN (qm,
 159                       TAO_Asynch_Queued_Message (buf,
 160                                                  this->orb_core_,
 161                                                  sz,
 162                                                  this->abs_timeout_,
 163                                                  nullptr,
 164                                                  true),
 165                       nullptr);
 166     }
 167
 168   return qm;
 169 }
 170
 171 void
 172 TAO_Asynch_Queued_Message::destroy ()
 173 {
 174   if (this->is_heap_created_)
 175     {
 176       // If we have an allocator release the memory to the allocator
 177       // pool.
 178       if (this->allocator_)
 179         {
 180           ACE_DES_FREE_THIS (this->allocator_->free,
 181                              TAO_Asynch_Queued_Message);
 182         }
 183       else // global release..
 184         {
 185           delete this;
 186         }
 187     }
 188 }
 189
 190 bool
 191 TAO_Asynch_Queued_Message::is_expired (const ACE_Time_Value &now) const
 192 {
 193   if (this->abs_timeout_ > ACE_Time_Value::zero)
 194     {
 195       if (this->offset_ > 0)
 196         {
 197           return false; //never expire partial messages
 198         }
 199       return this->abs_timeout_ < now;
 200     }
 201   return false;
 202 }
 203
 204 void
 205 TAO_Asynch_Queued_Message::copy_if_necessary (const ACE_Message_Block*)
 206 {
 207   // It's never necessary for asynchronously queued messages
 208 }
 209
 210 TAO_END_VERSIONED_NAMESPACE_DECL