TAO/tao/IIOP_Transport.cpp

   1 #include "tao/IIOP_Transport.h"
   2
   3 #if defined (TAO_HAS_IIOP) && (TAO_HAS_IIOP != 0)
   4
   5 #include "tao/IIOP_Acceptor.h"
   6 #include "tao/IIOPC.h"
   7 #include "tao/Acceptor_Registry.h"
   8 #include "tao/operation_details.h"
   9 #include "tao/Wait_Strategy.h"
  10 #include "tao/debug.h"
  11 #include "tao/GIOP_Message_Base.h"
  12 #include "tao/Protocols_Hooks.h"
  13 #include "tao/ORB_Core.h"
  14 #include "tao/Thread_Lane_Resources.h"
  15 #include "tao/Transport_Mux_Strategy.h"
  16 #include "tao/MMAP_Allocator.h"
  17
  18 #include "ace/OS_NS_sys_sendfile.h"
  19
  20 TAO_BEGIN_VERSIONED_NAMESPACE_DECL
  21
  22 TAO_IIOP_Transport::TAO_IIOP_Transport (TAO_IIOP_Connection_Handler *handler,
  23                                         TAO_ORB_Core *orb_core)
  24   : TAO_Transport (IOP::TAG_INTERNET_IOP,
  25                    orb_core)
  26   , connection_handler_ (handler)
  27 {
  28 }
  29
  30 TAO_IIOP_Transport::~TAO_IIOP_Transport ()
  31 {
  32 }
  33
  34 ACE_Event_Handler *
  35 TAO_IIOP_Transport::event_handler_i ()
  36 {
  37   return this->connection_handler_;
  38 }
  39
  40 TAO_Connection_Handler *
  41 TAO_IIOP_Transport::connection_handler_i ()
  42 {
  43   return this->connection_handler_;
  44 }
  45
  46 ssize_t
  47 TAO_IIOP_Transport::send (iovec *iov, int iovcnt,
  48                           size_t &bytes_transferred,
  49                           const ACE_Time_Value *max_wait_time)
  50 {
  51   ssize_t const retval =
  52     this->connection_handler_->peer ().sendv (iov,
  53                                               iovcnt,
  54                                               max_wait_time);
  55   if (retval > 0)
  56     bytes_transferred = retval;
  57   else
  58     {
  59       if (TAO_debug_level > 4)
  60         {
  61           TAOLIB_DEBUG ((LM_DEBUG,
  62                       ACE_TEXT ("TAO (%P|%t) - IIOP_Transport[%d]::send, ")
  63                       ACE_TEXT ("send failure (errno: %d) - %m\n"),
  64                       this->id (), ACE_ERRNO_GET));
  65         }
  66     }
  67
  68   return retval;
  69 }
  70
  71 #if TAO_HAS_SENDFILE == 1
  72 ssize_t
  73 TAO_IIOP_Transport::sendfile (TAO_MMAP_Allocator * allocator,
  74                               iovec * iov,
  75                               int iovcnt,
  76                               size_t &bytes_transferred,
  77                               TAO::Transport::Drain_Constraints const & dc)
  78 {
  79   // @@ We should probably set the TCP_CORK socket option to minimize
  80   //    network operations.  It may also be useful to adjust the
  81   //    socket send buffer size accordingly.
  82
  83   // If we don't have an allocator, fallback to the regular way of sending
  84   // data
  85   if (allocator == nullptr)
  86     return this->send (iov, iovcnt, bytes_transferred, this->io_timeout(dc));
  87
  88   // We can only use sendfile when all data is coming from the mmap allocator,
  89   // if not, we just fallback to to the regular way of sending data
  90   iovec * const off_check_begin = iov;
  91   iovec * const off_check_end   = iov + iovcnt;
  92   for (iovec * index = off_check_begin; index != off_check_end; ++index)
  93     {
  94       if (-1 == allocator->offset (index->iov_base))
  95         return this->send (iov, iovcnt, bytes_transferred,
  96                            this->io_timeout(dc));
  97     }
  98
  99   ssize_t retval = -1;
 100
 101   ACE_HANDLE const in_fd = allocator->handle ();
 102
 103   if (in_fd == ACE_INVALID_HANDLE)
 104     return retval;
 105
 106   ACE_HANDLE const out_fd =
 107     this->connection_handler_->peer ().get_handle ();
 108
 109   iovec * const begin = iov;
 110   iovec * const end   = iov + iovcnt;
 111   for (iovec * i = begin; i != end; ++i)
 112     {
 113       off_t offset = allocator->offset (i->iov_base);
 114
 115       if (this->io_timeout(dc))
 116         {
 117           int val = 0;
 118           if (ACE::enter_send_timedwait (out_fd,
 119                                          this->io_timeout(dc), val) == -1)
 120             return retval;
 121           else
 122             {
 123               retval =
 124                 ACE_OS::sendfile (out_fd, in_fd, &offset, i->iov_len);
 125               ACE::restore_non_blocking_mode (out_fd, val);
 126             }
 127         }
 128       else
 129         {
 130           retval = ACE_OS::sendfile (out_fd, in_fd, &offset, i->iov_len);
 131         }
 132
 133       if (retval <= 0)  // Report errors below.
 134         break;
 135
 136       bytes_transferred += static_cast<size_t> (retval);
 137     }
 138
 139   if (retval <= 0 && TAO_debug_level > 4)
 140     {
 141       TAOLIB_DEBUG ((LM_DEBUG,
 142                   ACE_TEXT ("TAO (%P|%t) - IIOP_Transport[%d]::sendfile, ")
 143                   ACE_TEXT ("sendfile failure - %m (errno: %d)\n"),
 144                   this->id (),
 145                   ACE_ERRNO_GET));
 146     }
 147
 148   return retval;
 149 }
 150 #endif  /* TAO_HAS_SENDFILE==1 */
 151
 152 ssize_t
 153 TAO_IIOP_Transport::recv (char *buf,
 154                           size_t len,
 155                           const ACE_Time_Value *max_wait_time)
 156 {
 157   this->connection_closed_on_read_ = false;
 158
 159   ssize_t const n = this->connection_handler_->peer ().recv (buf,
 160                                                              len,
 161                                                              max_wait_time);
 162
 163   // Do not print the error message if it is a timeout, which could
 164   // occur in thread-per-connection.
 165   if (n == -1 && TAO_debug_level > 4 && errno != ETIME)
 166     {
 167       TAOLIB_DEBUG ((LM_DEBUG,
 168                   ACE_TEXT ("TAO (%P|%t) - IIOP_Transport[%d]::recv, ")
 169                   ACE_TEXT ("read failure - %m errno %d\n"),
 170                   this->id (),
 171                   ACE_ERRNO_GET));
 172     }
 173
 174   // Error handling
 175   if (n == -1)
 176     {
 177       if (errno == EWOULDBLOCK)
 178         return 0;
 179
 180       return -1;
 181     }
 182
 183   // Most of the errors handling is common for
 184   // Now the message has been read
 185
 186   // @@ What are the other error handling here??
 187   else if (n == 0)
 188     {
 189       this->connection_closed_on_read_ = true;
 190       return -1;
 191     }
 192
 193   return n;
 194 }
 195
 196 int
 197 TAO_IIOP_Transport::send_request (TAO_Stub *stub,
 198                                   TAO_ORB_Core *orb_core,
 199                                   TAO_OutputCDR &stream,
 200                                   TAO_Message_Semantics message_semantics,
 201                                   ACE_Time_Value *max_wait_time)
 202 {
 203   if (this->ws_->sending_request (orb_core, message_semantics) == -1)
 204     {
 205       return -1;
 206     }
 207
 208   if (this->send_message (stream,
 209                           stub,
 210                           nullptr,
 211                           message_semantics,
 212                           max_wait_time) == -1)
 213     {
 214       return -1;
 215     }
 216
 217   this->first_request_sent();
 218
 219   return 0;
 220 }
 221
 222 int
 223 TAO_IIOP_Transport::send_message (TAO_OutputCDR &stream,
 224                                   TAO_Stub *stub,
 225                                   TAO_ServerRequest *request,
 226                                   TAO_Message_Semantics message_semantics,
 227                                   ACE_Time_Value *max_wait_time)
 228 {
 229   // Format the message in the stream first
 230   if (this->messaging_object ()->format_message (stream, stub, request) != 0)
 231     {
 232       return -1;
 233     }
 234
 235   // This guarantees to send all data (bytes) or return an error.
 236   ssize_t const n = this->send_message_shared (stub,
 237                                                message_semantics,
 238                                                stream.begin (),
 239                                                max_wait_time);
 240
 241   if (n == -1)
 242     {
 243       // Dont try to be smart and request for %p in the debug
 244       // statement.  If the event handler is destroyed the transport
 245       // would return -1 with errno set to ENOENT. %p then would dump
 246       // a core. %m would then be softer on this.
 247       if (TAO_debug_level)
 248         {
 249           TAOLIB_DEBUG ((LM_DEBUG,
 250                       ACE_TEXT ("TAO (%P|%t) - IIOP_Transport[%d]::send_message, ")
 251                       ACE_TEXT ("write failure - %m\n"),
 252                       this->id ()));
 253         }
 254       return -1;
 255     }
 256
 257   return 1;
 258 }
 259
 260 int
 261 TAO_IIOP_Transport::tear_listen_point_list (TAO_InputCDR &cdr)
 262 {
 263   CORBA::Boolean byte_order;
 264   if (!(cdr >> ACE_InputCDR::to_boolean (byte_order)))
 265     return -1;
 266
 267   cdr.reset_byte_order (static_cast<int> (byte_order));
 268
 269   IIOP::ListenPointList listen_list;
 270   if (!(cdr >> listen_list))
 271     return -1;
 272
 273   // As we have received a bidirectional information, set the flag to
 274   // 0 (i.e., non-originating side)
 275   this->bidirectional_flag (0);
 276
 277   return this->connection_handler_->process_listen_point_list (listen_list);
 278 }
 279
 280 void
 281 TAO_IIOP_Transport::set_bidir_context_info (TAO_Operation_Details &opdetails)
 282 {
 283   // Get a handle to the acceptor registry
 284   TAO_Acceptor_Registry &ar =
 285     this->orb_core ()->lane_resources ().acceptor_registry ();
 286
 287   IIOP::ListenPointList listen_point_list;
 288
 289   TAO_AcceptorSetIterator const end = ar.end ();
 290
 291   for (TAO_AcceptorSetIterator acceptor = ar.begin ();
 292        acceptor != end;
 293        ++acceptor)
 294     {
 295       // Check whether it is an IIOP acceptor
 296       if ((*acceptor)->tag () == this->tag ())
 297         {
 298           if (this->get_listen_point (listen_point_list, *acceptor) == -1)
 299             {
 300               if (TAO_debug_level > 0)
 301                 TAOLIB_ERROR ((LM_ERROR,
 302                             "TAO (%P|%t) - IIOP_Transport::set_bidir_context_info, "
 303                             "error getting listen_point\n"));
 304
 305               return;
 306             }
 307         }
 308     }
 309
 310   if (listen_point_list.length () == 0)
 311     {
 312       if (TAO_debug_level > 0)
 313         TAOLIB_ERROR ((LM_ERROR,
 314                     "TAO (%P|%t) - IIOP_Transport::set_bidir_context_info, "
 315                     "listen_point list is empty, client should send a list "
 316                     "with at least one point\n"));
 317
 318       return;
 319     }
 320
 321   // We have the ListenPointList at this point. Create a output CDR
 322   // stream at this point
 323   TAO_OutputCDR cdr;
 324
 325   // Marshal the information into the stream
 326   if (!(cdr << ACE_OutputCDR::from_boolean (TAO_ENCAP_BYTE_ORDER))
 327       || (!(cdr << listen_point_list)))
 328     return;
 329
 330   // Add this info in to the svc_list
 331   opdetails.request_service_context ().set_context (IOP::BI_DIR_IIOP, cdr);
 332
 333   return;
 334 }
 335
 336 int
 337 TAO_IIOP_Transport::get_listen_point (
 338     IIOP::ListenPointList &listen_point_list,
 339     TAO_Acceptor *acceptor)
 340 {
 341   TAO_IIOP_Acceptor *iiop_acceptor =
 342     dynamic_cast<TAO_IIOP_Acceptor *> (acceptor);
 343
 344   if (iiop_acceptor == nullptr)
 345     return -1;
 346
 347   // Get the array of endpoints serviced by TAO_IIOP_Acceptor
 348   const ACE_INET_Addr *endpoint_addr =
 349     iiop_acceptor->endpoints ();
 350   size_t count = iiop_acceptor->endpoint_count ();
 351
 352 #if defined (TAO_USE_BROKEN_BIDIR)
 353
 354   // Note: Looks like there is no point in sending the list of
 355   // endpoints on interfaces on which this connection has not
 356   // been established. If this is wrong, please correct me.
 357
 358   ACE_INET_Addr local_addr;
 359   if (this->connection_handler_->peer ().get_local_addr (local_addr) == -1)
 360     {
 361       TAOLIB_ERROR_RETURN ((LM_ERROR,
 362                          ACE_TEXT ("TAO (%P|%t) - IIOP_Transport::get_listen_point, ")
 363                          ACE_TEXT ("could not resolve local host address\n")),
 364                         -1);
 365     }
 366 #endif /* TAO_USE_BROKEN_BIDIR */
 367
 368   for (size_t index = 0; index < count; index++)
 369     {
 370 #if defined (TAO_USE_BROKEN_BIDIR)
 371       // Make sure port numbers are equal so the following comparison
 372       // only concerns the IP(v4/v6) address.
 373       local_addr.set_port_number (endpoint_addr[index].get_port_number ());
 374
 375       if (local_addr != endpoint_addr[index])
 376         continue;
 377 #endif /* TAO_USE_BROKEN_BIDIR */
 378
 379       // Get the local address of the connection
 380       CORBA::String_var interface_name;
 381
 382       // Get the hostname for the local address
 383       if (iiop_acceptor->hostname (this->orb_core_,
 384                                    endpoint_addr[index],
 385                                    interface_name.out ()) == -1)
 386         {
 387           TAOLIB_ERROR_RETURN ((LM_ERROR,
 388                              ACE_TEXT ("TAO (%P|%t) - IIOP_Transport::get_listen_point, ")
 389                              ACE_TEXT ("could not resolve local host name\n")),
 390                             -1);
 391         }
 392
 393 #if defined (ACE_HAS_IPV6)
 394       // If this is an IPv6 decimal linklocal address containing a scopeid than
 395       // remove the scopeid from the information being sent.
 396       const char *cp_scope = 0;
 397       if (endpoint_addr[index].get_type () == PF_INET6 &&
 398           (cp_scope = ACE_OS::strchr (interface_name.in (), '%')) != 0)
 399         {
 400           CORBA::ULong len = cp_scope - interface_name.in ();
 401           interface_name[len] = '\0';
 402         }
 403 #endif /* ACE_HAS_IPV6 */
 404
 405       // Get the count of the number of elements
 406       CORBA::ULong const len = listen_point_list.length ();
 407
 408       // Increase the length by 1
 409       listen_point_list.length (len + 1);
 410
 411       // We have the connection and the acceptor endpoint on the
 412       // same interface
 413       IIOP::ListenPoint & point = listen_point_list[len];
 414       point.host = CORBA::string_dup (interface_name.in ());
 415       point.port = endpoint_addr[index].get_port_number ();
 416
 417       if (TAO_debug_level >= 5)
 418         {
 419           TAOLIB_DEBUG ((LM_DEBUG,
 420                       ACE_TEXT("TAO (%P|%t) - Listen_Point_List[%d] = <%C:%d>\n"),
 421                       len,
 422                       point.host.in (),
 423                       point.port));
 424         }
 425     }
 426   return 1;
 427 }
 428
 429 TAO_END_VERSIONED_NAMESPACE_DECL
 430
 431 #endif /* TAO_HAS_IIOP && TAO_HAS_IIOP != 0 */