TAO_Strategies: TAO_SHMIOP_Transport class Reference

00028   : TAO_Transport (TAO_TAG_SHMEM_PROFILE,
00029                    orb_core),
00030     connection_handler_ (handler),
00031     messaging_object_ (0)
00032 {
00033 /*
00034  * Hook to customize the messaging object when the concrete messaging
00035  * object is known a priori. In this case, the flag is ignored.
00036  */
00037 //@@ MESSAGING_SPL_COMMENT_HOOK_START
00038   if (flag)
00039     {
00040       // Use the lite version of the protocol
00041       ACE_NEW (this->messaging_object_,
00042                TAO_GIOP_Message_Lite (orb_core));
00043     }
00044   else
00045     {
00046       // Use the normal GIOP object
00047       ACE_NEW (this->messaging_object_,
00048                TAO_GIOP_Message_Base (orb_core, this));
00049     }
00050 //@@ MESSAGING_SPL_COMMENT_HOOK_END
00051 }

00054 {
00055   delete this->messaging_object_;
00056 }

00066 {
00067   return this->connection_handler_;
00068 }

00060 {
00061   return this->connection_handler_;
00062 }

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

00364 {
00365   this->messaging_object_->init (major,
00366                                  minor);
00367   return 1;
00368 }

00072 {
00073   return this->messaging_object_;
00074 }

00101 {
00102   ssize_t n = 0;
00103 
00104   int read_break = 0;
00105 
00106   while (!read_break)
00107     {
00108       n = this->connection_handler_->peer ().recv (buf,
00109                                                    len,
00110                                                    max_wait_time);
00111 
00112       // If we get a EWOULBLOCK we try to read again.
00113       if (n == -1 && (errno == EAGAIN || errno == EWOULDBLOCK))
00114         {
00115           n = 0;
00116           continue;
00117         }
00118 
00119       // If there is anything else we just drop out of the loop.
00120       read_break = 1;
00121     }
00122 
00123   if (n == -1)
00124     {
00125       if (TAO_debug_level > 3 && errno != ETIME)
00126         {
00127           ACE_DEBUG ((LM_DEBUG,
00128                       ACE_TEXT ("TAO (%P|%t) - %p \n"),
00129                       ACE_TEXT ("TAO - read message failure ")
00130                       ACE_TEXT ("recv_i () \n")));
00131         }
00132     }
00133   else if (n == 0)
00134     {
00135       n = -1;
00136     }
00137   return n;
00138 }

00081 {
00082   bytes_transferred = 0;
00083   for (int i = 0; i < iovcnt; ++i)
00084     {
00085       ssize_t retval =
00086         this->connection_handler_->peer ().send (iov[i].iov_base,
00087                                                  iov[i].iov_len,
00088                                                  max_wait_time);
00089       if (retval > 0)
00090         bytes_transferred += retval;
00091       if (retval <= 0)
00092         return retval;
00093     }
00094   return bytes_transferred;
00095 }

00331 {
00332   // Format the message in the stream first
00333   if (this->messaging_object_->format_message (stream) != 0)
00334     return -1;
00335 
00336   // Strictly speaking, should not need to loop here because the
00337   // socket never gets set to a nonblocking mode ... some Linux
00338   // versions seem to need it though.  Leaving it costs little.
00339 
00340   // This guarantees to send all data (bytes) or return an error.
00341   ssize_t n = this->send_message_shared (stub,
00342                                          message_semantics,
00343                                          stream.begin (),
00344                                          max_wait_time);
00345 
00346   if (n == -1)
00347     {
00348       if (TAO_debug_level)
00349         ACE_DEBUG ((LM_DEBUG,
00350                     ACE_TEXT ("TAO: (%P|%t|%N|%l) closing transport %d after fault %p\n"),
00351                     this->id (),
00352                     ACE_TEXT ("send_message ()\n")));
00353 
00354       return -1;
00355     }
00356 
00357   return 1;
00358 }

00310 {
00311   if (this->ws_->sending_request (orb_core,
00312                                   message_semantics) == -1)
00313     return -1;
00314 
00315   if (this->send_message (stream,
00316                           stub,
00317                           message_semantics,
00318                           max_wait_time) == -1)
00319 
00320     return -1;
00321   this->first_request_sent();
00322 
00323   return 0;
00324 }


Public Member Functions
	TAO_SHMIOP_Transport (TAO_SHMIOP_Connection_Handler handler, TAO_ORB_Core orb_core, CORBA::Boolean flag)
	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, int block=0)
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