Made many changes to ensure race conditions do not occur. Among them --

a new return value to indicate the Done task has finished.  Additional
status values to indicate which operations are asynch and which are
synch.  RW_MUTEX protection around the destruction of the IO_Handler.
Creation of the Done task.

6 files changed, 212 insertions, 71 deletions

diff --git a/apps/JAWS/PROTOTYPE/JAWS/Concurrency.cpp b/apps/JAWS/PROTOTYPE/JAWS/Concurrency.cpp
index 0824039f536..3f1d589ce32 100644
--- a/apps/JAWS/PROTOTYPE/JAWS/Concurrency.cpp
+++ b/apps/JAWS/PROTOTYPE/JAWS/Concurrency.cpp
@@ -154,8 +154,8 @@ JAWS_Concurrency_Base::svc_hook (JAWS_Data_Block *ts_db)
       // the task should set the handler to the appropriate next step
       result = task->put (mb);
 
-      // For when result == 0
-      handler = mb->io_handler ();
+      if (result == 0)
+        handler = mb->io_handler ();
 
       if (result == 1 || result == 2)
         {
@@ -164,9 +164,7 @@ JAWS_Concurrency_Base::svc_hook (JAWS_Data_Block *ts_db)
 
           // We need a way to destroy all the handlers created by the
           // Asynch_Acceptor.  Figure this out later.
-          do
-            handler = waiter->wait_for_completion (waiter_index);
-          while (handler && handler->count () > 0);
+          handler = waiter->wait_for_completion (waiter_index);
           result = (handler == 0) ? -1 : 0;
         }
 
@@ -175,7 +173,7 @@ JAWS_Concurrency_Base::svc_hook (JAWS_Data_Block *ts_db)
           // something wrong.
           JAWS_TRACE ("JAWS_Concurrency_Base::svc_hook, negative result");
           ACE_ERROR ((LM_ERROR, "%p\n", "JAWS_Concurrency_Base::svc_hook"));
-          if (handler)
+          if (result > -2 && handler)
             handler->done ();
           handler = 0;
           JAWS_IO_Handler **ioh = waiter->find (waiter_index);
diff --git a/apps/JAWS/PROTOTYPE/JAWS/IO_Handler.cpp b/apps/JAWS/PROTOTYPE/JAWS/IO_Handler.cpp
index 82567413a0e..2ec4a39b887 100644
--- a/apps/JAWS/PROTOTYPE/JAWS/IO_Handler.cpp
+++ b/apps/JAWS/PROTOTYPE/JAWS/IO_Handler.cpp
@@ -26,8 +26,8 @@ JAWS_IO_Handler_Factory::create_io_handler (void)
 {
   JAWS_TRACE ("JAWS_IO_Handler_Factory::create");
 
-  JAWS_Asynch_IO_Handler *handler;
-  handler = new JAWS_Asynch_IO_Handler (this);
+  JAWS_IO_Handler *handler;
+  handler = new JAWS_IO_Handler (this);
 
   return handler;
 }
@@ -48,11 +48,7 @@ JAWS_IO_Handler::JAWS_IO_Handler (JAWS_IO_Handler_Factory *factory)
     mb_ (0),
     handle_ (ACE_INVALID_HANDLE),
     task_ (0),
-    factory_ (factory),
-    count_ (0)
-#if defined (ACE_WIN32) || defined (ACE_HAS_AIO_CALLS)
-  , handler_ (0)
-#endif /* defined (ACE_WIN32) || defined (ACE_HAS_AIO_CALLS) */
+    factory_ (factory)
 {
 }
 
@@ -65,11 +61,6 @@ JAWS_IO_Handler::~JAWS_IO_Handler (void)
 
   ACE_OS::close (this->handle_);
   this->handle_ = ACE_INVALID_HANDLE;
-
-#if defined (ACE_WIN32) || defined (ACE_HAS_AIO_CALLS)
-  delete this->handler_;
-  this->handler_ = 0;
-#endif /* defined (ACE_WIN32) || defined (ACE_HAS_AIO_CALLS) */
 }
 
 void
@@ -77,7 +68,8 @@ JAWS_IO_Handler::accept_complete (ACE_HANDLE handle)
 {
   // callback into pipeline task, notify that the accept has completed
   this->handle_ = handle;
-  this->status_ = ACCEPT_OK;
+  this->status_ |= ACCEPT_OK;
+  this->status_ &= (ACCEPT_OK+1);
 
   JAWS_Dispatch_Policy *policy = this->mb_->policy ();
 
@@ -89,7 +81,8 @@ void
 JAWS_IO_Handler::accept_error (void)
 {
   // callback into pipeline task, notify that the accept has failed
-  this->status_ = ACCEPT_ERROR;
+  this->status_ |= ACCEPT_ERROR;
+  this->status_ &= (ACCEPT_ERROR+1);
 }
 
 void
@@ -98,14 +91,16 @@ JAWS_IO_Handler::read_complete (ACE_Message_Block *data)
   ACE_UNUSED_ARG (data);
   // We can call back into the pipeline task at this point
   // this->pipeline_->read_complete (data);
-  this->status_ = READ_OK;
+  this->status_ |= READ_OK;
+  this->status_ &= (READ_OK+1);
 }
 
 void
 JAWS_IO_Handler::read_error (void)
 {
   // this->pipeline_->read_error ();
-  this->status_ = READ_ERROR;
+  this->status_ |= READ_ERROR;
+  this->status_ &= (READ_ERROR+1);
 }
 
 void
@@ -113,7 +108,8 @@ JAWS_IO_Handler::transmit_file_complete (void)
 {
   JAWS_TRACE ("JAWS_IO_Handler::transmit_file_complete");
   // this->pipeline_->transmit_file_complete ();
-  this->status_ = TRANSMIT_OK;
+  this->status_ |= TRANSMIT_OK;
+  this->status_ &= (TRANSMIT_OK+1);
 }
 
 void
@@ -122,20 +118,23 @@ JAWS_IO_Handler::transmit_file_error (int result)
   JAWS_TRACE ("JAWS_IO_Handler::transmit_file_error");
   ACE_UNUSED_ARG (result);
   // this->pipeline_->transmit_file_complete (result);
-  this->status_ = TRANSMIT_ERROR;
+  this->status_ |= TRANSMIT_ERROR;
+  this->status_ &= (TRANSMIT_ERROR+1);
 }
 
 void
 JAWS_IO_Handler::receive_file_complete (void)
 {
-  this->status_ = RECEIVE_OK;
+  this->status_ |= RECEIVE_OK;
+  this->status_ &= (RECEIVE_OK+1);
 }
 
 void
 JAWS_IO_Handler::receive_file_error (int result)
 {
   ACE_UNUSED_ARG(result);
-  this->status_ = RECEIVE_ERROR;
+  this->status_ |= RECEIVE_ERROR;
+  this->status_ &= (RECEIVE_ERROR+1);
 }
 
 void
@@ -143,20 +142,23 @@ JAWS_IO_Handler::write_error (void)
 {
   ACE_DEBUG ((LM_DEBUG, " (%t) error in writing response\n"));
 
-  this->status_ = WRITE_ERROR;
+  this->status_ |= WRITE_ERROR;
+  this->status_ &= (WRITE_ERROR+1);
   this->done ();
 }
 
 void
 JAWS_IO_Handler::confirmation_message_complete (void)
 {
-  this->status_ = WRITE_OK;
+  this->status_ |= WRITE_OK;
+  this->status_ &= (WRITE_OK+1);
 }
 
 void
 JAWS_IO_Handler::error_message_complete (void)
 {
-  this->status_ = WRITE_OK;
+  this->status_ |= WRITE_OK;
+  this->status_ &= (WRITE_OK+1);
 }
 
 JAWS_IO_Handler_Factory *
@@ -210,38 +212,97 @@ JAWS_IO_Handler::status (void)
 void
 JAWS_IO_Handler::idle (void)
 {
-  this->status_ = IDLE;
+  this->status_ &= (IDLE+1);
 }
 
 void
 JAWS_IO_Handler::acquire (void)
 {
-  this->count_ = 1;
 }
 
 void
-JAWS_IO_Handler::release (void)
+JAWS_IO_Handler::lock (void)
 {
-  if (this->count_ == 0)
-    this->done ();
-  else
-    this->count_ = 0;
 }
 
-int
-JAWS_IO_Handler::count (void)
+void
+JAWS_IO_Handler::release (void)
 {
-  return this->count_;
 }
 
 #if defined (ACE_WIN32) || defined (ACE_HAS_AIO_CALLS)
 
+JAWS_Asynch_IO_Handler_Factory::~JAWS_Asynch_IO_Handler_Factory (void)
+{
+}
+ 
+JAWS_IO_Handler *
+JAWS_Asynch_IO_Handler_Factory::create_io_handler (void)
+{
+  JAWS_TRACE ("JAWS_Asynch_IO_Handler_Factory::create");
+
+  JAWS_Asynch_IO_Handler *handler;
+  handler = new JAWS_Asynch_IO_Handler (this);
+
+  return handler;
+}
+ 
+void
+JAWS_Asynch_IO_Handler_Factory::destroy_io_handler (JAWS_IO_Handler *handler)
+{
+  JAWS_TRACE ("JAWS_IO_Handler_Factory::destroy");
+
+  if (handler != 0)
+    {
+cerr << "(" << thr_self () << ") locking for destruction: " << handler << endl;
+      handler->lock ();
+      delete handler->message_block ();
+      handler->message_block (0);
+      delete handler;
+    }
+}
+
+
+JAWS_Asynch_IO_Handler::JAWS_Asynch_IO_Handler (JAWS_Asynch_IO_Handler_Factory *factory)
+  : JAWS_IO_Handler (factory),
+    handler_ (0)
+{
+  this->status_ = 1;
+}
+
+JAWS_Asynch_IO_Handler::~JAWS_Asynch_IO_Handler (void)
+{
+  delete this->handler_;
+  this->handler_ = 0;
+}
+
 ACE_Handler *
-JAWS_IO_Handler::handler (void)
+JAWS_Asynch_IO_Handler::handler (void)
 {
   return this->handler_;
 }
 
+void
+JAWS_Asynch_IO_Handler::acquire (void)
+{
+cerr << "(" << thr_self () << ") acquire handler: " << this << endl;
+  this->count_.acquire_read ();
+}
+
+void
+JAWS_Asynch_IO_Handler::lock (void)
+{
+cerr << "(" << thr_self () << ") locking handler: " << this << endl;
+  this->count_.acquire_write ();
+}
+
+void
+JAWS_Asynch_IO_Handler::release (void)
+{
+cerr << "(" << thr_self () << ") release handler: " << this << endl;
+  this->count_.release ();
+}
+
 JAWS_Asynch_Handler::JAWS_Asynch_Handler (void)
   : ioh_ (0)
 {
@@ -284,7 +345,7 @@ JAWS_Asynch_Handler::act (const void *act_ref)
   JAWS_TRACE ("JAWS_Asynch_Handler::act");
 
   // Set the ioh from the act
-  this->ioh_ = (JAWS_IO_Handler *) act_ref;
+  this->ioh_ = (JAWS_Asynch_IO_Handler *) act_ref;
 }
 
 #if 0
@@ -417,12 +478,12 @@ JAWS_Asynch_Handler::handle_accept (const ACE_Asynch_Accept::Result &result)
 }
 
 void
-JAWS_Asynch_Handler::handler (JAWS_IO_Handler *ioh)
+JAWS_Asynch_Handler::handler (JAWS_Asynch_IO_Handler *ioh)
 {
   this->ioh_ = ioh;
 }
 
-JAWS_IO_Handler *
+JAWS_Asynch_IO_Handler *
 JAWS_Asynch_Handler::handler (void)
 {
   return this->ioh_;
diff --git a/apps/JAWS/PROTOTYPE/JAWS/IO_Handler.h b/apps/JAWS/PROTOTYPE/JAWS/IO_Handler.h
index ed49873c58d..b474ac6d0bf 100644
--- a/apps/JAWS/PROTOTYPE/JAWS/IO_Handler.h
+++ b/apps/JAWS/PROTOTYPE/JAWS/IO_Handler.h
@@ -31,6 +31,7 @@ class JAWS_IO_Handler;
 class JAWS_IO_Handler_Factory;
 class JAWS_Data_Block;
 class JAWS_Pipeline_Handler;
+class JAWS_Waiter;
 
 class JAWS_Export JAWS_Abstract_IO_Handler
   // = TITLE
@@ -118,17 +119,26 @@ public:
   virtual void idle (void) = 0;
   // puts handler in an idle state
 
-  enum { IDLE = 0,
-         ACCEPT_OK, ACCEPT_ERROR,
-         READ_OK, READ_ERROR,
-         WRITE_OK, WRITE_ERROR,
-         TRANSMIT_OK, TRANSMIT_ERROR,
-         RECEIVE_OK, RECEIVE_ERROR };
+  enum { IDLE = 0, IDLE_A = 1,
+         ACCEPT_OK = 2, ACCEPT_OK_A = 3,
+         ACCEPT_ERROR = 4, ACCEPT_ERROR_A = 5,
+         READ_OK = 6, READ_OK_A = 7,
+         READ_ERROR = 8, READ_ERROR_A = 9,
+         WRITE_OK = 10, WRITE_OK_A = 11,
+         WRITE_ERROR = 12, WRITE_ERROR_A = 13,
+         TRANSMIT_OK = 14, TRANSMIT_OK_A = 15,
+         TRANSMIT_ERROR = 16, TRANSMIT_ERROR_A = 17,
+         RECEIVE_OK = 18, RECEIVE_OK_A = 19,
+         RECEIVE_ERROR = 20, RECEIVE_ERROR_A = 21 };
   // The different states of the handler
 
 };
 
 #if defined(ACE_WIN32) || defined(ACE_HAS_AIO_CALLS)
+
+// Forward reference.
+class JAWS_Asynch_IO_Handler;
+
 class JAWS_Export JAWS_Asynch_Handler : public ACE_Service_Handler
 {
 public:
@@ -153,8 +163,8 @@ public:
   virtual void handle_accept (const ACE_Asynch_Accept::Result &result);
   // This method will be called when an asynchronous accept completes.
 
-  virtual void handler (JAWS_IO_Handler *ioh);
-  virtual JAWS_IO_Handler * handler (void);
+  virtual void handler (JAWS_Asynch_IO_Handler *ioh);
+  virtual JAWS_Asynch_IO_Handler * handler (void);
 
   virtual void dispatch_handler (void);
 
@@ -167,17 +177,13 @@ public:
   //virtual ACE_HANDLE handle (void) const;
 
 private:
-  JAWS_IO_Handler *ioh_;
+  JAWS_Asynch_IO_Handler *ioh_;
 };
 #endif /* defined(ACE_WIN32) || defined(ACE_HAS_AIO_CALLS) */
 
 
 class JAWS_Export JAWS_IO_Handler : public JAWS_Abstract_IO_Handler
 {
-#if defined(ACE_WIN32) || defined(ACE_HAS_AIO_CALLS)
-friend class JAWS_Asynch_Handler;
-#endif /* defined(ACE_WIN32) || defined(ACE_HAS_AIO_CALLS) */
-  // Provide implementations for the common functions.
 public:
   JAWS_IO_Handler (JAWS_IO_Handler_Factory *factory);
   virtual ~JAWS_IO_Handler (void);
@@ -204,8 +210,8 @@ public:
   virtual void idle (void);
 
   virtual void acquire (void);
+  virtual void lock (void);
   virtual void release (void);
-  virtual int count (void);
 
   virtual void task (JAWS_Pipeline_Handler *ph);
   virtual JAWS_Pipeline_Handler *task (void);
@@ -213,10 +219,6 @@ public:
   virtual void message_block (JAWS_Data_Block *mb);
   virtual JAWS_Data_Block *message_block (void);
 
-#if defined (ACE_WIN32) || defined (ACE_HAS_AIO_CALLS)
-  virtual ACE_Handler *handler (void);
-#endif /* ACE_WIN32 || ACE_HAS_AIO_CALLS */
-
 protected:
   int status_;
   // The state of the handler.
@@ -233,18 +235,9 @@ protected:
 
   JAWS_IO_Handler_Factory *factory_;
   // The reference to the handler's factory.
-
-  int count_;
-
-#if defined (ACE_WIN32) || defined (ACE_HAS_AIO_CALLS)
-  JAWS_Asynch_Handler *handler_;
-#endif /* ACE_WIN32 || ACE_HAS_AIO_CALLS */
 };
 
 class JAWS_Export JAWS_IO_Handler_Factory
-#if defined(ACE_WIN32) || defined(ACE_HAS_AIO_CALLS)
-  : public ACE_Service_Handler
-#endif /* defined(ACE_WIN32) || defined(ACE_HAS_AIO_CALLS) */
 {
 public:
   virtual ~JAWS_IO_Handler_Factory (void);
@@ -263,9 +256,52 @@ typedef JAWS_IO_Handler_Factory JAWS_Synch_IO_Handler_Factory;
 typedef ACE_Singleton<JAWS_Synch_IO_Handler_Factory, ACE_SYNCH_MUTEX>
         JAWS_Synch_IO_Handler_Factory_Singleton;
 
+#if defined(ACE_WIN32) || defined(ACE_HAS_AIO_CALLS)
+
+class JAWS_Export JAWS_Asynch_IO_Handler : public JAWS_IO_Handler
+{
+friend class JAWS_Asynch_Handler;
+friend class JAWS_Asynch_IO_Handler_Factory;
+friend class JAWS_Waiter;
+
+  // Provide implementations for the common functions.
+public:
+  JAWS_Asynch_IO_Handler (JAWS_Asynch_IO_Handler_Factory *factory);
+  virtual ~JAWS_Asynch_IO_Handler (void);
+
+  virtual ACE_Handler *handler (void);
+
+  virtual void acquire (void);
+  virtual void lock (void);
+  virtual void release (void);
+
+protected:
+
+  JAWS_Asynch_Handler *handler_;
+  ACE_SYNCH_RW_MUTEX count_;
+};
+
+
+class JAWS_Export JAWS_Asynch_IO_Handler_Factory : public JAWS_IO_Handler_Factory
+{
+public:
+  virtual ~JAWS_Asynch_IO_Handler_Factory (void);
+  // Destructor
+
+  virtual JAWS_IO_Handler *create_io_handler (void);
+  // This creates a new JAWS_IO_Handler
+
+  virtual void destroy_io_handler (JAWS_IO_Handler *handler);
+  // This deletes a JAWS_IO_Handler
+};
+
+#else
+
 typedef JAWS_IO_Handler JAWS_Asynch_IO_Handler;
 typedef JAWS_IO_Handler_Factory JAWS_Asynch_IO_Handler_Factory;
 
+#endif /* defined(ACE_WIN32) || defined(ACE_HAS_AIO_CALLS) */
+
 typedef ACE_Singleton<JAWS_Asynch_IO_Handler_Factory, ACE_SYNCH_MUTEX>
         JAWS_Asynch_IO_Handler_Factory_Singleton;
 
diff --git a/apps/JAWS/PROTOTYPE/JAWS/Pipeline_Tasks.cpp b/apps/JAWS/PROTOTYPE/JAWS/Pipeline_Tasks.cpp
index e36e46909b3..7197b945786 100644
--- a/apps/JAWS/PROTOTYPE/JAWS/Pipeline_Tasks.cpp
+++ b/apps/JAWS/PROTOTYPE/JAWS/Pipeline_Tasks.cpp
@@ -24,6 +24,7 @@ JAWS_Pipeline_Handler::put (ACE_Message_Block *mb, ACE_Time_Value *tv)
 {
   JAWS_Data_Block *db = ACE_dynamic_cast (JAWS_Data_Block *, mb);
 
+  // guarantee the handler remains for the duration of this call
   db->io_handler ()->acquire ();
 
   int status = this->handle_put (db, tv);
@@ -70,10 +71,16 @@ JAWS_Pipeline_Accept_Task::put (ACE_Message_Block *mb, ACE_Time_Value *tv)
       return -1;
     }
 
+  ioh->acquire ();
+
   ioh->task (next);
   db->io_handler (ioh);
 
-  return this->handle_put (ioh->message_block (), tv);
+  int result = this->handle_put (ioh->message_block (), tv);
+
+  ioh->release ();
+
+  return result;
 }
 
 int
@@ -164,6 +171,34 @@ JAWS_Pipeline_Accept_Task::new_handler (JAWS_Data_Block *data)
   return nioh;
 }
 
+int
+JAWS_Pipeline_Done_Task::put (ACE_Message_Block *mb, ACE_Time_Value *)
+{
+  JAWS_TRACE ("JAWS_HTTP_10_Write_Task::handle_put");
+
+  JAWS_Data_Block *data = ACE_dynamic_cast (JAWS_Data_Block *, mb);
+
+  JAWS_IO_Handler *handler = data->io_handler ();
+  JAWS_Dispatch_Policy *policy = this->policy ();
+  if (policy == 0) policy = data->policy ();
+
+  JAWS_IO *io = policy->io ();
+
+  data->task (0);
+  data->io_handler (0);
+
+  if (handler)
+    handler->done ();
+
+  // hack, let Concurrency know we are done.
+  return -2;
+}
+
+int
+JAWS_Pipeline_Done_Task::handle_put (JAWS_Data_Block *, ACE_Time_Value *)
+{
+  return 0;
+}
 
 #if defined (ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION)
 template class JAWS_Pipeline_Abstract_Handler<JAWS_Data_Block>;
diff --git a/apps/JAWS/PROTOTYPE/JAWS/Pipeline_Tasks.h b/apps/JAWS/PROTOTYPE/JAWS/Pipeline_Tasks.h
index 524eadcf6a4..0877a40aaf2 100644
--- a/apps/JAWS/PROTOTYPE/JAWS/Pipeline_Tasks.h
+++ b/apps/JAWS/PROTOTYPE/JAWS/Pipeline_Tasks.h
@@ -35,7 +35,17 @@ public:
   virtual JAWS_IO_Handler * new_handler (JAWS_Data_Block *data);
 };
 
+class JAWS_Pipeline_Done_Task : public JAWS_Pipeline_Handler
+{
+public:
+  virtual int put (ACE_Message_Block *mb, ACE_Time_Value *tv = 0);
+  virtual int handle_put (JAWS_Data_Block *data, ACE_Time_Value *tv);
+};
+
 typedef ACE_Singleton<JAWS_Pipeline_Accept_Task, ACE_SYNCH_MUTEX>
         JAWS_Pipeline_Accept_Task_Singleton;
 
+typedef ACE_Singleton<JAWS_Pipeline_Done_Task, ACE_SYNCH_NULL_MUTEX>
+        JAWS_Pipeline_Done_Task_Singleton;
+
 #endif /* !defined (JAWS_PIPELINE_TASKS_H) */
diff --git a/apps/JAWS/PROTOTYPE/JAWS/Waiter.cpp b/apps/JAWS/PROTOTYPE/JAWS/Waiter.cpp
index 61b053a2769..c20a0625225 100644
--- a/apps/JAWS/PROTOTYPE/JAWS/Waiter.cpp
+++ b/apps/JAWS/PROTOTYPE/JAWS/Waiter.cpp
@@ -3,6 +3,7 @@
 #include "ace/Proactor.h"
 
 #include "JAWS/Waiter.h"
+#include "JAWS/IO_Handler.h"
 
 ACE_RCSID(JAWS, Waiter, "$Id$")