ACE/apps/JAWS2/JAWS/Jaws_IO.cpp

   1 #include "ace/Message_Block.h"
   2 #include "ace/SOCK_Stream.h"
   3 #include "ace/Filecache.h"
   4 #include "ace/OS_NS_string.h"
   5 #include "ace/OS_NS_unistd.h"
   6 #include "ace/OS_NS_sys_uio.h"
   7 #include "ace/OS_NS_sys_socket.h"
   8 #include "ace/Min_Max.h"
   9
  10 #include "JAWS/Data_Block.h"
  11 #include "JAWS/Policy.h"
  12 #include "JAWS/Jaws_IO.h"
  13 #include "JAWS/IO_Handler.h"
  14 #include "JAWS/IO_Acceptor.h"
  15 #include "JAWS/Filecache.h"
  16
  17 #include "ace/Asynch_IO.h"  //for ACE_Asynch_Write_Stream
  18
  19 // #include "HTTP_Helpers.h"
  20
  21
  22 JAWS_IO::JAWS_IO ()
  23   : handle_ (ACE_INVALID_HANDLE),
  24     handler_ (0),
  25     inet_addr_ (0),
  26     acceptor_ (0)
  27 {
  28 }
  29
  30 JAWS_IO::~JAWS_IO ()
  31 {
  32 }
  33
  34 #if 0
  35 ACE_HANDLE
  36 JAWS_IO::handle ()
  37 {
  38   return this->handle_;
  39 }
  40
  41 void
  42 JAWS_IO::handle (ACE_HANDLE handle)
  43 {
  44   this->handle_ = handle;
  45 }
  46
  47 void
  48 JAWS_IO::handler (JAWS_IO_Handler *handler)
  49 {
  50   this->handler_ = handler;
  51 }
  52
  53 void
  54 JAWS_IO::acceptor (JAWS_IO_Acceptor *acceptor)
  55 {
  56   this->acceptor_ = acceptor;
  57 }
  58 #endif /* 0 */
  59
  60 JAWS_Synch_IO::JAWS_Synch_IO ()
  61 {
  62   this->acceptor_ = JAWS_IO_Synch_Acceptor_Singleton::instance ();
  63 }
  64
  65 JAWS_Synch_IO::~JAWS_Synch_IO ()
  66 {
  67   if (this->handle_ != ACE_INVALID_HANDLE)
  68     ACE_OS::closesocket (this->handle_);
  69 }
  70
  71 void
  72 JAWS_Synch_IO::accept (JAWS_IO_Handler *ioh,
  73                        ACE_Message_Block *,
  74                        unsigned int)
  75 {
  76   ACE_SOCK_Stream new_stream;
  77   new_stream.set_handle (ACE_INVALID_HANDLE);
  78   if (this->acceptor_->accept (new_stream) == -1)
  79     ioh->accept_error ();
  80   else
  81     ioh->accept_complete (new_stream.get_handle ());
  82 }
  83
  84 void
  85 JAWS_Synch_IO::read (JAWS_IO_Handler *ioh,
  86                      ACE_Message_Block *mb,
  87                      unsigned int size)
  88 {
  89   JAWS_TRACE ("JAWS_Synch_IO::read");
  90
  91   ACE_SOCK_Stream stream;
  92
  93   stream.set_handle (ioh->handle ());
  94   int result = stream.recv (mb->wr_ptr (), size);
  95
  96   if (result <= 0)
  97     ioh->read_error ();
  98   else
  99     {
 100       JAWS_TRACE ("JAWS_Synch_IO::read success");
 101       mb->wr_ptr (result);
 102       ioh->read_complete (mb);
 103     }
 104 }
 105
 106 void
 107 JAWS_Synch_IO::receive_file (JAWS_IO_Handler *ioh,
 108                              const char *filename,
 109                              void *initial_data,
 110                              unsigned int initial_data_length,
 111                              unsigned int entire_length)
 112 {
 113   ACE_Filecache_Handle handle (filename,
 114                                (int) entire_length);
 115
 116   int result = handle.error ();
 117
 118   if (result == ACE_Filecache_Handle::ACE_SUCCESS)
 119     {
 120       ACE_SOCK_Stream stream;
 121       stream.set_handle (ioh->handle ());
 122
 123       int bytes_to_memcpy = ACE_MIN (entire_length, initial_data_length);
 124       ACE_OS::memcpy (handle.address (), initial_data, bytes_to_memcpy);
 125
 126       int bytes_to_read = entire_length - bytes_to_memcpy;
 127
 128       int bytes = stream.recv_n ((char *)
 129                                  handle.address () + initial_data_length,
 130                                  bytes_to_read);
 131       if (bytes == bytes_to_read)
 132         ioh->receive_file_complete ();
 133       else
 134         result = -1;
 135     }
 136
 137   if (result != ACE_Filecache_Handle::ACE_SUCCESS)
 138     ioh->receive_file_error (result);
 139 }
 140
 141 void
 142 JAWS_Synch_IO::transmit_file (JAWS_IO_Handler *ioh,
 143                               ACE_HANDLE handle,
 144                               const char *header,
 145                               unsigned int header_size,
 146                               const char *trailer,
 147                               unsigned int trailer_size)
 148 {
 149   int result = 0;
 150
 151   if (handle != ACE_INVALID_HANDLE)
 152     {
 153       ACE_SOCK_Stream stream;
 154       stream.set_handle (ioh->handle ());
 155
 156       if ((unsigned long) stream.send_n (header, header_size) < header_size)
 157         {
 158           result = -1;
 159         }
 160       else
 161         {
 162           int count;
 163           char buf[BUFSIZ];
 164
 165           do
 166             {
 167               count = ACE_OS::read (handle, buf, sizeof (buf));
 168               if (count <= 0)
 169                 break;
 170
 171               if (stream.send_n (buf, count) < count)
 172                 {
 173                 result = -1;
 174                 }
 175             }
 176           while (result == 0);
 177
 178           if ((unsigned long) stream.send_n (trailer, trailer_size)
 179               < trailer_size)
 180             {
 181             result = -1;
 182             }
 183         }
 184     }
 185
 186   if (result == 0)
 187     ioh->transmit_file_complete ();
 188   else
 189     ioh->transmit_file_error (result);
 190 }
 191
 192 void
 193 JAWS_Synch_IO::transmit_file (JAWS_IO_Handler *ioh,
 194                               const char *filename,
 195                               const char *header,
 196                               unsigned int header_size,
 197                               const char *trailer,
 198                               unsigned int trailer_size)
 199 {
 200   int result = 0;
 201
 202   if (filename == 0)
 203     {
 204       ioh->transmit_file_error (-1);
 205       return;
 206     }
 207
 208   JAWS_Cached_FILE cf (filename);
 209
 210   if (cf.file ()->get_handle () != ACE_INVALID_HANDLE
 211       && cf.mmap () != 0)
 212     {
 213 #if defined (ACE_JAWS_BASELINE) || defined (ACE_WIN32)
 214       ACE_FILE_Info info;
 215       cf.file ()->get_info (info);
 216
 217       if (cf.file ()->get_info (info) == 0 && info.size_ > 0)
 218         {
 219           ACE_SOCK_Stream stream;
 220           stream.set_handle (ioh->handle ());
 221           if (((u_long) stream.send_n (header, header_size) == header_size)
 222               && (stream.send_n (cf.mmap ()->addr (), info.size_)
 223                   == info.size_)
 224               && ((u_long) stream.send_n (trailer, trailer_size)
 225                   == trailer_size))
 226             {
 227               ioh->transmit_file_complete ();
 228               return;
 229             }
 230           else
 231             {
 232               result = -1;
 233             }
 234         }
 235       else
 236         {
 237           result = -1;
 238         }
 239 #else
 240       // Attempting to use writev
 241       // Is this faster?
 242       iovec iov[3];
 243       int iovcnt = 0;
 244       if (header_size > 0)
 245         {
 246           iov[iovcnt].iov_base = const_cast<char*> (header);
 247           iov[iovcnt].iov_len =  header_size;
 248           iovcnt++;
 249         }
 250
 251       ACE_FILE_Info info;
 252
 253       if (cf.file ()->get_info (info) == 0 && info.size_ > 0)
 254         {
 255           iov[iovcnt].iov_base = (char *) cf.mmap ()->addr ();
 256           iov[iovcnt].iov_len = info.size_;
 257           iovcnt++;
 258         }
 259       if (trailer_size > 0)
 260         {
 261           iov[iovcnt].iov_base = const_cast<char*> (trailer);
 262           iov[iovcnt].iov_len = trailer_size;
 263           iovcnt++;
 264         }
 265       if (ACE_OS::writev (ioh->handle (), iov, iovcnt) < 0)
 266         {
 267           result = -1;
 268         }
 269       else
 270         {
 271           ioh->transmit_file_complete ();
 272           return;
 273         }
 274 #endif /* ACE_JAWS_BASELINE */
 275     }
 276   else if (cf.file ()->get_handle () != ACE_INVALID_HANDLE
 277            && cf.mmap () == 0)
 278     {
 279       this->transmit_file (ioh,
 280                            cf.file ()->get_handle (),
 281                            header, header_size,
 282                            trailer, trailer_size);
 283       return;
 284     }
 285   else
 286     {
 287       result = -1;
 288     }
 289
 290   if (result != 0)
 291     {
 292       ioh->transmit_file_error (result);
 293     }
 294 }
 295
 296 void
 297 JAWS_Synch_IO::send_confirmation_message (JAWS_IO_Handler *ioh,
 298                                           const char *buffer,
 299                                           unsigned int length)
 300 {
 301   this->send_message (ioh, buffer, length);
 302   ioh->confirmation_message_complete ();
 303 }
 304
 305 void
 306 JAWS_Synch_IO::send_error_message (JAWS_IO_Handler *ioh,
 307                                    const char *buffer,
 308                                    unsigned int length)
 309 {
 310   this->send_message (ioh, buffer, length);
 311   ioh->error_message_complete ();
 312 }
 313
 314 void
 315 JAWS_Synch_IO::send_message (JAWS_IO_Handler *ioh,
 316                              const char *buffer,
 317                              unsigned int length)
 318 {
 319   ACE_SOCK_Stream stream;
 320   stream.set_handle (ioh->handle ());
 321   stream.send_n (buffer, length);
 322 }
 323
 324 // This only works on asynch I/O-capable systems.
 325 #if defined (ACE_HAS_WIN32_OVERLAPPED_IO) || defined (ACE_HAS_AIO_CALLS)
 326
 327 JAWS_Asynch_IO::JAWS_Asynch_IO ()
 328 {
 329 }
 330
 331 JAWS_Asynch_IO::~JAWS_Asynch_IO ()
 332 {
 333   if (this->handle_ != ACE_INVALID_HANDLE)
 334     ACE_OS::closesocket (this->handle_);
 335 }
 336
 337 void
 338 JAWS_Asynch_IO::accept (JAWS_IO_Handler *ioh,
 339                         ACE_Message_Block *,
 340                         unsigned int)
 341 {
 342   JAWS_TRACE ("JAWS_Asynch_IO::accept");
 343
 344   ioh->idle ();
 345
 346   JAWS_Data_Block *db = ioh->message_block ();
 347   //ACE_HANDLE listen_handle = db->policy ()->acceptor ()->get_handle ();
 348
 349   //JAWS_Asynch_IO_Handler *aioh =
 350   //  dynamic_cast<JAWS_Asynch_IO_Handler *> (ioh);
 351
 352   size_t bytes_to_read = JAWS_Data_Block::JAWS_DATA_BLOCK_SIZE;
 353
 354   if (db->policy ()->acceptor ()->accept (bytes_to_read, ioh) == -1)
 355     ioh->accept_error ();
 356 }
 357
 358 void
 359 JAWS_Asynch_IO::read (JAWS_IO_Handler *ioh,
 360                       ACE_Message_Block* mb,
 361                       unsigned int size)
 362 {
 363   JAWS_TRACE ("JAWS_Asynch_IO::read");
 364
 365   ioh->idle ();
 366
 367   JAWS_Asynch_IO_Handler *aioh =
 368     dynamic_cast<JAWS_Asynch_IO_Handler *> (ioh);
 369
 370   ACE_Asynch_Read_Stream ar;
 371
 372   if (ar.open (*(aioh->handler ()), aioh->handle ()) == -1
 373       || ar.read (*mb, size) == -1)
 374     aioh->read_error ();
 375 }
 376
 377 void
 378 JAWS_Asynch_IO::receive_file (JAWS_IO_Handler *ioh,
 379                               const char *filename,
 380                               void *initial_data,
 381                               unsigned int initial_data_length,
 382                               unsigned int entire_length)
 383 {
 384   JAWS_TRACE ("JAWS_Asynch_IO::receive_file");
 385
 386   ioh->idle ();
 387
 388   JAWS_Asynch_IO_Handler *aioh =
 389     dynamic_cast<JAWS_Asynch_IO_Handler *> (ioh);
 390
 391   ACE_Message_Block *mb = 0;
 392   ACE_Filecache_Handle *handle;
 393
 394   ACE_NEW (handle, ACE_Filecache_Handle (filename, entire_length, ACE_NOMAP));
 395
 396   int result = handle->error ();
 397
 398   if (result == ACE_Filecache_Handle::ACE_SUCCESS)
 399     {
 400       ACE_OS::memcpy (handle->address (),
 401                       initial_data,
 402                       initial_data_length);
 403
 404       int bytes_to_read = entire_length - initial_data_length;
 405
 406       ACE_NEW (mb, ACE_Message_Block ((char *)handle->address ()
 407                                       + initial_data_length, bytes_to_read));
 408
 409       if (mb == 0)
 410         {
 411           errno = ENOMEM;
 412           result = -1;
 413         }
 414       else
 415         {
 416           ACE_Asynch_Read_Stream ar;
 417
 418           if (ar.open (*(aioh->handler ()), aioh->handle ()) == -1
 419               || ar.read (*mb, mb->size () - mb->length (), handle) == -1)
 420             result = -1;
 421         }
 422     }
 423
 424   if (result != ACE_Filecache_Handle::ACE_SUCCESS)
 425     {
 426       this->handler_->receive_file_error (result);
 427       delete mb;
 428       delete handle;
 429     }
 430 }
 431
 432 void
 433 JAWS_Asynch_IO::transmit_file (JAWS_IO_Handler *ioh,
 434                                ACE_HANDLE handle,
 435                                const char *header,
 436                                unsigned int header_size,
 437                                const char *trailer,
 438                                unsigned int trailer_size)
 439 {
 440   JAWS_TRACE ("JAWS_Asynch_IO::transmit_file");
 441
 442   ioh->idle ();
 443
 444   JAWS_Asynch_IO_Handler *aioh =
 445     dynamic_cast<JAWS_Asynch_IO_Handler *> (ioh);
 446
 447   ACE_Asynch_Transmit_File::Header_And_Trailer *header_and_trailer = 0;
 448
 449   int result = 0;
 450
 451   if (handle != ACE_INVALID_HANDLE)
 452     {
 453       ACE_Message_Block hdr_mb (header, header_size);
 454       ACE_Message_Block trl_mb (trailer, trailer_size);
 455
 456       header_and_trailer =
 457         new ACE_Asynch_Transmit_File::Header_And_Trailer (hdr_mb.duplicate (),
 458                                                           header_size,
 459                                                           trl_mb.duplicate (),
 460                                                           trailer_size);
 461
 462       ACE_Asynch_Transmit_File tf;
 463
 464       if (tf.open (*(aioh->handler ()), aioh->handle ()) == -1
 465           || tf.transmit_file (handle, // file handle
 466                                header_and_trailer, // header and trailer data
 467                                0,  // bytes_to_write
 468                                0,  // offset
 469                                0,  // offset_high
 470                                0,  // bytes_per_send
 471                                0,  // flags
 472                                0 // act
 473                                ) == -1)
 474         result = -1;
 475     }
 476
 477   if (result != 0)
 478     {
 479       ioh->transmit_file_error (result);
 480       delete header_and_trailer;
 481     }
 482 }
 483
 484 void
 485 JAWS_Asynch_IO::transmit_file (JAWS_IO_Handler *ioh,
 486                                const char *filename,
 487                                const char *header,
 488                                unsigned int header_size,
 489                                const char *trailer,
 490                                unsigned int trailer_size)
 491 {
 492   int result = 0;
 493
 494   JAWS_TRACE ("JAWS_Asynch_IO::transmit_file");
 495
 496   ioh->idle ();
 497
 498   JAWS_Asynch_IO_Handler *aioh =
 499     dynamic_cast<JAWS_Asynch_IO_Handler *> (ioh);
 500
 501   ACE_Asynch_Transmit_File::Header_And_Trailer *header_and_trailer = 0;
 502   JAWS_Cached_FILE *cf = new JAWS_Cached_FILE (filename);
 503
 504   if (cf->file ()->get_handle () != ACE_INVALID_HANDLE)
 505     {
 506       ACE_Message_Block hdr_mb (header, header_size);
 507       ACE_Message_Block trl_mb (trailer, trailer_size);
 508
 509       header_and_trailer = new ACE_Asynch_Transmit_File::Header_And_Trailer
 510         (hdr_mb.duplicate (), header_size, trl_mb.duplicate (), trailer_size);
 511
 512       ACE_Asynch_Transmit_File tf;
 513
 514       if (tf.open (*(aioh->handler ()), aioh->handle ()) == -1
 515           || tf.transmit_file (cf->file ()->get_handle (), // file handle
 516                                header_and_trailer, // header and trailer data
 517                                0,  // bytes_to_write
 518                                0,  // offset
 519                                0,  // offset_high
 520                                0,  // bytes_per_send
 521                                0,  // flags
 522                                cf // act
 523                                ) == -1)
 524         result = -1;
 525     }
 526
 527   if (result != 0)
 528     {
 529       ioh->transmit_file_error (result);
 530       delete header_and_trailer;
 531       delete cf;
 532     }
 533 }
 534
 535 void
 536 JAWS_Asynch_IO::send_confirmation_message (JAWS_IO_Handler *ioh,
 537                                            const char *buffer,
 538                                            unsigned int length)
 539 {
 540   this->send_message (ioh, buffer, length, CONFIRMATION);
 541 }
 542
 543 void
 544 JAWS_Asynch_IO::send_error_message (JAWS_IO_Handler *ioh,
 545                                     const char *buffer,
 546                                     unsigned int length)
 547 {
 548   this->send_message (ioh, buffer, length, ERROR_MESSAGE);
 549 }
 550
 551 void
 552 JAWS_Asynch_IO::send_message (JAWS_IO_Handler *ioh,
 553                               const char *buffer,
 554                               unsigned int length,
 555                               long act)
 556 {
 557   ioh->idle ();
 558
 559   JAWS_Asynch_IO_Handler *aioh =
 560     dynamic_cast<JAWS_Asynch_IO_Handler *> (ioh);
 561
 562   ACE_Message_Block *mb = 0;
 563   ACE_NEW (mb, ACE_Message_Block (buffer, length));
 564
 565   if (mb == 0)
 566     {
 567       this->handler_->error_message_complete ();
 568       return;
 569     }
 570
 571   ACE_Asynch_Write_Stream aw;
 572   if (aw.open (*(aioh->handler ()), aioh->handle ()) == -1
 573       || aw.write (*mb, length, (void *) static_cast<intptr_t> (act)) == -1)
 574     {
 575       mb->release ();
 576
 577       if (act == CONFIRMATION)
 578         ioh->confirmation_message_complete ();
 579       else
 580         ioh->error_message_complete ();
 581     }
 582 }
 583
 584 void
 585 JAWS_Asynch2_IO::accept (JAWS_IO_Handler *,
 586                          ACE_Message_Block *,
 587                          unsigned int)
 588 {
 589 }
 590
 591 #endif /* ACE_HAS_WIN32_OVERLAPPED_IO || ACE_HAS_AIO_CALLS */
 592