ACE/apps/JAWS2/JAWS/IO_Handler.cpp

   1 #include "ace/Proactor.h"
   2 #include "ace/Filecache.h"
   3 #include "ace/OS_NS_unistd.h"
   4 #include "ace/OS_NS_sys_socket.h"
   5
   6 #include "JAWS/Jaws_IO.h"
   7 #include "JAWS/IO_Handler.h"
   8 #include "JAWS/IO_Acceptor.h"
   9 #include "JAWS/Data_Block.h"
  10 #include "JAWS/Policy.h"
  11 #include "JAWS/Waiter.h"
  12 #include "JAWS/Filecache.h"
  13
  14 JAWS_Abstract_IO_Handler::~JAWS_Abstract_IO_Handler ()
  15 {
  16 }
  17
  18 JAWS_IO_Handler_Factory::~JAWS_IO_Handler_Factory ()
  19 {
  20 }
  21
  22 JAWS_IO_Handler *
  23 JAWS_IO_Handler_Factory::create_io_handler ()
  24 {
  25   JAWS_TRACE ("JAWS_IO_Handler_Factory::create");
  26
  27   JAWS_IO_Handler *handler;
  28   handler = new JAWS_IO_Handler (this);
  29
  30   return handler;
  31 }
  32
  33 void
  34 JAWS_IO_Handler_Factory::destroy_io_handler (JAWS_IO_Handler *handler)
  35 {
  36   JAWS_TRACE ("JAWS_IO_Handler_Factory::destroy");
  37   if (handler != 0)
  38     {
  39       delete handler->message_block ();
  40       delete handler;
  41     }
  42 }
  43
  44 JAWS_IO_Handler::JAWS_IO_Handler (JAWS_IO_Handler_Factory *factory)
  45   : status_ (0),
  46     mb_ (0),
  47     handle_ (ACE_INVALID_HANDLE),
  48     task_ (0),
  49     factory_ (factory)
  50 {
  51 }
  52
  53 JAWS_IO_Handler::~JAWS_IO_Handler ()
  54 {
  55   this->mb_ = 0;
  56   this->status_ = 0;
  57   this->task_ = 0;
  58   this->factory_ = 0;
  59
  60   ACE_OS::closesocket (this->handle_);
  61   this->handle_ = ACE_INVALID_HANDLE;
  62 }
  63
  64 void
  65 JAWS_IO_Handler::accept_complete (ACE_HANDLE handle)
  66 {
  67   // callback into pipeline task, notify that the accept has completed
  68   this->handle_ = handle;
  69   this->status_ |= ACCEPT_OK;
  70   this->status_ &= (ACCEPT_OK+1);
  71
  72   JAWS_Dispatch_Policy *policy = this->mb_->policy ();
  73
  74   // Do this so that Thread Per Request can spawn a new thread
  75   policy->concurrency ()->activate_hook ();
  76 }
  77
  78 void
  79 JAWS_IO_Handler::accept_error ()
  80 {
  81   // callback into pipeline task, notify that the accept has failed
  82   this->status_ |= ACCEPT_ERROR;
  83   this->status_ &= (ACCEPT_ERROR+1);
  84 }
  85
  86 void
  87 JAWS_IO_Handler::read_complete (ACE_Message_Block *data)
  88 {
  89   ACE_UNUSED_ARG (data);
  90   // We can call back into the pipeline task at this point
  91   // this->pipeline_->read_complete (data);
  92   this->status_ |= READ_OK;
  93   this->status_ &= (READ_OK+1);
  94 }
  95
  96 void
  97 JAWS_IO_Handler::read_error ()
  98 {
  99   // this->pipeline_->read_error ();
 100   this->status_ |= READ_ERROR;
 101   this->status_ &= (READ_ERROR+1);
 102 }
 103
 104 void
 105 JAWS_IO_Handler::transmit_file_complete ()
 106 {
 107   JAWS_TRACE ("JAWS_IO_Handler::transmit_file_complete");
 108   // this->pipeline_->transmit_file_complete ();
 109   this->status_ |= TRANSMIT_OK;
 110   this->status_ &= (TRANSMIT_OK+1);
 111 }
 112
 113 void
 114 JAWS_IO_Handler::transmit_file_error (int result)
 115 {
 116   JAWS_TRACE ("JAWS_IO_Handler::transmit_file_error");
 117   ACE_UNUSED_ARG (result);
 118   // this->pipeline_->transmit_file_complete (result);
 119   this->status_ |= TRANSMIT_ERROR;
 120   this->status_ &= (TRANSMIT_ERROR+1);
 121 }
 122
 123 void
 124 JAWS_IO_Handler::receive_file_complete ()
 125 {
 126   this->status_ |= RECEIVE_OK;
 127   this->status_ &= (RECEIVE_OK+1);
 128 }
 129
 130 void
 131 JAWS_IO_Handler::receive_file_error (int result)
 132 {
 133   ACE_UNUSED_ARG(result);
 134   this->status_ |= RECEIVE_ERROR;
 135   this->status_ &= (RECEIVE_ERROR+1);
 136 }
 137
 138 void
 139 JAWS_IO_Handler::write_error ()
 140 {
 141   ACE_DEBUG ((LM_DEBUG, " (%t) error in writing response\n"));
 142
 143   this->status_ |= WRITE_ERROR;
 144   this->status_ &= (WRITE_ERROR+1);
 145   this->done ();
 146 }
 147
 148 void
 149 JAWS_IO_Handler::confirmation_message_complete ()
 150 {
 151   this->status_ |= WRITE_OK;
 152   this->status_ &= (WRITE_OK+1);
 153 }
 154
 155 void
 156 JAWS_IO_Handler::error_message_complete ()
 157 {
 158   this->status_ |= WRITE_OK;
 159   this->status_ &= (WRITE_OK+1);
 160 }
 161
 162 JAWS_IO_Handler_Factory *
 163 JAWS_IO_Handler::factory ()
 164 {
 165   return this->factory_;
 166 }
 167
 168 ACE_HANDLE
 169 JAWS_IO_Handler::handle () const
 170 {
 171   return this->handle_;
 172 }
 173
 174 void
 175 JAWS_IO_Handler::task (JAWS_Pipeline_Handler *ph)
 176 {
 177   this->task_ = ph;
 178 }
 179
 180 JAWS_Pipeline_Handler *
 181 JAWS_IO_Handler::task ()
 182 {
 183   return this->task_;
 184 }
 185
 186 void
 187 JAWS_IO_Handler::message_block (JAWS_Data_Block *mb)
 188 {
 189   this->mb_ = mb;
 190 }
 191
 192 JAWS_Data_Block *
 193 JAWS_IO_Handler::message_block ()
 194 {
 195   return this->mb_;
 196 }
 197
 198 void
 199 JAWS_IO_Handler::done ()
 200 {
 201   this->factory ()->destroy_io_handler (this);
 202 }
 203
 204 int
 205 JAWS_IO_Handler::status ()
 206 {
 207   return this->status_;
 208 }
 209
 210 void
 211 JAWS_IO_Handler::idle ()
 212 {
 213   this->status_ &= (IDLE+1);
 214 }
 215
 216 void
 217 JAWS_IO_Handler::acquire ()
 218 {
 219 }
 220
 221 void
 222 JAWS_IO_Handler::lock ()
 223 {
 224 }
 225
 226 void
 227 JAWS_IO_Handler::release ()
 228 {
 229 }
 230
 231 #if defined (ACE_HAS_WIN32_OVERLAPPED_IO) || defined (ACE_HAS_AIO_CALLS)
 232
 233 JAWS_Asynch_IO_Handler_Factory::~JAWS_Asynch_IO_Handler_Factory ()
 234 {
 235 }
 236
 237 JAWS_IO_Handler *
 238 JAWS_Asynch_IO_Handler_Factory::create_io_handler ()
 239 {
 240   JAWS_TRACE ("JAWS_Asynch_IO_Handler_Factory::create");
 241
 242   JAWS_Asynch_IO_Handler *handler = 0;
 243   handler = new JAWS_Asynch_IO_Handler (this);
 244
 245   return handler;
 246 }
 247
 248 void
 249 JAWS_Asynch_IO_Handler_Factory::destroy_io_handler (JAWS_IO_Handler *handler)
 250 {
 251   JAWS_TRACE ("JAWS_IO_Handler_Factory::destroy");
 252
 253   if (handler != 0)
 254     {
 255       //cerr << "(" << thr_self () << ") locking for destruction: " << handler << endl;
 256       handler->lock ();
 257       delete handler->message_block ();
 258       handler->message_block (0);
 259       delete handler;
 260     }
 261 }
 262
 263
 264 JAWS_Asynch_IO_Handler::JAWS_Asynch_IO_Handler (JAWS_Asynch_IO_Handler_Factory *factory)
 265   : JAWS_IO_Handler (factory),
 266     handler_ (0)
 267 {
 268   this->status_ = 1;
 269 }
 270
 271 JAWS_Asynch_IO_Handler::~JAWS_Asynch_IO_Handler ()
 272 {
 273   delete this->handler_;
 274   this->handler_ = 0;
 275 }
 276
 277 ACE_Handler *
 278 JAWS_Asynch_IO_Handler::handler ()
 279 {
 280   return this->handler_;
 281 }
 282
 283 void
 284 JAWS_Asynch_IO_Handler::acquire ()
 285 {
 286   //cerr << "(" << thr_self () << ") acquire handler: " << this << endl;
 287   this->count_.acquire_read ();
 288 }
 289
 290 void
 291 JAWS_Asynch_IO_Handler::lock ()
 292 {
 293   //cerr << "(" << thr_self () << ") locking handler: " << this << endl;
 294   this->count_.acquire_write ();
 295 }
 296
 297 void
 298 JAWS_Asynch_IO_Handler::release ()
 299 {
 300   //cerr << "(" << thr_self () << ") release handler: " << this << endl;
 301   this->count_.release ();
 302 }
 303
 304 JAWS_Asynch_Handler::JAWS_Asynch_Handler ()
 305   : ioh_ (0)
 306 {
 307   this->proactor (ACE_Proactor::instance ());
 308 }
 309
 310 JAWS_Asynch_Handler::~JAWS_Asynch_Handler ()
 311 {
 312 }
 313
 314 void
 315 JAWS_Asynch_Handler::open (ACE_HANDLE h,
 316                            ACE_Message_Block &mb)
 317 {
 318   JAWS_TRACE ("JAWS_Asynch_Handler::open");
 319
 320   // This currently does nothing, but just in case.
 321   ACE_Service_Handler::open (h, mb);
 322
 323   // ioh_ set from the ACT hopefully
 324   //this->dispatch_handler ();
 325
 326 #if !defined (ACE_WIN32)
 327   // Assume at this point there is no data.
 328   mb.rd_ptr (mb.wr_ptr ());
 329   mb.crunch ();
 330 #else
 331   // AcceptEx reads some initial data from the socket.
 332   this->handler ()->message_block ()->copy (mb.rd_ptr (), mb.length ());
 333 #endif
 334
 335   ACE_Asynch_Accept_Result_Impl *fake_result
 336     = ACE_Proactor::instance ()->create_asynch_accept_result
 337       (this->proxy (), JAWS_IO_Asynch_Acceptor_Singleton::instance ()->get_handle (),
 338        h, mb, JAWS_Data_Block::JAWS_DATA_BLOCK_SIZE,
 339        this->ioh_, ACE_INVALID_HANDLE, 0);
 340
 341   this->handler ()->handler_ = this;
 342
 343   fake_result->complete (0, 1, 0);
 344 }
 345
 346 void
 347 JAWS_Asynch_Handler::act (const void *act_ref)
 348 {
 349   JAWS_TRACE ("JAWS_Asynch_Handler::act");
 350
 351   // Set the ioh from the act
 352   this->ioh_ = (JAWS_Asynch_IO_Handler *) act_ref;
 353 }
 354
 355 #if 0
 356 ACE_HANDLE
 357 JAWS_Asynch_Handler::handle () const
 358 {
 359   return this->ioh_->handle ();
 360 }
 361 #endif
 362
 363 void
 364 JAWS_Asynch_Handler::dispatch_handler ()
 365 {
 366 #if 0
 367   // A future version of ACE will support this.
 368   ACE_Thread_ID tid = ACE_OS::thr_self ();
 369 #else
 370   // Do it this way for now
 371   ACE_thread_t thr_name;
 372   thr_name = ACE_OS::thr_self ();
 373
 374   JAWS_Thread_ID tid (thr_name);
 375 #endif /* 0 */
 376
 377   JAWS_IO_Handler **iohref = JAWS_Waiter_Singleton::instance ()->find (tid);
 378
 379   *iohref = this->handler ();
 380 }
 381
 382 void
 383 JAWS_Asynch_Handler::handle_read_stream (const ACE_Asynch_Read_Stream::Result
 384                                          &result)
 385 {
 386   JAWS_TRACE ("JAWS_Asynch_Handler::handle_read_stream");
 387
 388   this->dispatch_handler ();
 389
 390   if (result.act () != 0)
 391     {
 392       // This callback is for io->receive_file()
 393       JAWS_TRACE ("JAWS_Asynch_Handler::handle_read_stream (recv_file)");
 394
 395       int code = 0;
 396       if (result.success () && result.bytes_transferred () != 0)
 397         {
 398           if (result.message_block ().length ()
 399               == result.message_block ().size ())
 400             code = ACE_Filecache_Handle::ACE_SUCCESS;
 401           else
 402             {
 403               ACE_Asynch_Read_Stream ar;
 404               if (ar.open (*this, this->handler ()->handle ()) == -1
 405                   || ar.read (result.message_block (),
 406                               result.message_block ().size ()
 407                               - result.message_block ().length (),
 408                               result.act ()) == -1)
 409                 code = -1;
 410               else
 411                 return;
 412             }
 413         }
 414       else
 415         code = -1;
 416
 417       if (code == ACE_Filecache_Handle::ACE_SUCCESS)
 418         this->handler ()->receive_file_complete ();
 419       else
 420         this->handler ()->receive_file_error (code);
 421
 422       result.message_block ().release ();
 423       delete (ACE_Filecache_Handle *) result.act ();
 424     }
 425   else
 426     {
 427       // This callback is for this->read()
 428       JAWS_TRACE ("JAWS_Asynch_Handler::handle_read_stream (read)");
 429
 430       if (result.success ()
 431           && result.bytes_transferred () != 0)
 432         this->handler ()->read_complete (&result.message_block ());
 433       else
 434         this->handler ()->read_error ();
 435     }
 436 }
 437
 438 void
 439 JAWS_Asynch_Handler::handle_write_stream (const ACE_Asynch_Write_Stream::Result
 440                                           &result)
 441 {
 442   this->dispatch_handler ();
 443
 444   result.message_block ().release ();
 445
 446   if (result.act () == (void *) JAWS_Asynch_IO::CONFIRMATION)
 447     this->handler ()->confirmation_message_complete ();
 448   else
 449     this->handler ()->error_message_complete ();
 450 }
 451
 452 void
 453 JAWS_Asynch_Handler::handle_transmit_file (const
 454                                            ACE_Asynch_Transmit_File::Result
 455                                            &result)
 456 {
 457   this->dispatch_handler ();
 458
 459   if (result.success ())
 460     this->handler ()->transmit_file_complete ();
 461   else
 462     this->handler ()->transmit_file_error (-1);
 463
 464   result.header_and_trailer ()->header ()->release ();
 465   result.header_and_trailer ()->trailer ()->release ();
 466   delete result.header_and_trailer ();
 467   delete (JAWS_Cached_FILE *) result.act ();
 468 }
 469
 470 void
 471 JAWS_Asynch_Handler::handle_accept (const ACE_Asynch_Accept::Result &result)
 472 {
 473   JAWS_TRACE ("JAWS_Asynch_Handler::handle_accept");
 474   this->dispatch_handler ();
 475
 476   if (result.success ())
 477     {
 478       JAWS_TRACE ("JAWS_Asynch_Handler::handle_accept, success");
 479       this->handler ()->accept_complete (result.accept_handle ());
 480     }
 481   else
 482     this->handler ()->accept_error ();
 483 }
 484
 485 void
 486 JAWS_Asynch_Handler::handler (JAWS_Asynch_IO_Handler *ioh)
 487 {
 488   this->ioh_ = ioh;
 489 }
 490
 491 JAWS_Asynch_IO_Handler *
 492 JAWS_Asynch_Handler::handler ()
 493 {
 494   return this->ioh_;
 495 }
 496
 497 #endif /* ACE_HAS_WIN32_OVERLAPPED_IO || ACE_HAS_AIO_CALLS */