ACE/apps/JAWS2/JAWS/Pipeline_Tasks.cpp

   1 #include "JAWS/Jaws_IO.h"
   2 #include "JAWS/Pipeline_Tasks.h"
   3 #include "JAWS/Pipeline_Handler_T.h"
   4 #include "JAWS/Data_Block.h"
   5 #include "JAWS/IO_Handler.h"
   6 #include "JAWS/Policy.h"
   7
   8
   9 JAWS_Pipeline_Handler::JAWS_Pipeline_Handler ()
  10   : policy_ (0)
  11 {
  12 }
  13
  14 JAWS_Pipeline_Handler::~JAWS_Pipeline_Handler ()
  15 {
  16 }
  17
  18 int
  19 JAWS_Pipeline_Handler::put (ACE_Message_Block *mb, ACE_Time_Value *tv)
  20 {
  21   JAWS_Data_Block *db = dynamic_cast<JAWS_Data_Block *> (mb);
  22   JAWS_IO_Handler *ioh = db->io_handler ();
  23
  24   // guarantee the handler remains for the duration of this call
  25   ioh->acquire ();
  26
  27   int status = this->handle_put (db, tv);
  28
  29   if (status != -1 && status != 2)
  30     {
  31       JAWS_Pipeline_Handler *task = ioh->task ();
  32       JAWS_Pipeline_Handler *next
  33         = dynamic_cast<JAWS_Pipeline_Handler *> (task->next ());
  34
  35       ioh->task (next);
  36     }
  37
  38   ioh->release ();
  39
  40   return status;
  41 }
  42
  43 JAWS_Dispatch_Policy *
  44 JAWS_Pipeline_Handler::policy ()
  45 {
  46   return this->policy_;
  47 }
  48
  49 void
  50 JAWS_Pipeline_Handler::policy (JAWS_Dispatch_Policy *policy)
  51 {
  52   this->policy_ = policy;
  53 }
  54
  55 int
  56 JAWS_Pipeline_Accept_Task::put (ACE_Message_Block *mb, ACE_Time_Value *tv)
  57 {
  58   JAWS_Data_Block *db = dynamic_cast<JAWS_Data_Block *> (mb);
  59
  60   JAWS_Pipeline_Handler *task = db->task ();
  61   JAWS_Pipeline_Handler *next
  62     = dynamic_cast<JAWS_Pipeline_Handler *> (task->next ());
  63
  64   JAWS_IO_Handler *ioh = this->new_handler (db);
  65   if (ioh == 0)
  66     {
  67       ACE_ERROR ((LM_ERROR, "%p\n", "JAWS_Pipeline_Accept_Task::put"));
  68       return -1;
  69     }
  70
  71   ioh->acquire ();
  72
  73   ioh->task (next);
  74   db->io_handler (ioh);
  75
  76   int result = this->handle_put (ioh->message_block (), tv);
  77
  78   ioh->release ();
  79
  80   return result;
  81 }
  82
  83 int
  84 JAWS_Pipeline_Accept_Task::handle_put (JAWS_Data_Block *data,
  85                                        ACE_Time_Value *)
  86 {
  87   int result = -1;
  88
  89   // JAWS_Data_Block should contain an INET_Addr and an IO
  90   JAWS_IO_Handler *handler = data->io_handler ();
  91   JAWS_Dispatch_Policy *policy = this->policy ();
  92
  93   if (policy == 0) policy = data->policy ();
  94
  95   // data->policy ()->update (handler);
  96
  97   JAWS_IO *io = policy->io ();
  98   io->accept (handler);
  99
 100   // When accept returns, the resulting handle should be stored into
 101   // the JAWS_DATA_BLOCK somewhere.
 102
 103   // Check the handler for status of the io call
 104   switch (handler->status ())
 105     {
 106     case JAWS_IO_Handler::ACCEPT_OK:
 107       {
 108         ACE_DEBUG ((LM_DEBUG, "(%t) ACCEPT_OK\n"));
 109         result = 0;
 110         JAWS_TRACE ("JAWS_Pipeline_Accept_Task::handle_put ACCEPT_OK");
 111         // Move on to next stage in pipeline
 112         break;
 113       }
 114     case JAWS_IO_Handler::ACCEPT_ERROR:
 115       {
 116         ACE_DEBUG ((LM_DEBUG, "(%t) ACCEPT_ERROR\n"));
 117         result = -1;
 118         JAWS_TRACE ("JAWS_Pipeline_Accept_Task::handle_put ACCEPT_ERROR");
 119         // Should recycle the thread
 120         break;
 121       }
 122     default:
 123       {
 124         result = 1;
 125         JAWS_TRACE ("JAWS_Pipeline_Accept_Task::handle_put ACCEPT_IDLE");
 126         // Should mean that the IO is asynchronous, and the word isn't out
 127         // yet.
 128         break;
 129       }
 130     }
 131
 132   // In asynchronous and synchronous models, we can --
 133   //   have the io_handler set the new task in the data_block
 134
 135   // In asynchronous model, we can --
 136   //   insert a wait task into the task queue
 137
 138   ACE_DEBUG ((LM_DEBUG, "(%t) Returning %d\n", result));
 139   return result;
 140 }
 141
 142 JAWS_IO_Handler *
 143 JAWS_Pipeline_Accept_Task::new_handler (JAWS_Data_Block *data)
 144 {
 145   // Create a new handler and message block
 146   JAWS_Data_Block *ndb = new JAWS_Data_Block (*data);
 147   if (ndb == 0)
 148     {
 149       JAWS_TRACE ("JAWS_Pipeline_Accept_Task::new_handler, failed DB");
 150       return 0;
 151     }
 152
 153   JAWS_Dispatch_Policy *policy =
 154     (this->policy () == 0) ? data->policy () : this->policy ();
 155   JAWS_IO_Handler_Factory *ioh_factory = policy->ioh_factory ();
 156
 157   JAWS_IO_Handler *nioh = ioh_factory->create_io_handler ();
 158   if (nioh == 0)
 159     {
 160       delete ndb;
 161       return 0;
 162     }
 163
 164   ndb->io_handler (nioh);
 165   nioh->task (data->task ());
 166   nioh->message_block (ndb);
 167
 168   return nioh;
 169 }
 170
 171 int
 172 JAWS_Pipeline_Done_Task::put (ACE_Message_Block *mb, ACE_Time_Value *)
 173 {
 174   JAWS_TRACE ("JAWS_Pipeline_Done_Task::put");
 175
 176   JAWS_Data_Block *data = dynamic_cast<JAWS_Data_Block *> (mb);
 177
 178   JAWS_IO_Handler *handler = data->io_handler ();
 179   JAWS_Dispatch_Policy *policy = this->policy ();
 180   if (policy == 0) policy = data->policy ();
 181
 182   // JAWS_IO *io = policy->io ();
 183
 184   data->task (0);
 185   data->io_handler (0);
 186
 187   if (handler)
 188     handler->done ();
 189
 190   // hack, let Concurrency know we are done.
 191   return -2;
 192 }
 193
 194 int
 195 JAWS_Pipeline_Done_Task::handle_put (JAWS_Data_Block *, ACE_Time_Value *)
 196 {
 197   return 0;
 198 }
 199