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;
 102       if (sv_addr.set (my_addr.get_port_number (), ACE_LOCALHOST) == -1)
 103         {
 104           result = -1;
 105         }
 106       // Establish a connection within the same process.
 107       else if (connector.connect (writer, sv_addr) == -1)
 108         result = -1;
 109       else if (acceptor.accept (reader) == -1)
 110         {
 111           writer.close ();
 112           result = -1;
 113         }
 114     }
 115
 116   // Close down the acceptor endpoint since we don't need it anymore.
 117   acceptor.close ();
 118   if (result == -1)
 119     return -1;
 120
 121   this->handles_[0] = reader.get_handle ();
 122   this->handles_[1] = writer.get_handle ();
 123
 124 # if !defined (ACE_LACKS_TCP_NODELAY)
 125   int one = 1;
 126
 127   // Make sure that the TCP stack doesn't try to buffer small writes.
 128   // Since this communication is purely local to the host it doesn't
 129   // affect network performance.
 130
 131   if (writer.set_option (ACE_IPPROTO_TCP,
 132                          TCP_NODELAY,
 133                          &one,
 134                          sizeof one) == -1)
 135     {
 136       this->close ();
 137       return -1;
 138     }
 139 # endif /* ! ACE_LACKS_TCP_NODELAY */
 140
 141 # if defined (ACE_LACKS_SO_RCVBUF) && defined (ACE_LACKS_SO_SNDBUF)
 142     ACE_UNUSED_ARG (buffer_size);
 143 # endif
 144 # if !defined (ACE_LACKS_SO_RCVBUF)
 145   if (reader.set_option (SOL_SOCKET,
 146                          SO_RCVBUF,
 147                          reinterpret_cast <void *> (&buffer_size),
 148                          sizeof (buffer_size)) == -1
 149       && errno != ENOTSUP)
 150     {
 151       this->close ();
 152       return -1;
 153     }
 154 # endif /* !ACE_LACKS_SO_RCVBUF */
 155 # if !defined (ACE_LACKS_SO_SNDBUF)
 156   if (writer.set_option (SOL_SOCKET,
 157                          SO_SNDBUF,
 158                          reinterpret_cast <void *> (&buffer_size),
 159                          sizeof (buffer_size)) == -1
 160            && errno != ENOTSUP)
 161     {
 162       this->close ();
 163       return -1;
 164     }
 165 # endif /* !ACE_LACKS_SO_SNDBUF */
 166
 167 #elif defined (ACE_HAS_STREAM_PIPES) || defined (__QNX__)
 168   ACE_UNUSED_ARG (buffer_size);
 169   if (ACE_OS::pipe (this->handles_) == -1)
 170     ACELIB_ERROR_RETURN ((LM_ERROR,
 171                        ACE_TEXT ("%p\n"),
 172                        ACE_TEXT ("pipe")),
 173                       -1);
 174
 175 # if !defined(__QNX__)
 176   int arg = RMSGN;
 177
 178   // Enable "msg no discard" mode, which ensures that record
 179   // boundaries are maintained when messages are sent and received.
 180   if (ACE_OS::ioctl (this->handles_[0],
 181                      I_SRDOPT,
 182                      (void *) arg) == -1
 183       || ACE_OS::ioctl (this->handles_[1],
 184                         I_SRDOPT,
 185                         (void *) arg) == -1)
 186     {
 187       this->close ();
 188       ACELIB_ERROR_RETURN ((LM_ERROR,
 189                          ACE_TEXT ("%p\n"),
 190                          ACE_TEXT ("ioctl")), -1);
 191     }
 192 # endif /* __QNX__ */
 193
 194 #else  /* ! ACE_LACKS_SOCKETPAIR && ! ACE_HAS_STREAM_PIPES */
 195   if (ACE_OS::socketpair (AF_UNIX,
 196                           SOCK_STREAM,
 197                           0,
 198                           this->handles_) == -1)
 199     ACELIB_ERROR_RETURN ((LM_ERROR,
 200                        ACE_TEXT ("%p\n"),
 201                        ACE_TEXT ("socketpair")),
 202                       -1);
 203 # if defined (ACE_LACKS_SO_SNDBUF) && defined (ACE_LACKS_SO_RCVBUF)
 204   ACE_UNUSED_ARG (buffer_size);
 205 # endif
 206 # if !defined (ACE_LACKS_SO_RCVBUF)
 207   if (ACE_OS::setsockopt (this->handles_[0],
 208                           SOL_SOCKET,
 209                           SO_RCVBUF,
 210                           reinterpret_cast <const char *> (&buffer_size),
 211                           sizeof (buffer_size)) == -1
 212       && errno != ENOTSUP)
 213     {
 214       this->close ();
 215       return -1;
 216     }
 217 # endif
 218 # if !defined (ACE_LACKS_SO_SNDBUF)
 219   if (ACE_OS::setsockopt (this->handles_[1],
 220                           SOL_SOCKET,
 221                           SO_SNDBUF,
 222                           reinterpret_cast <const char *> (&buffer_size),
 223                           sizeof (buffer_size)) == -1
 224       && errno != ENOTSUP)
 225     {
 226       this->close ();
 227       return -1;
 228     }
 229 # endif /* ! ACE_LACKS_SO_SNDBUF */
 230 # if defined (ACE_OPENVMS) && !defined (ACE_LACKS_TCP_NODELAY)
 231   int one = 1;
 232   // OpenVMS implements socketpair(AF_UNIX...) by returning AF_INET sockets.
 233   // Since these are plagued by Nagle as any other INET socket we need to set
 234   // TCP_NODELAY on the write handle.
 235   if (ACE_OS::setsockopt (this->handles_[1],
 236                           ACE_IPPROTO_TCP,
 237                           TCP_NODELAY,
 238                           reinterpret_cast <const char *> (&one),
 239                           sizeof (one)) == -1)
 240     {
 241       this->close ();
 242       return -1;
 243     }
 244 # endif /* ACE_OPENVMS && !ACE_LACKS_TCP_NODELAY */
 245 #endif  /* ! ACE_LACKS_SOCKETPAIR && ! ACE_HAS_STREAM_PIPES */
 246   // Point both the read and write HANDLES to the appropriate socket
 247   // HANDLEs.
 248
 249   return 0;
 250 }
 251
 252 int
 253 ACE_Pipe::open (ACE_HANDLE handles[2])
 254 {
 255   ACE_TRACE ("ACE_Pipe::open");
 256
 257   if (this->open () == -1)
 258     return -1;
 259   else
 260     {
 261       handles[0] = this->handles_[0];
 262       handles[1] = this->handles_[1];
 263       return 0;
 264     }
 265 }
 266
 267 // Do nothing...
 268
 269 ACE_Pipe::ACE_Pipe (void)
 270 {
 271   ACE_TRACE ("ACE_Pipe::ACE_Pipe");
 272
 273   this->handles_[0] = ACE_INVALID_HANDLE;
 274   this->handles_[1] = ACE_INVALID_HANDLE;
 275 }
 276
 277 ACE_Pipe::ACE_Pipe (ACE_HANDLE handles[2])
 278 {
 279   ACE_TRACE ("ACE_Pipe::ACE_Pipe");
 280
 281   if (this->open (handles) == -1)
 282     ACELIB_ERROR ((LM_ERROR,
 283                 ACE_TEXT ("ACE_Pipe::ACE_Pipe")));
 284 }
 285
 286 ACE_Pipe::ACE_Pipe (ACE_HANDLE read,
 287                     ACE_HANDLE write)
 288 {
 289   ACE_TRACE ("ACE_Pipe::ACE_Pipe");
 290   this->handles_[0] = read;
 291   this->handles_[1] = write;
 292 }
 293
 294 int
 295 ACE_Pipe::close (void)
 296 {
 297   ACE_TRACE ("ACE_Pipe::close");
 298
 299   int result = this->close_read ();
 300   result |= this->close_write ();
 301   return result;
 302 }
 303
 304 int
 305 ACE_Pipe::close_read (void)
 306 {
 307   return this->close_handle (0);
 308 }
 309
 310 int ACE_Pipe::close_write (void)
 311 {
 312   return this->close_handle (1);
 313 }
 314
 315 // Send N char *ptrs and int lengths.  Note that the char *'s precede
 316 // the ints (basically, an varargs version of writev).  The count N is
 317 // the *total* number of trailing arguments, *not* a couple of the
 318 // number of tuple pairs!
 319 #if !defined (ACE_LACKS_VA_FUNCTIONS)
 320 ssize_t
 321 ACE_Pipe::send (size_t n, ...) const
 322 {
 323   ACE_TRACE ("ACE_Pipe::send");
 324   va_list argp;
 325   int total_tuples = ACE_Utils::truncate_cast<int> (n / 2);
 326   iovec *iovp;
 327 #if defined (ACE_HAS_ALLOCA)
 328   iovp = (iovec *) alloca (total_tuples * sizeof (iovec));
 329 #else
 330 # ifdef ACE_HAS_ALLOC_HOOKS
 331   ACE_ALLOCATOR_RETURN (iovp, (iovec *)
 332                         ACE_Allocator::instance ()->malloc (total_tuples *
 333                                                             sizeof (iovec)),
 334                         -1);
 335 # else
 336   ACE_NEW_RETURN (iovp,
 337                   iovec[total_tuples],
 338                   -1);
 339 # endif /* ACE_HAS_ALLOC_HOOKS */
 340 #endif /* !defined (ACE_HAS_ALLOCA) */
 341
 342   va_start (argp, n);
 343
 344   for (int i = 0; i < total_tuples; ++i)
 345     {
 346       iovp[i].iov_base = va_arg (argp, char *);
 347       iovp[i].iov_len  = va_arg (argp, int);
 348     }
 349
 350 #if defined (ACE_WIN32)
 351   ssize_t result = ACE::sendv (this->write_handle (),
 352                                iovp,
 353                                total_tuples);
 354 #else
 355   ssize_t result = ACE_OS::writev (this->write_handle (),
 356                                    iovp,
 357                                    total_tuples);
 358 #endif /* ACE_WIN32 */
 359
 360 #if !defined (ACE_HAS_ALLOCA)
 361 # ifdef ACE_HAS_ALLOC_HOOKS
 362   ACE_Allocator::instance ()->free (iovp);
 363 # else
 364   delete [] iovp;
 365 # endif /* ACE_HAS_ALLOC_HOOKS */
 366 #endif /* !defined (ACE_HAS_ALLOCA) */
 367   va_end (argp);
 368   return result;
 369 }
 370
 371 // This is basically an interface to ACE_OS::readv, that doesn't use
 372 // the struct iovec explicitly.  The ... can be passed as an arbitrary
 373 // number of (char *ptr, int len) tuples.  However, the count N is the
 374 // *total* number of trailing arguments, *not* a couple of the number
 375 // of tuple pairs!
 376
 377 ssize_t
 378 ACE_Pipe::recv (size_t n, ...) const
 379 {
 380   ACE_TRACE ("ACE_Pipe::recv");
 381   va_list argp;
 382   int total_tuples = ACE_Utils::truncate_cast<int> (n / 2);
 383   iovec *iovp;
 384 #if defined (ACE_HAS_ALLOCA)
 385   iovp = (iovec *) alloca (total_tuples * sizeof (iovec));
 386 #else
 387 # ifdef ACE_HAS_ALLOC_HOOKS
 388   ACE_ALLOCATOR_RETURN (iovp, (iovec *)
 389                         ACE_Allocator::instance ()->malloc (total_tuples *
 390                                                             sizeof (iovec)),
 391                         -1);
 392 # else
 393   ACE_NEW_RETURN (iovp,
 394                   iovec[total_tuples],
 395                   -1);
 396 # endif /* ACE_HAS_ALLOC_HOOKS */
 397 #endif /* !defined (ACE_HAS_ALLOCA) */
 398
 399   va_start (argp, n);
 400
 401   for (int i = 0; i < total_tuples; ++i)
 402     {
 403       iovp[i].iov_base = va_arg (argp, char *);
 404       iovp[i].iov_len  = va_arg (argp, int);
 405     }
 406
 407 #if defined (ACE_WIN32)
 408   ssize_t const result = ACE::recvv (this->read_handle (),
 409                                      iovp,
 410                                      total_tuples);
 411 #else
 412   ssize_t const result = ACE_OS::readv (this->read_handle (),
 413                                         iovp,
 414                                         total_tuples);
 415 #endif /* ACE_WIN32 */
 416
 417 #if !defined (ACE_HAS_ALLOCA)
 418 # ifdef ACE_HAS_ALLOC_HOOKS
 419   ACE_Allocator::instance ()->free (iovp);
 420 # else
 421   delete [] iovp;
 422 # endif /* ACE_HAS_ALLOC_HOOKS */
 423 #endif /* !defined (ACE_HAS_ALLOCA) */
 424   va_end (argp);
 425   return result;
 426 }
 427 #endif /* !ACE_LACKS_VA_FUNCTIONS */
 428
 429 ACE_END_VERSIONED_NAMESPACE_DECL