TAO_Strategies: TAO_SHMIOP_Transport class Reference

00026   : TAO_Transport (TAO_TAG_SHMEM_PROFILE,
00027                    orb_core),
00028     connection_handler_ (handler),
00029     messaging_object_ (0)
00030 {
00031 /*
00032  * Hook to customize the messaging object when the concrete messaging
00033  * object is known a priori. In this case, the flag is ignored.
00034  */
00035 //@@ MESSAGING_SPL_COMMENT_HOOK_START
00036   ACE_NEW (this->messaging_object_,
00037             TAO_GIOP_Message_Base (orb_core, this));
00038 //@@ MESSAGING_SPL_COMMENT_HOOK_END
00039 }

00042 {
00043   delete this->messaging_object_;
00044 }

00054 {
00055   return this->connection_handler_;
00056 }

00048 {
00049   return this->connection_handler_;
00050 }

00131 {
00132   if (TAO_debug_level > 3)
00133     {
00134       ACE_DEBUG ((LM_DEBUG,
00135                   "TAO (%P|%t) - SHMIOP_Transport[%d]::handle_input\n",
00136                   this->id ()));
00137     }
00138 
00139    // The buffer on the stack which will be used to hold the input
00140   // messages, compensate shrink due to alignment
00141   char buf [TAO_MAXBUFSIZE + ACE_CDR::MAX_ALIGNMENT];
00142 
00143 
00144 #if defined (ACE_INITIALIZE_MEMORY_BEFORE_USE)
00145   (void) ACE_OS::memset (buf,
00146                          '\0',
00147                          sizeof buf);
00148 #endif /* ACE_INITIALIZE_MEMORY_BEFORE_USE */
00149 
00150   // Create a data block
00151   ACE_Data_Block db (sizeof (buf),
00152                      ACE_Message_Block::MB_DATA,
00153                      buf,
00154                      this->orb_core_->input_cdr_buffer_allocator (),
00155                      this->orb_core_->locking_strategy (),
00156                      ACE_Message_Block::DONT_DELETE,
00157                      this->orb_core_->input_cdr_dblock_allocator ());
00158 
00159   // Create a message block
00160   ACE_Message_Block message_block (&db,
00161                                    ACE_Message_Block::DONT_DELETE,
00162                                    this->orb_core_->input_cdr_msgblock_allocator ());
00163 
00164 
00165   // Align the message block
00166   ACE_CDR::mb_align (&message_block);
00167 
00168   const size_t missing_header_data = this->messaging_object ()->header_length ();
00169 
00170   if (missing_header_data == 0)
00171     {
00172       return -1;
00173     }
00174 
00175   // .. do a read on the socket again.
00176   ssize_t bytes = 0;
00177 
00178   // As this used for transports where things are available in one
00179   // shot this looping should not create any problems.
00180   for (size_t m = missing_header_data;
00181        m != 0;
00182        m -= bytes)
00183     {
00184       bytes = 0; // reset
00185 
00186       // We would have liked to use something like a recv_n ()
00187       // here. But at the time when the code was written, the MEM_Stream
00188       // classes had poor support  for recv_n (). Till a day when we
00189       // get proper recv_n (), let us stick with this. The other
00190       // argument that can be said against this is that, this is the
00191       // bad layer in which this is being done ie. recv_n is
00192       // simulated. But...
00193       bytes = this->recv (message_block.wr_ptr (),
00194                           m,
00195                           max_wait_time);
00196 
00197       if (bytes == 0 ||
00198           bytes == -1)
00199         {
00200           return -1;
00201         }
00202 
00203       message_block.wr_ptr (bytes);
00204     }
00205 
00206     TAO_Queued_Data qd (&message_block);
00207     size_t mesg_length = 0;
00208 
00209     // Parse the incoming message for validity. The check needs to be
00210     // performed by the messaging objects.
00211     if (this->messaging_object ()->parse_next_message (qd,
00212                                                        mesg_length) == -1)
00213       return -1;
00214 
00215     if (qd.missing_data () == TAO_MISSING_DATA_UNDEFINED)
00216       {
00217         // parse/marshal error happened
00218         return -1;
00219       }
00220 
00221     if (message_block.length () > mesg_length)
00222       {
00223         // we read too much data
00224         return -1;
00225       }
00226 
00227     if (message_block.space () < qd.missing_data ())
00228       {
00229         size_t const message_size = message_block.length ()
00230                                   + qd.missing_data ();
00231 
00232         // reallocate buffer with correct size on heap
00233         if (ACE_CDR::grow (&message_block, message_size) == -1)
00234           {
00235             if (TAO_debug_level > 0)
00236               {
00237                 ACE_ERROR ((LM_ERROR,
00238                  "TAO (%P|%t) - SHMIOP_Transport[%d]::handle_input, "
00239                  "error growing message buffer\n",
00240                  this->id () ));
00241               }
00242             return -1;
00243           }
00244 
00245       }
00246 
00247     // As this used for transports where things are available in one
00248     // shot this looping should not create any problems.
00249     for (size_t n = qd.missing_data ();
00250        n != 0;
00251        n -= bytes)
00252     {
00253         bytes = 0; // reset
00254 
00255       // We would have liked to use something like a recv_n ()
00256       // here. But at the time when the code was written, the MEM_Stream
00257       // classes had poor support  for recv_n (). Till a day when we
00258       // get proper recv_n (), let us stick with this. The other
00259       // argument that can be said against this is that, this is the
00260       // bad layer in which this is being done ie. recv_n is
00261       // simulated. But...
00262       bytes = this->recv (message_block.wr_ptr (),
00263                           n,
00264                           max_wait_time);
00265 
00266       if (bytes == 0 ||
00267           bytes == -1)
00268         {
00269           return -1;
00270         }
00271 
00272       message_block.wr_ptr (bytes);
00273     }
00274 
00275   qd.missing_data (0);
00276 
00277   // Now we have a full message in our buffer. Just go ahead and
00278   // process that
00279   if (this->process_parsed_messages (&qd, rh) == -1)
00280     {
00281       return -1;
00282     }
00283 
00284   return 0;
00285 }

00349 {
00350   this->messaging_object_->init (major, minor);
00351   return 1;
00352 }

00060 {
00061   return this->messaging_object_;
00062 }

00089 {
00090   ssize_t n = 0;
00091 
00092   int read_break = 0;
00093 
00094   while (!read_break)
00095     {
00096       n = this->connection_handler_->peer ().recv (buf,
00097                                                    len,
00098                                                    max_wait_time);
00099 
00100       // If we get a EWOULBLOCK we try to read again.
00101       if (n == -1 && (errno == EAGAIN || errno == EWOULDBLOCK))
00102         {
00103           n = 0;
00104           continue;
00105         }
00106 
00107       // If there is anything else we just drop out of the loop.
00108       read_break = 1;
00109     }
00110 
00111   if (n == -1)
00112     {
00113       if (TAO_debug_level > 3 && errno != ETIME)
00114         {
00115           ACE_DEBUG ((LM_DEBUG,
00116                       ACE_TEXT ("TAO (%P|%t) - %p \n"),
00117                       ACE_TEXT ("TAO - read message failure ")
00118                       ACE_TEXT ("recv_i () \n")));
00119         }
00120     }
00121   else if (n == 0)
00122     {
00123       n = -1;
00124     }
00125   return n;
00126 }

00069 {
00070   bytes_transferred = 0;
00071   for (int i = 0; i < iovcnt; ++i)
00072     {
00073       ssize_t retval =
00074         this->connection_handler_->peer ().send (iov[i].iov_base,
00075                                                  iov[i].iov_len,
00076                                                  max_wait_time);
00077       if (retval > 0)
00078         bytes_transferred += retval;
00079       if (retval <= 0)
00080         return retval;
00081     }
00082   return bytes_transferred;
00083 }

00316 {
00317   // Format the message in the stream first
00318   if (this->messaging_object_->format_message (stream) != 0)
00319     return -1;
00320 
00321   // Strictly speaking, should not need to loop here because the
00322   // socket never gets set to a nonblocking mode ... some Linux
00323   // versions seem to need it though.  Leaving it costs little.
00324 
00325   // This guarantees to send all data (bytes) or return an error.
00326   ssize_t n = this->send_message_shared (stub,
00327                                          message_semantics,
00328                                          stream.begin (),
00329                                          max_wait_time);
00330 
00331   if (n == -1)
00332     {
00333       if (TAO_debug_level)
00334         ACE_DEBUG ((LM_DEBUG,
00335                     ACE_TEXT ("TAO: (%P|%t|%N|%l) closing transport %d after fault %p\n"),
00336                     this->id (),
00337                     ACE_TEXT ("send_message ()\n")));
00338 
00339       return -1;
00340     }
00341 
00342   return 1;
00343 }

00295 {
00296   if (this->ws_->sending_request (orb_core,
00297                                   message_semantics) == -1)
00298     return -1;
00299 
00300   if (this->send_message (stream,
00301                           stub,
00302                           message_semantics,
00303                           max_wait_time) == -1)
00304 
00305     return -1;
00306   this->first_request_sent();
00307 
00308   return 0;
00309 }


Public Member Functions
	TAO_SHMIOP_Transport (TAO_SHMIOP_Connection_Handler handler, TAO_ORB_Core orb_core)
	Constructor.
	~TAO_SHMIOP_Transport (void)
	Default destructor.
virtual int	send_request (TAO_Stub stub, TAO_ORB_Core orb_core, TAO_OutputCDR &stream, int message_semantics, ACE_Time_Value *max_wait_time)
virtual int	send_message (TAO_OutputCDR &stream, TAO_Stub stub=0, int message_semantics=TAO_Transport::TAO_TWOWAY_REQUEST, ACE_Time_Value max_time_wait=0)
virtual int	messaging_init (CORBA::Octet major, CORBA::Octet minor)
	Initialising the messaging object.
Protected Member Functions
Overridden Template Methods
These are implementations of template methods declared by TAO_Transport.
virtual ACE_Event_Handler *	event_handler_i (void)
virtual TAO_Connection_Handler *	connection_handler_i (void)
virtual TAO_Pluggable_Messaging *	messaging_object (void)
virtual ssize_t	send (iovec iov, int iovcnt, size_t &bytes_transferred, const ACE_Time_Value timeout=0)
	Write the complete Message_Block chain to the connection.
virtual ssize_t	recv (char buf, size_t len, const ACE_Time_Value s=0)
	Read len bytes from into buf.
virtual int	handle_input (TAO_Resume_Handle &rh, ACE_Time_Value *max_wait_time)
Private Attributes
TAO_SHMIOP_Connection_Handler *	connection_handler_
TAO_Pluggable_Messaging *	messaging_object_
	Our messaging object.

TAO_SHMIOP_Transport Class Reference

Public Member Functions

Protected Member Functions

Private Attributes

Detailed Description

Constructor & Destructor Documentation

Member Function Documentation

Member Data Documentation