ChangeLogTag:Thu Jul 5 14:39:45 2001 Balachandran Natarajan <bala@cs.wustl.edu>

12 files changed, 209 insertions, 24 deletions

diff --git a/ace/CDR_Stream.cpp b/ace/CDR_Stream.cpp
index 5bb165a0237..b633862af35 100644
--- a/ace/CDR_Stream.cpp
+++ b/ace/CDR_Stream.cpp
@@ -10,6 +10,7 @@
 ACE_RCSID(ace, CDR_Stream, "$Id$")
 
 
+
 // ****************************************************************
 
 ACE_OutputCDR::ACE_OutputCDR (size_t size,
@@ -625,7 +626,8 @@ ACE_InputCDR::ACE_InputCDR (ACE_Data_Block *data,
 ACE_InputCDR::ACE_InputCDR (const ACE_InputCDR& rhs,
                             size_t size,
                             ACE_CDR::Long offset)
-  : start_ (rhs.start_.data_block ()->duplicate ()),
+  : start_ (rhs.start_,
+            ACE_CDR::MAX_ALIGNMENT),
     do_byte_swap_ (rhs.do_byte_swap_),
     good_bit_ (1),
     char_translator_ (0),
@@ -633,10 +635,16 @@ ACE_InputCDR::ACE_InputCDR (const ACE_InputCDR& rhs,
     major_version_ (rhs.major_version_),
     minor_version_ (rhs.minor_version_)
 {
-  char* newpos = rhs.start_.rd_ptr() + offset;
-  if (this->start_.base () <= newpos
-      && newpos <= this->start_.end ()
-      && newpos + size <= this->start_.end ())
+  // Align the base pointer assuming that the incoming stream is also
+  // aligned the way we are aligned
+  char *incoming_start = ACE_ptr_align_binary (rhs.start_.base (),
+                                               ACE_CDR::MAX_ALIGNMENT);
+
+  size_t newpos =
+    (rhs.start_.rd_ptr() - incoming_start)  + offset;
+
+  if (newpos <= this->start_.space ()
+      && newpos + size <= this->start_.space ())
     {
       this->start_.rd_ptr (newpos);
       this->start_.wr_ptr (newpos + size);
@@ -649,7 +657,8 @@ ACE_InputCDR::ACE_InputCDR (const ACE_InputCDR& rhs,
 
 ACE_InputCDR::ACE_InputCDR (const ACE_InputCDR& rhs,
                             size_t size)
-  : start_ (rhs.start_.data_block ()->duplicate ()),
+  : start_ (rhs.start_,
+            ACE_CDR::MAX_ALIGNMENT),
     do_byte_swap_ (rhs.do_byte_swap_),
     good_bit_ (1),
     char_translator_ (0),
@@ -657,10 +666,16 @@ ACE_InputCDR::ACE_InputCDR (const ACE_InputCDR& rhs,
     major_version_ (rhs.major_version_),
     minor_version_ (rhs.minor_version_)
 {
-  char* newpos = rhs.start_.rd_ptr();
-  if (this->start_.base () <= newpos
-      && newpos <= this->start_.end ()
-      && newpos + size <= this->start_.end ())
+  // Align the base pointer assuming that the incoming stream is also
+  // aligned the way we are aligned
+  char *incoming_start = ACE_ptr_align_binary (rhs.start_.base (),
+                                               ACE_CDR::MAX_ALIGNMENT);
+
+  size_t newpos =
+    rhs.start_.rd_ptr() - incoming_start;
+
+  if (newpos <= this->start_.space ()
+      && newpos + size <= this->start_.space ())
     {
       // Notice that ACE_Message_Block::duplicate may leave the
       // wr_ptr() with a higher value that what we actually want.
@@ -678,7 +693,8 @@ ACE_InputCDR::ACE_InputCDR (const ACE_InputCDR& rhs,
 }
 
 ACE_InputCDR::ACE_InputCDR (const ACE_InputCDR& rhs)
-  : start_ (rhs.start_.data_block ()->duplicate ()),
+  : start_ (rhs.start_,
+            ACE_CDR::MAX_ALIGNMENT),
     do_byte_swap_ (rhs.do_byte_swap_),
     good_bit_ (1),
     char_translator_ (rhs.char_translator_),
@@ -686,8 +702,13 @@ ACE_InputCDR::ACE_InputCDR (const ACE_InputCDR& rhs)
     major_version_ (rhs.major_version_),
     minor_version_ (rhs.minor_version_)
 {
-  this->start_.rd_ptr (rhs.start_.rd_ptr ());
-  this->start_.wr_ptr (rhs.start_.wr_ptr ());
+  char *buf = ACE_ptr_align_binary (rhs.start_.base (),
+                                    ACE_CDR::MAX_ALIGNMENT);
+
+  size_t rd_offset = rhs.start_.rd_ptr () - buf;
+  size_t wr_offset = rhs.start_.rd_ptr () - buf;
+  this->start_.rd_ptr (rd_offset);
+  this->start_.wr_ptr (wr_offset);
 }
 
 ACE_InputCDR::ACE_InputCDR (ACE_InputCDR::Transfer_Contents x)
@@ -783,18 +804,18 @@ ACE_InputCDR::read_wchar (ACE_CDR::WChar& x)
           return this->read_octet_array (
                            ACE_reinterpret_cast (ACE_CDR::Octet*,
                                                  &x),
-                           ACE_static_cast (ACE_CDR::ULong, 
+                           ACE_static_cast (ACE_CDR::ULong,
                                             len)
                          );
         }
     }
   else if (this->wchar_translator_ == 0)
     {
-      return this->read_2 (ACE_reinterpret_cast (ACE_CDR::UShort*, 
+      return this->read_2 (ACE_reinterpret_cast (ACE_CDR::UShort*,
                                                  &x));
     }
 
-  return this->wchar_translator_->read_wchar (*this, 
+  return this->wchar_translator_->read_wchar (*this,
                                               x);
 }
 
@@ -806,7 +827,7 @@ ACE_InputCDR::read_string (char *&x)
   // smaller and should be better for the cache ;-) ;-)
   if (this->char_translator_ != 0)
     {
-      return this->char_translator_->read_string (*this, 
+      return this->char_translator_->read_string (*this,
                                                   x);
     }
 
diff --git a/ace/Message_Block.cpp b/ace/Message_Block.cpp
index 4d8d3473d5a..335004f9768 100644
--- a/ace/Message_Block.cpp
+++ b/ace/Message_Block.cpp
@@ -513,6 +513,89 @@ ACE_Message_Block::ACE_Message_Block (ACE_Data_Block *data_block,
                 ACE_LIB_TEXT ("ACE_Message_Block")));
 }
 
+ACE_Message_Block::ACE_Message_Block (const ACE_Message_Block &mb,
+                                      size_t align)
+  :flags_ (0),
+   data_block_ (0)
+{
+  ACE_TRACE ("ACE_Message_Block::ACE_Message_Block");
+
+  if (ACE_BIT_DISABLED (mb.flags_,
+                        ACE_Message_Block::DONT_DELETE))
+    {
+      if (this->init_i (0,         // size
+                        MB_NORMAL, // type
+                        0,         // cont
+                        0,         // data
+                        0,         // allocator
+                        0,         // locking strategy
+                        0,         // flags
+                        0,         // priority
+                        ACE_Time_Value::zero,     // execution time
+                        ACE_Time_Value::max_time, // absolute time of deadline
+                        mb.data_block ()->duplicate (), // data block
+                        mb.data_block ()->data_block_allocator (),
+                        mb.message_block_allocator_) == -1)
+        ACE_ERROR ((LM_ERROR,
+                    ACE_LIB_TEXT ("ACE_Message_Block")));
+
+      // Align ourselves
+      char *start = ACE_ptr_align_binary (this->base (),
+                                          align);
+      // Set our rd & wr pointers
+      this->rd_ptr (start);
+      this->wr_ptr (start);
+
+    }
+  else
+    {
+      if (this->init_i (0,         // size
+                        MB_NORMAL, // type
+                        0,         // cont
+                        0,         // data
+                        0,         // allocator
+                        0,         // locking strategy
+                        0,         // flags
+                        0,         // priority
+                        ACE_Time_Value::zero,     // execution time
+                        ACE_Time_Value::max_time, // absolute time of deadline
+                        mb.data_block ()->clone_nocopy (),// data block
+                        mb.data_block ()->data_block_allocator (),
+                        mb.message_block_allocator_) == -1)
+        ACE_ERROR ((LM_ERROR,
+                    ACE_LIB_TEXT ("ACE_Message_Block")));
+
+      // Align ourselves
+      char *start = ACE_ptr_align_binary (this->base (),
+                                          align);
+      // Set our rd & wr pointers
+      this->rd_ptr (start);
+      this->wr_ptr (start);
+
+      // Get the alignment offset of the incoming ACE_Message_Block
+      start = ACE_ptr_align_binary (mb.base (),
+                                    align);
+
+
+      // Actual offset for the incoming message block assuming that it
+      // is also aligned to the same "align" byte
+      size_t wr_offset = mb.wr_ptr_ - (start - mb.base ());
+
+      // Copy wr_offset amount of data in to <this->data_block>
+      (void) ACE_OS::memcpy (this->wr_ptr (),
+                             start,
+                             wr_offset);
+
+      // Dont move the write pointer, just leave it to the application
+      // to do what it wants
+
+    }
+
+
+
+
+}
+
 int
 ACE_Message_Block::init_i (size_t size,
                            ACE_Message_Type msg_type,
diff --git a/ace/Message_Block.h b/ace/Message_Block.h
index b14dac673ac..6b7c0042299 100644
--- a/ace/Message_Block.h
+++ b/ace/Message_Block.h
@@ -177,6 +177,20 @@ public:
                      ACE_Allocator *message_block_allocator = 0);
 
   /**
+   * A copy constructor. This constructor is a bit different. If the
+   * incoming Message Block has a data block from the stack this
+   * constructor does a deep copy ie. allocates a new data block on
+   * the heap and does a copy of the data from the incoming message
+   * block. As a final note, the alignment information is used to
+   * align the data block if it is created afresh. If the incoming
+   * <mb> has a data block has a data block allocated from the heap,
+   * then this constructor just duplicates (ie. a shallow copy) the
+   * data block of the incoming <mb>.
+   */
+  ACE_Message_Block (const ACE_Message_Block &mb,
+                     size_t align);
+
+  /**
    * Create a Message Block that assumes ownership of <data> (i.e.,
    * doesn't delete it since it didn't malloc it!).  Note that the
    * <size> of the <Message_Block> will be <size>, but the <length>
@@ -291,6 +305,7 @@ public:
   /// Return a "shallow" copy that increments our reference count by 1.
   ACE_Message_Block *duplicate (void) const;
 
+
   /**
    * Return a "shallow" copy that increments our reference count by 1.
    * This is similar to CORBA's <_duplicate> method, which is useful
@@ -299,6 +314,7 @@ public:
    */
   static ACE_Message_Block *duplicate (const ACE_Message_Block *mb);
 
+
   /**
    * Decrease the shared ACE_Data_Block's reference count by 1.  If the
    * ACE_Data_Block's reference count goes to 0, it is deleted.
diff --git a/ace/Reactor.cpp b/ace/Reactor.cpp
index 312f6d92aa1..041f822ad00 100644
--- a/ace/Reactor.cpp
+++ b/ace/Reactor.cpp
@@ -1,5 +1,6 @@
 // $Id$
 
+
 #include "ace/Reactor.h"
 #include "ace/Reactor_Impl.h"
 #include "ace/Handle_Set.h"
@@ -20,6 +21,7 @@
 
 ACE_RCSID(ace, Reactor, "$Id$")
 
+
 ACE_ALLOC_HOOK_DEFINE(ACE_Reactor)
 
 ACE_Reactor::ACE_Reactor (ACE_Reactor_Impl *impl,
@@ -264,6 +266,13 @@ ACE_Reactor::reset_reactor_event_loop (void)
   this->implementation_->deactivate (0);
 }
 
+
+int
+ACE_Reactor::resumable_handler (void)
+{
+  return this->implementation ()->resumable_handler ();
+}
+
 void
 ACE_Reactor::dump (void) const
 {
@@ -271,4 +280,3 @@ ACE_Reactor::dump (void) const
 
   implementation_->dump ();
 }
-
diff --git a/ace/Reactor.h b/ace/Reactor.h
index 75a08edfdc7..5572ee705a7 100644
--- a/ace/Reactor.h
+++ b/ace/Reactor.h
@@ -397,6 +397,12 @@ public:
   /// Resume all <handles>.
   virtual int resume_handlers (void);
 
+  /// Does the reactor allow the application to resume the handle on
+  /// its own ie. can it pass on the control of handle resumption to
+  /// the application. A positive value indicates that the handlers
+  /// are application resumable. A value of 0 indicates otherwise.
+  virtual int resumable_handler (void);
+
   // = Timer management.
 
   /**
@@ -411,11 +417,11 @@ public:
    * @see reset_timer_interval()
    *
    * @param event_handler  event handler to schedule on reactor
-   * @param arg   argument passed to the handle_timeout() method of  event_handler  
+   * @param arg   argument passed to the handle_timeout() method of  event_handler
    * @param delta  time interval after which the timer will expire
    * @param interval  time interval after which the timer will be automatically rescheduled
    * @return -1 on failure, a timer_id value on success
-   */                          
+   */
   virtual long schedule_timer (ACE_Event_Handler *event_handler,
                                const void *arg,
                                const ACE_Time_Value &delay,
diff --git a/ace/Reactor_Impl.h b/ace/Reactor_Impl.h
index 5721d1f14c3..6dd4dcbe6b9 100644
--- a/ace/Reactor_Impl.h
+++ b/ace/Reactor_Impl.h
@@ -335,6 +335,11 @@ public:
   /// Resume all <handles>.
   virtual int resume_handlers (void) = 0;
 
+  /// Does the reactor allow the application to resume the handle on
+  /// its own ie. can it pass on the control of handle resumption to
+  /// the application
+  virtual int resumable_handler (void) = 0;
+
   /// Return 1 if we any event associations were made by the reactor
   /// for the handles that it waits on, 0 otherwise.
   virtual int uses_event_associations (void) = 0;
@@ -355,11 +360,11 @@ public:
    * @see reset_timer_interval()
    *
    * @param event_handler  event handler to schedule on reactor
-   * @param arg   argument passed to the handle_timeout() method of  event_handler  
+   * @param arg   argument passed to the handle_timeout() method of  event_handler
    * @param delta  time interval after which the timer will expire
    * @param interval  time interval after which the timer will be automatically rescheduled
    * @return -1 on failure, a timer_id value on success
-   */                          
+   */
   virtual long schedule_timer (ACE_Event_Handler *event_handler,
                                const void *arg,
                                const ACE_Time_Value &delay,
diff --git a/ace/Select_Reactor_Base.cpp b/ace/Select_Reactor_Base.cpp
index 3867d40b3bb..4eef29a9b6d 100644
--- a/ace/Select_Reactor_Base.cpp
+++ b/ace/Select_Reactor_Base.cpp
@@ -15,6 +15,7 @@
 
 ACE_RCSID(ace, Select_Reactor_Base, "$Id$")
 
+
 #if defined (ACE_WIN32)
 #define ACE_SELECT_REACTOR_HANDLE(H) (this->event_handlers_[(H)].handle_)
 #define ACE_SELECT_REACTOR_EVENT_HANDLER(THIS,H) ((THIS)->event_handlers_[(H)].event_handler_)
@@ -1014,6 +1015,15 @@ ACE_Select_Reactor_Impl::bit_ops (ACE_HANDLE handle,
   return omask;
 }
 
+int
+ACE_Select_Reactor_Impl::resumable_handler (void)
+{
+  // The select reactor has no handlers that can be resumed by the
+  // application. So return 0;
+
+  return 0;
+}
+
 #if defined (ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION)
 #if defined (ACE_HAS_REACTOR_NOTIFICATION_QUEUE)
 template class ACE_Unbounded_Queue <ACE_Notification_Buffer *>;
diff --git a/ace/Select_Reactor_Base.h b/ace/Select_Reactor_Base.h
index 071de0c6278..9274952caf4 100644
--- a/ace/Select_Reactor_Base.h
+++ b/ace/Select_Reactor_Base.h
@@ -404,6 +404,12 @@ public:
   virtual int purge_pending_notifications (ACE_Event_Handler * = 0,
                                            ACE_Reactor_Mask    = ACE_Event_Handler::ALL_EVENTS_MASK);
 
+  /// Does the reactor allow the application to resume the handle on
+  /// its own ie. can it pass on the control of handle resumption to
+  /// the application.  The select reactor has no handlers that can be
+  /// resumed by the  application. So return 0;
+  virtual int resumable_handler (void);
+
 protected:
   /// Allow manipulation of the <wait_set_> mask and <ready_set_> mask.
   virtual int bit_ops (ACE_HANDLE handle,
diff --git a/ace/TP_Reactor.cpp b/ace/TP_Reactor.cpp
index 24f12f1fe65..9b3881ea3b4 100644
--- a/ace/TP_Reactor.cpp
+++ b/ace/TP_Reactor.cpp
@@ -1,5 +1,6 @@
 // $Id$
 
+
 #include "ace/TP_Reactor.h"
 #include "ace/Reactor.h"
 #include "ace/Thread.h"
@@ -10,6 +11,7 @@
 
 ACE_RCSID(ace, TP_Reactor, "$Id$")
 
+
 ACE_ALLOC_HOOK_DEFINE (ACE_TP_Reactor)
 
 ACE_TP_Reactor::ACE_TP_Reactor (ACE_Sig_Handler *sh,
@@ -433,6 +435,12 @@ ACE_TP_Reactor::notify_handle (ACE_EH_Dispatch_Info &dispatch_info)
   return 0;
 }
 
+int
+ACE_TP_Reactor::resumable_handler (void)
+{
+  return 1;
+}
+
 
 ACE_EH_Dispatch_Info::ACE_EH_Dispatch_Info (void)
 {
diff --git a/ace/TP_Reactor.h b/ace/TP_Reactor.h
index dab47eedda7..65f85383058 100644
--- a/ace/TP_Reactor.h
+++ b/ace/TP_Reactor.h
@@ -151,6 +151,12 @@ public:
 
   virtual int handle_events (ACE_Time_Value &max_wait_time);
 
+  /// Does the reactor allow the application to resume the handle on
+  /// its own ie. can it pass on the control of handle resumption to
+  /// the application.  The TP reactor has can allow applications to
+  /// resume handles.  So return a +ve value.
+  virtual int resumable_handler (void);
+
   /// GET/SET/ADD/CLR the dispatch mask "bit" bound with the <eh> and
   /// <mask>.
   virtual int mask_ops (ACE_Event_Handler *eh,
diff --git a/ace/WFMO_Reactor.cpp b/ace/WFMO_Reactor.cpp
index 7039cde14ed..34325fcb333 100644
--- a/ace/WFMO_Reactor.cpp
+++ b/ace/WFMO_Reactor.cpp
@@ -1,5 +1,6 @@
 // $Id$
 
+
 #include "ace/WFMO_Reactor.h"
 
 #include "ace/Handle_Set.h"
@@ -12,6 +13,7 @@
 
 ACE_RCSID(ace, WFMO_Reactor, "$Id$")
 
+
 #if defined (ACE_WIN32) && !defined (ACE_HAS_WINCE)
 
 #include "ace/Auto_Ptr.h"
@@ -2480,6 +2482,14 @@ ACE_WFMO_Reactor::purge_pending_notifications (ACE_Event_Handler *eh,
   return this->notify_handler_->purge_pending_notifications (eh, mask);
 }
 
+int
+ACE_WFMO_Reactor::resumable_handler (void)
+{
+  ACE_TRACE ("ACE_WFMO_Reactor::resumable_handler");
+  return 0;
+}
+
+
 // No-op WinSOCK2 methods to help WFMO_Reactor compile
 #if !defined (ACE_HAS_WINSOCK2) || (ACE_HAS_WINSOCK2 == 0)
 int
diff --git a/ace/WFMO_Reactor.h b/ace/WFMO_Reactor.h
index 4cbc38daef3..a9fbec8ecc4 100644
--- a/ace/WFMO_Reactor.h
+++ b/ace/WFMO_Reactor.h
@@ -883,6 +883,12 @@ public:
   /// Resume all <handles>.
   virtual int resume_handlers (void);
 
+  /// Does the reactor allow the application to resume the handle on
+  /// its own ie. can it pass on the control of handle resumption to
+  /// the application. A positive value indicates that the handlers
+  /// are application resumable. A value of 0 indicates otherwise.
+  virtual int resumable_handler (void);
+
   /**
    * Return 1 if we any event associations were made by the reactor
    * for the handles that it waits on, 0 otherwise. Since the
@@ -905,11 +911,11 @@ public:
    * @see reset_timer_interval()
    *
    * @param event_handler  event handler to schedule on reactor
-   * @param arg   argument passed to the handle_timeout() method of  event_handler  
+   * @param arg   argument passed to the handle_timeout() method of  event_handler
    * @param delta  time interval after which the timer will expire
    * @param interval  time interval after which the timer will be automatically rescheduled
    * @return -1 on failure, a timer_id value on success
-   */                          
+   */
   virtual long schedule_timer (ACE_Event_Handler *event_handler,
                                const void *arg,
                                const ACE_Time_Value &delay,