diff options
Diffstat (limited to 'apps/JAWS/PROTOTYPE/JAWS/Concurrency.cpp')
-rw-r--r-- | apps/JAWS/PROTOTYPE/JAWS/Concurrency.cpp | 263 |
1 files changed, 0 insertions, 263 deletions
diff --git a/apps/JAWS/PROTOTYPE/JAWS/Concurrency.cpp b/apps/JAWS/PROTOTYPE/JAWS/Concurrency.cpp deleted file mode 100644 index 03cc659c70d..00000000000 --- a/apps/JAWS/PROTOTYPE/JAWS/Concurrency.cpp +++ /dev/null @@ -1,263 +0,0 @@ -// $Id$ - -#include "JAWS/JAWS.h" - -#include "JAWS/Concurrency.h" -#include "JAWS/IO_Handler.h" -#include "JAWS/Pipeline.h" -#include "JAWS/Pipeline_Tasks.h" -#include "JAWS/Policy.h" -#include "JAWS/Data_Block.h" - -JAWS_Concurrency_Base::JAWS_Concurrency_Base (void) - : mb_acquired_ (0), - mb_ (0) -{ -} - -JAWS_Concurrency_Base::~JAWS_Concurrency_Base (void) -{ -} - -ACE_Message_Block * -JAWS_Concurrency_Base::singleton_mb (void) -{ - if (this->mb_acquired_ == 0) - { - ACE_Guard<ACE_Thread_Mutex> g(this->lock_); - - if (this->mb_acquired_ == 0) - { - int result; - ACE_Message_Block *mb; - - result = this->getq (mb); - this->mb_acquired_ = 1; - - if (result == -1 || mb == 0) - return 0; - - this->mb_ = mb; - } - } - - return this->mb_; -} - -int -JAWS_Concurrency_Base::put (ACE_Message_Block *mb, ACE_Time_Value *tv) -{ - return this->putq (mb, tv); -} - -int -JAWS_Concurrency_Base::svc (void) -{ - JAWS_TRACE ("JAWS_Concurrency_Base::svc"); - - ACE_Message_Block *mb; // The message queue element - JAWS_Data_Block *db; // Contains the task list - - mb = this->singleton_mb (); - - // A NULL data block indicates that the thread should shut - // itself down - if (mb == 0) - { - JAWS_TRACE ("JAWS_Concurrency_Base::svc, empty message block"); - return -1; - } - - db = ACE_dynamic_cast (JAWS_Data_Block *, mb); - - this->svc_loop (db); - - return 0; -} - -int -JAWS_Concurrency_Base::svc_loop (JAWS_Data_Block *db) -{ - JAWS_TRACE ("JAWS_Concurrency_Base::svc_loop"); - - for (;;) - this->svc_hook (db); - - return 0; -} - -int -JAWS_Concurrency_Base::svc_hook (JAWS_Data_Block *db) -{ - JAWS_TRACE ("JAWS_Concurrency_Base::svc_hook"); - - int result = 0; - - JAWS_Dispatch_Policy *policy; // Contains task policies - JAWS_IO_Handler *handler; // Keeps the state of the task - JAWS_Pipeline_Handler *task; // The task itself - - // Thread specific message block and data block - JAWS_Data_Block *ts_db = new JAWS_Data_Block; - if (ts_db == 0) - { - ACE_ERROR ((LM_ERROR, "%p\n", "JAWS_Concurrency_Base::svc_hook")); - return -1; - } - - ts_db->task (db->task ()); - ts_db->policy (db->policy ()); - - // ACE_DEBUG ((LM_DEBUG, "yo")); - - policy = db->policy (); - - // Each time we iterate, we create a handler to maintain - // our state for us. - handler = policy->ioh_factory ()->create_io_handler (); - if (handler == 0) - { - ACE_DEBUG ((LM_DEBUG, "JAWS_Server::open, can't create handler\n")); - ts_db->release (); - return -1; - } - - // Set the initial task in the handler - handler->task (db->task ()); - handler->message_block (ts_db); - ts_db->io_handler (handler); - - do - { - JAWS_TRACE ("JAWS_Concurrency_Base::svc_hook, looping"); - - // handler maintains the state of the protocol - task = handler->task (); - ts_db = handler->message_block (); - - // Use a NULL task to make the thread recycle now - if (task == 0) - break; - - // the task should set the handler to the appropriate next step - result = task->put (ts_db); - - if (result == 1) - { - JAWS_TRACE ("JAWS_Concurrency_Base::svc_hook, waiting"); - // need to wait for an asynchronous event - // I don't know how to do this yet, so pretend it's - // an error - - // handler = wait for completion (); - // db->io_handler (handler); - // result = 0; - result = -1; - } - - if (result == -1) - { - // definately something wrong. - JAWS_TRACE ("JAWS_Concurrency_Base::svc_hook, negative result"); - ACE_ERROR ((LM_ERROR, "%p\n", "JAWS_Concurrency_Base::svc_hook")); - break; - } - - } - while (result == 0); - - policy->ioh_factory ()->destroy_io_handler (handler); - ts_db->release (); - - return result; -} - -int -JAWS_Concurrency_Base::activate_hook (void) -{ - return 0; -} - -JAWS_Dispatcher::JAWS_Dispatcher (void) - : policy_(0) -{ -} - -int -JAWS_Dispatcher::dispatch (ACE_Message_Block *mb) -{ - return this->policy ()->concurrency ()->put (mb); -} - -JAWS_Dispatch_Policy * -JAWS_Dispatcher::policy (void) -{ - return this->policy_; -} - -JAWS_Dispatch_Policy * -JAWS_Dispatcher::policy (JAWS_Dispatch_Policy *p) -{ - this->policy_ = p; - return this->policy_; -} - -int -JAWS_Thread_Pool_Task::open (long flags, int nthreads, int maxthreads) -{ - this->flags_ = flags; - this->nthreads_ = nthreads; - this->maxthreads_ = maxthreads; - - if (this->activate (flags, nthreads) == -1) - ACE_ERROR_RETURN ((LM_ERROR, "%p\n", "JAWS_Thread_Pool_Task::activate"), - -1); - - return 0; -} - -int -JAWS_Thread_Per_Task::open (long flags, int maxthreads) -{ - this->flags_ = flags; - this->maxthreads_ = maxthreads; - return 0; -} - -int -JAWS_Thread_Per_Task::put (ACE_Message_Block *mb, ACE_Time_Value *tv) -{ - JAWS_TRACE ("JAWS_Thread_Per_Task::put"); - - this->putq (mb, tv); - return this->activate_hook (); -} - -int -JAWS_Thread_Per_Task::svc_loop (JAWS_Data_Block *db) -{ - return this->svc_hook (db); -} - -int -JAWS_Thread_Per_Task::activate_hook (void) -{ - const int force_active = 1; - const int nthreads = 1; - - if (this->activate (this->flags_, nthreads, force_active) == -1) - ACE_ERROR_RETURN ((LM_ERROR, "%p\n", "JAWS_Thread_Pool_Task::activate"), - -1); - - return 0; -} - -#if defined (ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION) -template class ACE_Singleton<JAWS_Dispatcher, ACE_SYNCH_MUTEX>; -template class ACE_Singleton<JAWS_Thread_Pool_Task, ACE_SYNCH_MUTEX>; -template class ACE_Singleton<JAWS_Thread_Per_Task, ACE_SYNCH_MUTEX>; -#elif defined (ACE_HAS_TEMPLATE_INSTANTIATION_PRAGMA) -#pragma instantiate ACE_Singleton<JAWS_Dispatcher, ACE_SYNCH_MUTEX> -#pragma instantiate ACE_Singleton<JAWS_Thread_Pool_Task, ACE_SYNCH_MUTEX> -#pragma instantiate ACE_Singleton<JAWS_Thread_Per_Task, ACE_SYNCH_MUTEX> -#endif /* ACE_HAS_EXPLICIT_TEMPLATE_INSTANTIATION */ |