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