ACE/ace/Pipe.cpp

   1 #include "ace/Pipe.h"
   2 #include "ace/SOCK_Acceptor.h"
   3 #include "ace/SOCK_Connector.h"
   4 #include "ace/Log_Category.h"
   5 #include "ace/OS_NS_stdio.h"
   6 #include "ace/OS_NS_sys_socket.h"
   7 #include "ace/OS_Memory.h"
   8 #include "ace/Truncate.h"
   9 #include "ace/Malloc_Base.h"
  10
  11 #if defined (ACE_HAS_STREAM_PIPES) || defined (__QNX__)
  12 #  include "ace/OS_NS_unistd.h"
  13 #endif  // ACE_HAS_STREAM_PIPES || __QNX__
  14
  15 #if defined (ACE_LACKS_LISTEN) && defined (ACE_LACKS_SOCKETPAIR) \
  16   && !defined (ACE_HAS_STREAM_PIPES)
  17 #  include "ace/OS_NS_time.h"
  18 #  include "ace/os_include/sys/os_un.h"
  19 #endif
  20
  21 #include "ace/os_include/netinet/os_tcp.h"
  22
  23 #if !defined (__ACE_INLINE__)
  24 #include "ace/Pipe.inl"
  25 #endif /* __ACE_INLINE__ */
  26
  27
  28
  29 ACE_BEGIN_VERSIONED_NAMESPACE_DECL
  30
  31 void
  32 ACE_Pipe::dump (void) const
  33 {
  34 #if defined (ACE_HAS_DUMP)
  35   ACE_TRACE ("ACE_Pipe::dump");
  36   ACELIB_DEBUG ((LM_DEBUG, ACE_BEGIN_DUMP, this));
  37   ACELIB_DEBUG ((LM_DEBUG, ACE_TEXT ("handles_[0] = %d"), this->handles_[0]));
  38   ACELIB_DEBUG ((LM_DEBUG, ACE_TEXT ("\nhandles_[1] = %d\n"), this->handles_[1]));
  39   ACELIB_DEBUG ((LM_DEBUG, ACE_END_DUMP));
  40 #endif /* ACE_HAS_DUMP */
  41 }
  42
  43 int
  44 ACE_Pipe::open (int buffer_size)
  45 {
  46   ACE_TRACE ("ACE_Pipe::open");
  47
  48 #if defined (ACE_LACKS_LISTEN) && defined (ACE_LACKS_SOCKETPAIR) \
  49   && !defined (ACE_HAS_STREAM_PIPES)
  50   ACE_UNUSED_ARG (buffer_size);
  51
  52   if ((this->handles_[0] = ACE_OS::socket (AF_LOCAL, SOCK_DGRAM, 0)) == -1)
  53     {
  54       return -1;
  55     }
  56
  57   sockaddr_un addr = {
  58 #if defined(ACE_VXWORKS) || defined(__APPLE__)
  59     sizeof (sockaddr_un),
  60 #endif
  61     AF_LOCAL, {}};
  62   unsigned seed = static_cast<unsigned> (ACE_OS::time ());
  63   ACE_OS::snprintf (addr.sun_path, sizeof addr.sun_path, "/tmp/ACE-Pipe-%d-%p",
  64                     ACE_OS::rand_r (&seed), this);
  65
  66   if (ACE_OS::bind (this->handles_[0], (sockaddr*) &addr, sizeof addr) == -1)
  67     {
  68       this->close ();
  69       return -1;
  70     }
  71
  72   if ((this->handles_[1] = ACE_OS::socket (AF_LOCAL, SOCK_DGRAM, 0)) == -1 ||
  73       ACE_OS::connect (this->handles_[1], (sockaddr*) &addr, sizeof addr) == -1)
  74     {
  75       ACE_OS::unlink (addr.sun_path);
  76       this->close ();
  77       return -1;
  78     }
  79
  80   ACE_OS::unlink (addr.sun_path);
  81
  82 #elif defined (ACE_LACKS_SOCKETPAIR)
  83   ACE_INET_Addr my_addr;
  84   ACE_SOCK_Acceptor acceptor;
  85   ACE_SOCK_Connector connector;
  86   ACE_SOCK_Stream reader;
  87   ACE_SOCK_Stream writer;
  88   int result = 0;
  89 # if defined (ACE_WIN32)
  90   ACE_INET_Addr local_any  (static_cast<u_short> (0), ACE_LOCALHOST);
  91 # else
  92   ACE_Addr local_any = ACE_Addr::sap_any;
  93 # endif /* ACE_WIN32 */
  94
  95   // Bind listener to any port and then find out what the port was.
  96   if (acceptor.open (local_any) == -1
  97       || acceptor.get_local_addr (my_addr) == -1)
  98     result = -1;
  99   else
 100     {
 101       ACE_INET_Addr sv_addr (my_addr.get_port_number (),
 102                              ACE_LOCALHOST);
 103
 104       // Establish a connection within the same process.
 105       if (connector.connect (writer, sv_addr) == -1)
 106         result = -1;
 107       else if (acceptor.accept (reader) == -1)
 108         {
 109           writer.close ();
 110           result = -1;
 111         }
 112     }
 113
 114   // Close down the acceptor endpoint since we don't need it anymore.
 115   acceptor.close ();
 116   if (result == -1)
 117     return -1;
 118
 119   this->handles_[0] = reader.get_handle ();
 120   this->handles_[1] = writer.get_handle ();
 121
 122 # if !defined (ACE_LACKS_TCP_NODELAY)
 123   int one = 1;
 124
 125   // Make sure that the TCP stack doesn't try to buffer small writes.
 126   // Since this communication is purely local to the host it doesn't
 127   // affect network performance.
 128
 129   if (writer.set_option (ACE_IPPROTO_TCP,
 130                          TCP_NODELAY,
 131                          &one,
 132                          sizeof one) == -1)
 133     {
 134       this->close ();
 135       return -1;
 136     }
 137 # endif /* ! ACE_LACKS_TCP_NODELAY */
 138
 139 # if defined (ACE_LACKS_SO_RCVBUF) && defined (ACE_LACKS_SO_SNDBUF)
 140     ACE_UNUSED_ARG (buffer_size);
 141 # endif
 142 # if !defined (ACE_LACKS_SO_RCVBUF)
 143   if (reader.set_option (SOL_SOCKET,
 144                          SO_RCVBUF,
 145                          reinterpret_cast <void *> (&buffer_size),
 146                          sizeof (buffer_size)) == -1
 147       && errno != ENOTSUP)
 148     {
 149       this->close ();
 150       return -1;
 151     }
 152 # endif /* !ACE_LACKS_SO_RCVBUF */
 153 # if !defined (ACE_LACKS_SO_SNDBUF)
 154   if (writer.set_option (SOL_SOCKET,
 155                          SO_SNDBUF,
 156                          reinterpret_cast <void *> (&buffer_size),
 157                          sizeof (buffer_size)) == -1
 158            && errno != ENOTSUP)
 159     {
 160       this->close ();
 161       return -1;
 162     }
 163 # endif /* !ACE_LACKS_SO_SNDBUF */
 164
 165 #elif defined (ACE_HAS_STREAM_PIPES) || defined (__QNX__)
 166   ACE_UNUSED_ARG (buffer_size);
 167   if (ACE_OS::pipe (this->handles_) == -1)
 168     ACELIB_ERROR_RETURN ((LM_ERROR,
 169                        ACE_TEXT ("%p\n"),
 170                        ACE_TEXT ("pipe")),
 171                       -1);
 172
 173 # if !defined(__QNX__)
 174   int arg = RMSGN;
 175
 176   // Enable "msg no discard" mode, which ensures that record
 177   // boundaries are maintained when messages are sent and received.
 178   if (ACE_OS::ioctl (this->handles_[0],
 179                      I_SRDOPT,
 180                      (void *) arg) == -1
 181       || ACE_OS::ioctl (this->handles_[1],
 182                         I_SRDOPT,
 183                         (void *) arg) == -1)
 184     {
 185       this->close ();
 186       ACELIB_ERROR_RETURN ((LM_ERROR,
 187                          ACE_TEXT ("%p\n"),
 188                          ACE_TEXT ("ioctl")), -1);
 189     }
 190 # endif /* __QNX__ */
 191
 192 #else  /* ! ACE_LACKS_SOCKETPAIR && ! ACE_HAS_STREAM_PIPES */
 193   if (ACE_OS::socketpair (AF_UNIX,
 194                           SOCK_STREAM,
 195                           0,
 196                           this->handles_) == -1)
 197     ACELIB_ERROR_RETURN ((LM_ERROR,
 198                        ACE_TEXT ("%p\n"),
 199                        ACE_TEXT ("socketpair")),
 200                       -1);
 201 # if defined (ACE_LACKS_SO_SNDBUF) && defined (ACE_LACKS_SO_RCVBUF)
 202   ACE_UNUSED_ARG (buffer_size);
 203 # endif
 204 # if !defined (ACE_LACKS_SO_RCVBUF)
 205   if (ACE_OS::setsockopt (this->handles_[0],
 206                           SOL_SOCKET,
 207                           SO_RCVBUF,
 208                           reinterpret_cast <const char *> (&buffer_size),
 209                           sizeof (buffer_size)) == -1
 210       && errno != ENOTSUP)
 211     {
 212       this->close ();
 213       return -1;
 214     }
 215 # endif
 216 # if !defined (ACE_LACKS_SO_SNDBUF)
 217   if (ACE_OS::setsockopt (this->handles_[1],
 218                           SOL_SOCKET,
 219                           SO_SNDBUF,
 220                           reinterpret_cast <const char *> (&buffer_size),
 221                           sizeof (buffer_size)) == -1
 222       && errno != ENOTSUP)
 223     {
 224       this->close ();
 225       return -1;
 226     }
 227 # endif /* ! ACE_LACKS_SO_SNDBUF */
 228 # if defined (ACE_OPENVMS) && !defined (ACE_LACKS_TCP_NODELAY)
 229   int one = 1;
 230   // OpenVMS implements socketpair(AF_UNIX...) by returning AF_INET sockets.
 231   // Since these are plagued by Nagle as any other INET socket we need to set
 232   // TCP_NODELAY on the write handle.
 233   if (ACE_OS::setsockopt (this->handles_[1],
 234                           ACE_IPPROTO_TCP,
 235                           TCP_NODELAY,
 236                           reinterpret_cast <const char *> (&one),
 237                           sizeof (one)) == -1)
 238     {
 239       this->close ();
 240       return -1;
 241     }
 242 # endif /* ACE_OPENVMS && !ACE_LACKS_TCP_NODELAY */
 243 #endif  /* ! ACE_LACKS_SOCKETPAIR && ! ACE_HAS_STREAM_PIPES */
 244   // Point both the read and write HANDLES to the appropriate socket
 245   // HANDLEs.
 246
 247   return 0;
 248 }
 249
 250 int
 251 ACE_Pipe::open (ACE_HANDLE handles[2])
 252 {
 253   ACE_TRACE ("ACE_Pipe::open");
 254
 255   if (this->open () == -1)
 256     return -1;
 257   else
 258     {
 259       handles[0] = this->handles_[0];
 260       handles[1] = this->handles_[1];
 261       return 0;
 262     }
 263 }
 264
 265 // Do nothing...
 266
 267 ACE_Pipe::ACE_Pipe (void)
 268 {
 269   ACE_TRACE ("ACE_Pipe::ACE_Pipe");
 270
 271   this->handles_[0] = ACE_INVALID_HANDLE;
 272   this->handles_[1] = ACE_INVALID_HANDLE;
 273 }
 274
 275 ACE_Pipe::ACE_Pipe (ACE_HANDLE handles[2])
 276 {
 277   ACE_TRACE ("ACE_Pipe::ACE_Pipe");
 278
 279   if (this->open (handles) == -1)
 280     ACELIB_ERROR ((LM_ERROR,
 281                 ACE_TEXT ("ACE_Pipe::ACE_Pipe")));
 282 }
 283
 284 ACE_Pipe::ACE_Pipe (ACE_HANDLE read,
 285                     ACE_HANDLE write)
 286 {
 287   ACE_TRACE ("ACE_Pipe::ACE_Pipe");
 288   this->handles_[0] = read;
 289   this->handles_[1] = write;
 290 }
 291
 292 int
 293 ACE_Pipe::close (void)
 294 {
 295   ACE_TRACE ("ACE_Pipe::close");
 296
 297   int result = this->close_read ();
 298   result |= this->close_write ();
 299   return result;
 300 }
 301
 302 int
 303 ACE_Pipe::close_read (void)
 304 {
 305   return this->close_handle (0);
 306 }
 307
 308 int ACE_Pipe::close_write (void)
 309 {
 310   return this->close_handle (1);
 311 }
 312
 313 // Send N char *ptrs and int lengths.  Note that the char *'s precede
 314 // the ints (basically, an varargs version of writev).  The count N is
 315 // the *total* number of trailing arguments, *not* a couple of the
 316 // number of tuple pairs!
 317 #if !defined (ACE_LACKS_VA_FUNCTIONS)
 318 ssize_t
 319 ACE_Pipe::send (size_t n, ...) const
 320 {
 321   ACE_TRACE ("ACE_Pipe::send");
 322   va_list argp;
 323   int total_tuples = ACE_Utils::truncate_cast<int> (n / 2);
 324   iovec *iovp;
 325 #if defined (ACE_HAS_ALLOCA)
 326   iovp = (iovec *) alloca (total_tuples * sizeof (iovec));
 327 #else
 328 # ifdef ACE_HAS_ALLOC_HOOKS
 329   ACE_ALLOCATOR_RETURN (iovp, (iovec *)
 330                         ACE_Allocator::instance ()->malloc (total_tuples *
 331                                                             sizeof (iovec)),
 332                         -1);
 333 # else
 334   ACE_NEW_RETURN (iovp,
 335                   iovec[total_tuples],
 336                   -1);
 337 # endif /* ACE_HAS_ALLOC_HOOKS */
 338 #endif /* !defined (ACE_HAS_ALLOCA) */
 339
 340   va_start (argp, n);
 341
 342   for (int i = 0; i < total_tuples; ++i)
 343     {
 344       iovp[i].iov_base = va_arg (argp, char *);
 345       iovp[i].iov_len  = va_arg (argp, int);
 346     }
 347
 348 #if defined (ACE_WIN32)
 349   ssize_t result = ACE::sendv (this->write_handle (),
 350                                iovp,
 351                                total_tuples);
 352 #else
 353   ssize_t result = ACE_OS::writev (this->write_handle (),
 354                                    iovp,
 355                                    total_tuples);
 356 #endif /* ACE_WIN32 */
 357
 358 #if !defined (ACE_HAS_ALLOCA)
 359 # ifdef ACE_HAS_ALLOC_HOOKS
 360   ACE_Allocator::instance ()->free (iovp);
 361 # else
 362   delete [] iovp;
 363 # endif /* ACE_HAS_ALLOC_HOOKS */
 364 #endif /* !defined (ACE_HAS_ALLOCA) */
 365   va_end (argp);
 366   return result;
 367 }
 368
 369 // This is basically an interface to ACE_OS::readv, that doesn't use
 370 // the struct iovec explicitly.  The ... can be passed as an arbitrary
 371 // number of (char *ptr, int len) tuples.  However, the count N is the
 372 // *total* number of trailing arguments, *not* a couple of the number
 373 // of tuple pairs!
 374
 375 ssize_t
 376 ACE_Pipe::recv (size_t n, ...) const
 377 {
 378   ACE_TRACE ("ACE_Pipe::recv");
 379   va_list argp;
 380   int total_tuples = ACE_Utils::truncate_cast<int> (n / 2);
 381   iovec *iovp;
 382 #if defined (ACE_HAS_ALLOCA)
 383   iovp = (iovec *) alloca (total_tuples * sizeof (iovec));
 384 #else
 385 # ifdef ACE_HAS_ALLOC_HOOKS
 386   ACE_ALLOCATOR_RETURN (iovp, (iovec *)
 387                         ACE_Allocator::instance ()->malloc (total_tuples *
 388                                                             sizeof (iovec)),
 389                         -1);
 390 # else
 391   ACE_NEW_RETURN (iovp,
 392                   iovec[total_tuples],
 393                   -1);
 394 # endif /* ACE_HAS_ALLOC_HOOKS */
 395 #endif /* !defined (ACE_HAS_ALLOCA) */
 396
 397   va_start (argp, n);
 398
 399   for (int i = 0; i < total_tuples; ++i)
 400     {
 401       iovp[i].iov_base = va_arg (argp, char *);
 402       iovp[i].iov_len  = va_arg (argp, int);
 403     }
 404
 405 #if defined (ACE_WIN32)
 406   ssize_t const result = ACE::recvv (this->read_handle (),
 407                                      iovp,
 408                                      total_tuples);
 409 #else
 410   ssize_t const result = ACE_OS::readv (this->read_handle (),
 411                                         iovp,
 412                                         total_tuples);
 413 #endif /* ACE_WIN32 */
 414
 415 #if !defined (ACE_HAS_ALLOCA)
 416 # ifdef ACE_HAS_ALLOC_HOOKS
 417   ACE_Allocator::instance ()->free (iovp);
 418 # else
 419   delete [] iovp;
 420 # endif /* ACE_HAS_ALLOC_HOOKS */
 421 #endif /* !defined (ACE_HAS_ALLOCA) */
 422   va_end (argp);
 423   return result;
 424 }
 425 #endif /* !ACE_LACKS_VA_FUNCTIONS */
 426
 427 ACE_END_VERSIONED_NAMESPACE_DECL