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