Optimized protocol and transport objects to use pre-allocated buffers.

7 files changed, 49 insertions, 52 deletions

diff --git a/TAO/orbsvcs/orbsvcs/AV/RTCP.cpp b/TAO/orbsvcs/orbsvcs/AV/RTCP.cpp
index fa2469891fd..8b274be5021 100644
--- a/TAO/orbsvcs/orbsvcs/AV/RTCP.cpp
+++ b/TAO/orbsvcs/orbsvcs/AV/RTCP.cpp
@@ -587,7 +587,7 @@ TAO_AV_RTCP_Flow_Factory::init (int /* argc */,
   return 0;
 }
 
-TAO_AV_Protocol_Object* 
+TAO_AV_Protocol_Object*
 TAO_AV_RTCP_Flow_Factory::make_protocol_object (TAO_FlowSpec_Entry *entry,
                                                 TAO_Base_StreamEndPoint *endpoint,
                                                 TAO_AV_Flow_Handler *handler,
@@ -742,7 +742,7 @@ TAO_AV_RTCP_Callback::schedule (int ms)
 int
 TAO_AV_RTCP_Callback::handle_start (void)
 {
-  // 
+  //
   /*
    * schedule a timer for our first report using half the
    * min rtcp interval.  This gives us some time before
@@ -806,15 +806,16 @@ TAO_AV_RTCP_Callback::receive_frame (ACE_Message_Block *frame,
 {
   char *buf = frame->rd_ptr ();
   TAO_AV_RTP::rtphdr *rh = (TAO_AV_RTP::rtphdr *)buf;
-  char *rd_ptr = frame->rd_ptr ()+sizeof (TAO_AV_RTP::rtphdr);
-  frame->rd_ptr (rd_ptr);
+
+  frame->rd_ptr (sizeof (TAO_AV_RTP::rtphdr));
   int result = this->demux (rh,
                             frame,
                             peer_address);
+  frame->rd_ptr (buf);
+
   if (result < 0)
     return result;
-  rd_ptr = frame->rd_ptr ()-sizeof (TAO_AV_RTP::rtphdr);
-  frame->rd_ptr (rd_ptr);
+
   return 0;
 }
 
diff --git a/TAO/orbsvcs/orbsvcs/AV/RTP.cpp b/TAO/orbsvcs/orbsvcs/AV/RTP.cpp
index eff4967d050..dfa53e76c64 100644
--- a/TAO/orbsvcs/orbsvcs/AV/RTP.cpp
+++ b/TAO/orbsvcs/orbsvcs/AV/RTP.cpp
@@ -89,7 +89,7 @@ TAO_AV_RTP::write_header (rtphdr &header,
 }
 
 // TAO_AV_RTP_Object
-           
+
 int
 TAO_AV_RTP_Object::handle_input (void)
 {
@@ -97,35 +97,30 @@ TAO_AV_RTP_Object::handle_input (void)
 
   // Handles the incoming RTP packet input.
 
-//   size_t bufsiz = 2*this->transport_->mtu ();
-//   ACE_NEW_RETURN (data,
-//                   ACE_Message_Block (bufsiz),
-//                   -1);
-  int n = this->transport_->recv (this->data_->rd_ptr (),this->data_->size ());
+  this->frame_.rd_ptr (this->frame_.base ());
+
+  int n = this->transport_->recv (this->frame_.rd_ptr (),
+                                  this->frame_.size ());
   if (n == 0)
     ACE_ERROR_RETURN ( (LM_ERROR,"TAO_AV_RTP::handle_input:connection closed\n"),-1);
   if (n < 0)
     ACE_ERROR_RETURN ( (LM_ERROR,"TAO_AV_RTP::handle_input:recv error\n"),-1);
-  char *wr_ptr = this->data_->rd_ptr ()+n;
-  this->data_->wr_ptr (wr_ptr);
+  this->frame_.wr_ptr (this->frame_.rd_ptr () + n);
   ACE_Addr *addr = this->transport_->get_peer_addr ();
 
-  int result = TAO_AV_RTP::handle_input (this->data_,
+  ACE_Message_Block* data = &this->frame_;
+  int result = TAO_AV_RTP::handle_input (data,
                                          frame_info);
   if (result < 0)
     return 0;
-  result = this->control_object_->handle_control_input (this->data_,
-                                                        *addr);
-  if (result < 0)
-    return 0;
+  this->control_object_->handle_control_input (data,
+                                               *addr);
 
-  char *rd_ptr = this->data_->rd_ptr ()+sizeof (rtphdr);
-  this->data_->rd_ptr (rd_ptr);
-  result = this->callback_->receive_frame (this->data_,
+  this->frame_.rd_ptr (sizeof(rtphdr));
+  result = this->callback_->receive_frame (&this->frame_,
                                            frame_info,
                                            *addr);
-  rd_ptr = this->data_->rd_ptr ()-sizeof (rtphdr);
-  this->data_->rd_ptr (rd_ptr);
+
   return 0;
 }
 
@@ -214,13 +209,11 @@ TAO_AV_RTP_Object::TAO_AV_RTP_Object (TAO_AV_Callback *callback,
                                       TAO_AV_Transport *transport)
   :TAO_AV_Protocol_Object (callback,transport),
    sequence_num_ (0),
-   control_object_ (0),
-   data_ (0)
+   control_object_ (0)
 {
   this->sequence_num_ = ACE_OS::rand ();
-  // Allocate a static message block.
-  ACE_NEW (data_,
-           ACE_Message_Block (2*transport->mtu ()));
+
+  this->frame_.size (2 * this->transport_->mtu ());
 }
 
 TAO_AV_RTP_Object::~TAO_AV_RTP_Object (void)
@@ -306,7 +299,7 @@ TAO_AV_RTP_Flow_Factory::init (int /* argc */,
   return 0;
 }
 
-TAO_AV_Protocol_Object* 
+TAO_AV_Protocol_Object*
 TAO_AV_RTP_Flow_Factory::make_protocol_object (TAO_FlowSpec_Entry *entry,
                                                TAO_Base_StreamEndPoint *endpoint,
                                                TAO_AV_Flow_Handler *handler,
diff --git a/TAO/orbsvcs/orbsvcs/AV/RTP.h b/TAO/orbsvcs/orbsvcs/AV/RTP.h
index 7ab37bcd5e0..a6de439f594 100644
--- a/TAO/orbsvcs/orbsvcs/AV/RTP.h
+++ b/TAO/orbsvcs/orbsvcs/AV/RTP.h
@@ -267,12 +267,15 @@ public:
   virtual int destroy (void);
   virtual int set_policies (const TAO_AV_PolicyList &policy_list);
   virtual void control_object (TAO_AV_Protocol_Object *object);
+
 protected:
   ACE_UINT16 sequence_num_;
   int format_;
   CORBA::ULong ssrc_;
   TAO_AV_Protocol_Object *control_object_;
-  ACE_Message_Block *data_;
+
+  ACE_Message_Block frame_;
+  // Pre-allocated memory to receive the data...
 };
 
 class TAO_ORBSVCS_Export TAO_AV_RTP_Flow_Factory : public TAO_AV_Flow_Protocol_Factory
diff --git a/TAO/orbsvcs/orbsvcs/AV/TCP.cpp b/TAO/orbsvcs/orbsvcs/AV/TCP.cpp
index 54232f6b7cc..88544071641 100644
--- a/TAO/orbsvcs/orbsvcs/AV/TCP.cpp
+++ b/TAO/orbsvcs/orbsvcs/AV/TCP.cpp
@@ -200,20 +200,15 @@ TAO_AV_TCP_Factory::make_connector (void)
 int
 TAO_AV_TCP_Object::handle_input (void)
 {
-  size_t size = BUFSIZ;
-  ACE_Message_Block *frame = 0;
-  ACE_NEW_RETURN (frame,
-                  ACE_Message_Block (size),
-                  -1);
-  int n = this->transport_->recv (frame->rd_ptr (),
-                                  size);
+  int n = this->transport_->recv (this->frame_.rd_ptr (),
+                                  this->frame_.size ());
   if (n == -1)
     ACE_ERROR_RETURN ((LM_ERROR,"TAO_AV_TCP_Flow_Handler::handle_input recv failed\n"),-1);
   if (n == 0)
     ACE_ERROR_RETURN ((LM_ERROR,"TAO_AV_TCP_Flow_Handler::handle_input connection closed\n"),-1);
-  frame->wr_ptr (n);
+  this->frame_.wr_ptr (this->frame_.rd_ptr () + n);
 
-  return this->callback_->receive_frame (frame);
+  return this->callback_->receive_frame (&this->frame_);
 }
 
 int
@@ -226,7 +221,7 @@ TAO_AV_TCP_Object::send_frame (ACE_Message_Block *frame,
   return 0;
 }
 
-int 
+int
 TAO_AV_TCP_Object::send_frame (const iovec *iov,
                                int iovcnt,
                                TAO_AV_frame_info *frame_info)
@@ -238,6 +233,8 @@ TAO_AV_TCP_Object::TAO_AV_TCP_Object (TAO_AV_Callback *callback,
                                       TAO_AV_Transport *transport)
   :TAO_AV_Protocol_Object (callback,transport)
 {
+  // @@ Is this a good size?
+  this->frame_.size (BUFSIZ);
 }
 
 TAO_AV_TCP_Object::~TAO_AV_TCP_Object (void)
@@ -485,7 +482,7 @@ TAO_AV_TCP_Acceptor::make_svc_handler (TAO_AV_TCP_Flow_Handler *&tcp_handler)
 //       this->endpoint_->set_protocol_object (this->flowname_.c_str (),
 //                                             object);
       this->endpoint_->set_handler (this->flowname_.c_str (),tcp_handler);
-      this->entry_->protocol_object (object);      
+      this->entry_->protocol_object (object);
       this->entry_->handler (tcp_handler);
     }
   return 0;
@@ -629,7 +626,7 @@ TAO_AV_TCP_Flow_Handler::handle_input (ACE_HANDLE /*fd*/)
 }
 
 int
-TAO_AV_TCP_Flow_Handler::handle_timeout (const ACE_Time_Value &tv, 
+TAO_AV_TCP_Flow_Handler::handle_timeout (const ACE_Time_Value &tv,
                                                const void *arg)
 {
   return TAO_AV_Flow_Handler::handle_timeout (tv,arg);
diff --git a/TAO/orbsvcs/orbsvcs/AV/TCP.h b/TAO/orbsvcs/orbsvcs/AV/TCP.h
index 3ddc3e24f20..da9b2d37c04 100644
--- a/TAO/orbsvcs/orbsvcs/AV/TCP.h
+++ b/TAO/orbsvcs/orbsvcs/AV/TCP.h
@@ -208,6 +208,9 @@ public:
   virtual int destroy (void);
   // end the stream.
 
+private:
+  ACE_Message_Block frame_;
+  // Pre-allocated memory to receive the data...
 };
 
 class TAO_AV_TCP_Flow_Factory : public TAO_AV_Flow_Protocol_Factory
diff --git a/TAO/orbsvcs/orbsvcs/AV/UDP.cpp b/TAO/orbsvcs/orbsvcs/AV/UDP.cpp
index c60fc0bb97e..fc7b0b42bb0 100644
--- a/TAO/orbsvcs/orbsvcs/AV/UDP.cpp
+++ b/TAO/orbsvcs/orbsvcs/AV/UDP.cpp
@@ -608,20 +608,15 @@ TAO_AV_UDP_Factory::init (int /* argc */,
 int
 TAO_AV_UDP_Object::handle_input (void)
 {
-  size_t size = 2*this->transport_->mtu ();
-  ACE_Message_Block *frame = 0;
-  ACE_NEW_RETURN (frame,
-                  ACE_Message_Block (size),
-                  -1);
-  int n = this->transport_->recv (frame->rd_ptr (),
-                                  frame->size ());
+  int n = this->transport_->recv (this->frame_.rd_ptr (),
+                                  this->frame_.size ());
   if (n == -1)
     ACE_ERROR_RETURN ((LM_ERROR,"TAO_AV_UDP_Flow_Handler::handle_input recv failed\n"),-1);
   if (n == -1)
     ACE_ERROR_RETURN ((LM_ERROR,"TAO_AV_UDP_Flow_Handler::handle_input connection closed\n"),-1);
-  frame->wr_ptr (n);
+  this->frame_.wr_ptr (this->frame_.rd_ptr () + n);
 
-  return this->callback_->receive_frame (frame);
+  return this->callback_->receive_frame (&this->frame_);
 }
 
 int
@@ -647,6 +642,7 @@ TAO_AV_UDP_Object::TAO_AV_UDP_Object (TAO_AV_Callback *callback,
                                       TAO_AV_Transport *transport)
   :TAO_AV_Protocol_Object (callback,transport)
 {
+  this->frame_.size (2 * this->transport_->mtu ());
 }
 
 TAO_AV_UDP_Object::~TAO_AV_UDP_Object (void)
diff --git a/TAO/orbsvcs/orbsvcs/AV/UDP.h b/TAO/orbsvcs/orbsvcs/AV/UDP.h
index c009779e651..f15031ff9e3 100644
--- a/TAO/orbsvcs/orbsvcs/AV/UDP.h
+++ b/TAO/orbsvcs/orbsvcs/AV/UDP.h
@@ -185,6 +185,10 @@ public:
 
   virtual int destroy (void);
   // end the stream.
+
+private:
+  ACE_Message_Block frame_;
+  // Pre-allocated memory to receive the data...
 };
 
 class TAO_AV_UDP_Flow_Factory : public TAO_AV_Flow_Protocol_Factory