diff options
| -rw-r--r-- | ace/POSIX_Asynch_IO.cpp | 376 | ||||
| -rw-r--r-- | ace/POSIX_Asynch_IO.h | 8 | ||||
| -rw-r--r-- | ace/POSIX_Proactor.cpp | 302 | ||||
| -rw-r--r-- | ace/POSIX_Proactor.h | 19 | ||||
| -rw-r--r-- | ace/Proactor.cpp | 1 | ||||
| -rw-r--r-- | ace/Proactor.h | 6 | ||||
| -rw-r--r-- | ace/Proactor_Impl.h | 12 | ||||
| -rw-r--r-- | ace/WIN32_Proactor.cpp | 8 |
8 files changed, 359 insertions, 373 deletions
diff --git a/ace/POSIX_Asynch_IO.cpp b/ace/POSIX_Asynch_IO.cpp index e66a15aec80..fd4e6f7e9fc 100644 --- a/ace/POSIX_Asynch_IO.cpp +++ b/ace/POSIX_Asynch_IO.cpp @@ -74,11 +74,11 @@ ACE_POSIX_Asynch_Result::priority (void) const } int -ACE_POSIX_Asynch_Result::post_completion (ACE_Proactor_Impl *proactor) +ACE_POSIX_Asynch_Result::post_completion (ACE_Proactor_Impl *proactor_impl) { // Get to the platform specific implementation. ACE_POSIX_Proactor *posix_proactor = ACE_dynamic_cast (ACE_POSIX_Proactor *, - proactor); + proactor_impl); if (posix_proactor == 0) ACE_ERROR_RETURN ((LM_ERROR, "Dynamic cast to POSIX Proactor failed\n"), -1); @@ -184,6 +184,12 @@ ACE_POSIX_Asynch_Operation::ACE_POSIX_Asynch_Operation (void) // ********************************************************************* +ACE_POSIX_AIOCB_Proactor * +ACE_POSIX_AIOCB_Asynch_Operation::posix_proactor (void) const +{ + return this->posix_aiocb_proactor_; +} + ACE_POSIX_AIOCB_Asynch_Operation::ACE_POSIX_AIOCB_Asynch_Operation (ACE_POSIX_AIOCB_Proactor *posix_aiocb_proactor) : ACE_Asynch_Operation_Impl (), ACE_POSIX_Asynch_Operation (), @@ -201,11 +207,17 @@ ACE_POSIX_AIOCB_Asynch_Operation::~ACE_POSIX_AIOCB_Asynch_Operation (void) int ACE_POSIX_AIOCB_Asynch_Operation::register_aio_with_proactor (aiocb *aiocb_ptr) { - return this->posix_aiocb_proactor_->register_aio_with_proactor (aiocb_ptr); + return this->posix_proactor ()->register_aio_with_proactor (aiocb_ptr); } // ********************************************************************* +ACE_POSIX_SIG_Proactor * +ACE_POSIX_SIG_Asynch_Operation::posix_proactor (void) const +{ + return this->posix_sig_proactor_; +} + ACE_POSIX_SIG_Asynch_Operation::ACE_POSIX_SIG_Asynch_Operation (ACE_POSIX_SIG_Proactor *posix_sig_proactor) : ACE_Asynch_Operation_Impl (), ACE_POSIX_Asynch_Operation (), @@ -307,7 +319,7 @@ ACE_POSIX_AIOCB_Asynch_Read_Stream::read (ACE_Message_Block &message_block, message_block, bytes_to_read, act, - this->posix_aiocb_proactor_->get_handle (), + this->posix_proactor ()->get_handle (), priority), -1); @@ -518,7 +530,7 @@ ACE_POSIX_AIOCB_Asynch_Write_Stream::write (ACE_Message_Block &message_block, message_block, bytes_to_write, act, - this->posix_aiocb_proactor_->get_handle (), + this->posix_proactor ()->get_handle (), priority), -1); @@ -714,7 +726,7 @@ ACE_POSIX_AIOCB_Asynch_Read_File::read (ACE_Message_Block &message_block, act, offset, offset_high, - this->posix_aiocb_proactor_->get_handle (), + this->posix_proactor ()->get_handle (), priority), -1); @@ -874,7 +886,7 @@ ACE_POSIX_AIOCB_Asynch_Write_File::write (ACE_Message_Block &message_block, act, offset, offset_high, - this->posix_aiocb_proactor_->get_handle (), + this->posix_proactor ()->get_handle (), priority), -1); @@ -1023,42 +1035,43 @@ ACE_POSIX_Asynch_Accept_Result::~ACE_POSIX_Asynch_Accept_Result (void) class ACE_Export ACE_POSIX_Asynch_Accept_Handler : public ACE_Event_Handler { // = TITLE - // For the POSIX implementation, this class takes care of doing - // Asynch_Accept. + // For the POSIX implementation, we have two helper classes + // (ACE_POSIX_AIOCB_Asynch_Accept_Hander and + // ACE_POSIX_SIG_Asynch_Accept_Handler) to do <Asynch_Accept>. This + // class abstracts out the commonalities on these two helper classes. // // = DESCRIPTION // public: - ACE_POSIX_Asynch_Accept_Handler (ACE_Reactor* reactor); - // Constructor. Give the reactor so that it can activate/deactivate - // the handlers. Give also the proactor used here, so that the - // handler can send information through the notification pipe of the - // proactor, in case AIO_CONTROL_BLOCKS strategy is used. - ~ACE_POSIX_Asynch_Accept_Handler (void); // Destructor. int register_accept_call (ACE_POSIX_Asynch_Accept_Result* result); // Register this <accept> call with the local handler. - // virtual int handle_input (ACE_HANDLE fd = ACE_INVALID_HANDLE) = 0; - // Called when accept event comes up on the <listen_handle>. - // This is defined in ACE_Handler, the derived - // Asynch_Accept_Handler's will define this again. - protected: + ACE_POSIX_Asynch_Accept_Handler (ACE_Reactor* reactor, + ACE_POSIX_Proactor *posix_proactor); + // Constructor. Give the reactor so that it can activate/deactivate + // the handlers. Give also the proactor used here, so that the + // handler can send the <POSIX_Asynch_Accept> result block through + // <post_completion>. + ACE_POSIX_Asynch_Accept_Result* deregister_accept_call (void); // Undo the things done when registering. - ACE_Unbounded_Queue<ACE_POSIX_Asynch_Accept_Result*> result_queue_; - // Queue of Result pointers that correspond to all the <accept>'s - // pending. - ACE_Reactor* reactor_; // Reactor used by the Asynch_Accept. We need this here to enable // and disable the <handle> now and then, depending on whether any // <accept> is pending or no. + ACE_POSIX_Proactor *posix_proactor_; + // POSIX_Proactor. + + ACE_Unbounded_Queue<ACE_POSIX_Asynch_Accept_Result*> result_queue_; + // Queue of Result pointers that correspond to all the <accept>'s + // pending. + ACE_Thread_Mutex lock_; // The lock to protect the result queue which is shared. The queue // is updated by main thread in the register function call and @@ -1066,8 +1079,61 @@ protected: // mutex it. }; -ACE_POSIX_Asynch_Accept_Handler::ACE_POSIX_Asynch_Accept_Handler (ACE_Reactor* reactor) - : reactor_ (reactor) +// ********************************************************************* + +class ACE_Export ACE_POSIX_AIOCB_Asynch_Accept_Handler : public ACE_POSIX_Asynch_Accept_Handler +{ + // = TITLE + // For the POSIX implementation, this class takes care of doing + // <Asynch_Accept> for AIOCB strategy. + // + // = DESCRIPTION + // +public: + ACE_POSIX_AIOCB_Asynch_Accept_Handler (ACE_Reactor *reactor, + ACE_POSIX_AIOCB_Proactor *posix_aiocb_proactor); + // Constructor. Give the reactor so that it can activate/deactivate + // the handlers. Give also the proactor used here, so that the + // handler can send information through the notification pipe + // (<post_completion>). + + ~ACE_POSIX_AIOCB_Asynch_Accept_Handler (void); + // Destructor. + + virtual int handle_input (ACE_HANDLE fd = ACE_INVALID_HANDLE); + // Called when accept event comes up on the <listen_handle>. +}; + +// ********************************************************************* + +class ACE_Export ACE_POSIX_SIG_Asynch_Accept_Handler : public ACE_POSIX_Asynch_Accept_Handler +{ + // = TITLE + // For the POSIX implementation, this class takes care of doing + // Asynch_Accept. + // + // = DESCRIPTION + // +public: + ACE_POSIX_SIG_Asynch_Accept_Handler (ACE_Reactor* reactor, + ACE_POSIX_SIG_Proactor *posix_sig_proactor); + // Constructor. Give the reactor so that it can activate/deactivate + // the handlers. Give also the proactor used here, so that the + // handler can send information through <post_completion>. + + ~ACE_POSIX_SIG_Asynch_Accept_Handler (void); + // Destructor. + + virtual int handle_input (ACE_HANDLE fd = ACE_INVALID_HANDLE); + // Called when accept event comes up on the <listen_handle>. +}; + +// ********************************************************************* + +ACE_POSIX_Asynch_Accept_Handler::ACE_POSIX_Asynch_Accept_Handler (ACE_Reactor* reactor, + ACE_POSIX_Proactor *posix_proactor) + : reactor_ (reactor), + posix_proactor_ (posix_proactor) { } @@ -1078,8 +1144,6 @@ ACE_POSIX_Asynch_Accept_Handler::~ACE_POSIX_Asynch_Accept_Handler (void) int ACE_POSIX_Asynch_Accept_Handler::register_accept_call (ACE_POSIX_Asynch_Accept_Result* result) { - ACE_DEBUG ((LM_DEBUG, "ACE_Asynch_Accept_Handler::register_accept_call called\n")); - // The queue is updated by main thread in the register function call // and thru the auxillary thread in the deregister fun. So let us // mutex it. @@ -1089,7 +1153,7 @@ ACE_POSIX_Asynch_Accept_Handler::register_accept_call (ACE_POSIX_Asynch_Accept_R int insert_result = this->result_queue_.enqueue_tail (result); if (insert_result == -1) ACE_ERROR_RETURN ((LM_ERROR, - "%N:%l:ACE_Asynch_Accept_Handler::register_accept_call failed\n"), + "%N:%l:ACE_POSIX_Asynch_Accept_Handler::register_accept_call failed\n"), -1); // If this is the only item, then it means there the set was empty @@ -1097,21 +1161,19 @@ ACE_POSIX_Asynch_Accept_Handler::register_accept_call (ACE_POSIX_Asynch_Accept_R if (this->result_queue_.size () == 1) { int return_val = this->reactor_->resume_handler (result->listen_handle ()); - ACE_DEBUG ((LM_DEBUG, "%N:%l:return_val = %d\n", return_val)); if (return_val == -1) ACE_ERROR_RETURN ((LM_ERROR, - "%N:%l:Reactor::resume_handler failed\n"), + "%N:%l:ACE_POSIX_Asynch_Accept_Handler::register_accept_call: " + "Reactor::resume_handler failed\n"), -1); } - + return 0; } ACE_POSIX_Asynch_Accept_Result * ACE_POSIX_Asynch_Accept_Handler::deregister_accept_call (void) { - ACE_DEBUG ((LM_DEBUG, "ACE_Asynch_Accept_Handler::deregister_accept_call\n")); - // The queue is updated by main thread in the register function call and // thru the auxillary thread in the deregister fun. So let us mutex // it. @@ -1119,18 +1181,24 @@ ACE_POSIX_Asynch_Accept_Handler::deregister_accept_call (void) // Get the first item (result ptr) from the Queue. ACE_POSIX_Asynch_Accept_Result* result = 0; - int return_dequeue = this->result_queue_.dequeue_head (result); - if (return_dequeue == -1) - return 0; + if (this->result_queue_.dequeue_head (result) != 0) + ACE_ERROR_RETURN ((LM_ERROR, + "%N:%l:(%P | %t):%p\n", + "ACE_POSIX_Asynch_Accept_Handler::" + "deregister_accept_call:dequeueing failed"), + 0); ACE_ASSERT (result != 0); // Disable the <handle> in the reactor if no <accept>'s are pending. if (this->result_queue_.size () == 0) { - int return_val = this->reactor_->suspend_handler (result->listen_handle ()); - if (return_val != 0) - return 0; + if (this->reactor_->suspend_handler (result->listen_handle ()) != 0) + ACE_ERROR_RETURN ((LM_ERROR, + "%N:%l:(%P | %t):%p\n", + "ACE_POSIX_Asynch_Accept_Handler::" + "deregister_accept_call:suspend handler failed"), + 0); } // Return the result pointer. @@ -1139,37 +1207,9 @@ ACE_POSIX_Asynch_Accept_Handler::deregister_accept_call (void) // ********************************************************************* -class ACE_Export ACE_POSIX_AIOCB_Asynch_Accept_Handler : public ACE_POSIX_Asynch_Accept_Handler -{ - // = TITLE - // For the POSIX implementation, this class takes care of doing - // Asynch_Accept. - // - // = DESCRIPTION - // -public: - ACE_POSIX_AIOCB_Asynch_Accept_Handler (ACE_Reactor *reactor, - ACE_POSIX_AIOCB_Proactor *posix_aiocb_proactor); - // Constructor. Give the reactor so that it can activate/deactivate - // the handlers. Give also the proactor used here, so that the - // handler can send information through the notification pipe of the - // proactor, in case AIO_CONTROL_BLOCKS strategy is used. - - ~ACE_POSIX_AIOCB_Asynch_Accept_Handler (void); - // Destructor. - - virtual int handle_input (ACE_HANDLE fd = ACE_INVALID_HANDLE); - // Called when accept event comes up on the <listen_handle>. - -private: - ACE_POSIX_AIOCB_Proactor* posix_aiocb_proactor_; - // Proactor used by the Asynch_Accept class. -}; - ACE_POSIX_AIOCB_Asynch_Accept_Handler::ACE_POSIX_AIOCB_Asynch_Accept_Handler (ACE_Reactor* reactor, ACE_POSIX_AIOCB_Proactor* posix_aiocb_proactor) - : ACE_POSIX_Asynch_Accept_Handler (reactor), - posix_aiocb_proactor_ (posix_aiocb_proactor) + : ACE_POSIX_Asynch_Accept_Handler (reactor, posix_aiocb_proactor) { } @@ -1184,17 +1224,14 @@ ACE_POSIX_AIOCB_Asynch_Accept_Handler::handle_input (ACE_HANDLE fd) // able to just go ahead and do the <accept> now on this <fd>. This // should be the same as the <listen_handle>. - // @@ Debugging. - ACE_DEBUG ((LM_DEBUG, "ACE_POSIX_AIOCB_Asynch_Accept_Handler::handle_input called\n")); - // Deregister this info pertaining to this <accept> call. ACE_POSIX_Asynch_Accept_Result* result = this->deregister_accept_call (); - - // @@ Debugging. - ACE_DEBUG ((LM_DEBUG, - "(%P:%t):ACE_Asynch_Accept_Handler::handle_input : fd = [%d], Result->listen_handle = [%d]\n", - fd, - result->listen_handle ())); + if (result == 0) + ACE_ERROR_RETURN ((LM_ERROR, + "%N:%l:(%P | %t):%p\n", + "ACE_POSIX_AIOCB_Asynch_Accept_Handler::" + "handle_input:deregister_accept_call failed"), + -1); // Issue <accept> now. // @@ We shouldnt block here since we have already done poll/select @@ -1202,8 +1239,9 @@ ACE_POSIX_AIOCB_Asynch_Accept_Handler::handle_input (ACE_HANDLE fd) ACE_HANDLE new_handle = ACE_OS::accept (result->listen_handle (), 0, 0); if (new_handle == ACE_INVALID_HANDLE) ACE_ERROR_RETURN ((LM_ERROR, - "Error:(%P | %t):%p\n", - "<accept> system call failed"), + "%N:%l:(%P | %t):%p\n", + "ACE_POSIX_AIOCB_Asynch_Accept_Handler::" + "handle_input:<accept> system call failed"), -1); // Accept has completed. @@ -1212,16 +1250,69 @@ ACE_POSIX_AIOCB_Asynch_Accept_Handler::handle_input (ACE_HANDLE fd) result->aio_fildes = new_handle; // Notify the main process about this completion - // Send the result pointer thru the notify pipe depending on what is Completion - // Notification Strategy. - // @@ Debugging. - ACE_DEBUG ((LM_DEBUG, - "ACE_Asynch_Accept_Handler::handle_input: AIO_CONTROL_BLOCKS\n")); - // Send the Result through the notification pipe. - if (this->posix_aiocb_proactor_->notify_asynch_accept (result) == -1) - return -1; + if (this->posix_proactor_->post_completion (result) == -1) + ACE_ERROR_RETURN ((LM_ERROR, + "Error:(%P | %t):%p\n", + "ACE_POSIX_AIOCB_Asynch_Accept_Handler::" + "handle_input:<post_completion> failed"), + -1); + + return 0; +} + +// ********************************************************************* + +ACE_POSIX_SIG_Asynch_Accept_Handler::ACE_POSIX_SIG_Asynch_Accept_Handler (ACE_Reactor* reactor, + ACE_POSIX_SIG_Proactor *posix_sig_proactor) + : ACE_POSIX_Asynch_Accept_Handler (reactor, posix_sig_proactor) +{ +} +ACE_POSIX_SIG_Asynch_Accept_Handler::~ACE_POSIX_SIG_Asynch_Accept_Handler (void) +{ +} + +int +ACE_POSIX_SIG_Asynch_Accept_Handler::handle_input (ACE_HANDLE fd) +{ + // An <accept> has been sensed on the <listen_handle>. We should be + // able to just go ahead and do the <accept> now on this <fd>. This + // should be the same as the <listen_handle>. + + // Deregister this info pertaining to this <accept> call. + ACE_POSIX_Asynch_Accept_Result* result = this->deregister_accept_call (); + if (result == 0) + ACE_ERROR_RETURN ((LM_ERROR, + "%N:%l:(%P | %t):%p\n", + "ACE_POSIX_SIG_Asynch_Accept_Handler::" + "handle_input:deregister_accept_call failed"), + -1); + + // Issue <accept> now. + // @@ We shouldnt block here since we have already done poll/select + // thru reactor. But are we sure? + ACE_HANDLE new_handle = ACE_OS::accept (result->listen_handle (), 0, 0); + if (new_handle == ACE_INVALID_HANDLE) + ACE_ERROR_RETURN ((LM_ERROR, + "Error:(%P | %t):%p\n", + "ACE_POSIX_SIG_Asynch_Accept_Handler::" + "handle_input:<accept> system call failed"), + -1); + + // Accept has completed. + + // Store the new handle. + result->aio_fildes = new_handle; + + // Notify the main process about this completion. + if (this->posix_proactor_->post_completion (result) == -1) + ACE_ERROR_RETURN ((LM_ERROR, + "Error:(%P | %t):%p\n", + "ACE_POSIX_SIG_Asynch_Accept_Handler::" + "handle_input:<post_completion> failed"), + -1); + return 0; } @@ -1261,7 +1352,7 @@ ACE_POSIX_AIOCB_Asynch_Accept::accept (ACE_Message_Block &message_block, message_block, bytes_to_read, act, - this->posix_aiocb_proactor_->get_handle (), + this->posix_proactor ()->get_handle (), priority), -1); @@ -1288,7 +1379,7 @@ ACE_POSIX_AIOCB_Asynch_Accept::open (ACE_Handler &handler, // also with it. ACE_NEW_RETURN (this->accept_handler_, ACE_POSIX_AIOCB_Asynch_Accept_Handler (&this->reactor_, - this->posix_aiocb_proactor_), + this->posix_proactor ()), -1); // Register the handle with the reactor. @@ -1359,99 +1450,6 @@ ACE_POSIX_AIOCB_Asynch_Accept::thread_function (void* arg_reactor) // ********************************************************************* -class ACE_Export ACE_POSIX_SIG_Asynch_Accept_Handler : public ACE_POSIX_Asynch_Accept_Handler -{ - // = TITLE - // For the POSIX implementation, this class takes care of doing - // Asynch_Accept. - // - // = DESCRIPTION - // -public: - ACE_POSIX_SIG_Asynch_Accept_Handler (ACE_Reactor* reactor); - // Constructor. Give the reactor so that it can activate/deactivate - // the handlers. Give also the proactor used here, so that the - // handler can send information through the notification pipe of the - // proactor, in case AIO_CONTROL_BLOCKS strategy is used. - - ~ACE_POSIX_SIG_Asynch_Accept_Handler (void); - // Destructor. - - virtual int handle_input (ACE_HANDLE fd = ACE_INVALID_HANDLE); - // Called when accept event comes up on the <listen_handle>. -}; - -ACE_POSIX_SIG_Asynch_Accept_Handler::ACE_POSIX_SIG_Asynch_Accept_Handler (ACE_Reactor* reactor) - : ACE_POSIX_Asynch_Accept_Handler (reactor) -{ -} - -ACE_POSIX_SIG_Asynch_Accept_Handler::~ACE_POSIX_SIG_Asynch_Accept_Handler (void) -{ -} - -int -ACE_POSIX_SIG_Asynch_Accept_Handler::handle_input (ACE_HANDLE fd) -{ - // An <accept> has been sensed on the <listen_handle>. We should be - // able to just go ahead and do the <accept> now on this <fd>. This - // should be the same as the <listen_handle>. - - // @@ Debugging. - ACE_DEBUG ((LM_DEBUG, "ACE_POSIX_AIOCB_Asynch_Accept_Handler::handle_input called\n")); - - // Deregister this info pertaining to this <accept> call. - ACE_POSIX_Asynch_Accept_Result* result = this->deregister_accept_call (); - - // @@ Debugging. - ACE_DEBUG ((LM_DEBUG, - "(%t):ACE_Asynch_Accept_Handler::handle_input : fd = [%d], Result->listen_handle = [%d]\n", - fd, - result->listen_handle ())); - - // Issue <accept> now. - // @@ We shouldnt block here since we have already done poll/select - // thru reactor. But are we sure? - ACE_HANDLE new_handle = ACE_OS::accept (result->listen_handle (), 0, 0); - if (new_handle == ACE_INVALID_HANDLE) - ACE_ERROR_RETURN ((LM_ERROR, - "Error:(%P | %t):%p\n", - "<accept> system call failed"), - -1); - - // Accept has completed. - - // Store the new handle. - result->aio_fildes = new_handle; - - // Notify the mail process about this completion. - - // @@ Debugging. - ACE_DEBUG ((LM_DEBUG, - "ACE_Asynch_Accept_Handler::handle_input: RT_SIGNALS\n")); - - // Get this process id. - pid_t pid = ACE_OS::getpid (); - if (pid == (pid_t) -1) - ACE_ERROR_RETURN ((LM_ERROR, - "Error:(%P | %t):%p", - "<getpid> failed\n"), - -1); - - // Set the signal information. - sigval value; - value.sival_ptr = (void *) result; - - // Queue the signal. - if (sigqueue (pid, ACE_SIG_AIO, value) == -1) - ACE_ERROR_RETURN ((LM_ERROR, - "Error:(%P | %t):%p", - "<sigqueue> failed\n"), - -1); -} - -// ********************************************************************* - ACE_POSIX_SIG_Asynch_Accept::ACE_POSIX_SIG_Asynch_Accept (ACE_POSIX_SIG_Proactor *posix_sig_proactor) : ACE_Asynch_Operation_Impl (), ACE_Asynch_Accept_Impl (), @@ -1513,7 +1511,8 @@ ACE_POSIX_SIG_Asynch_Accept::open (ACE_Handler &handler, // Init the Asynch_Accept_Handler now. It needs to keep Proactor // also with it. ACE_NEW_RETURN (this->accept_handler_, - ACE_POSIX_SIG_Asynch_Accept_Handler (&this->reactor_), + ACE_POSIX_SIG_Asynch_Accept_Handler (&this->reactor_, + this->posix_proactor ()), -1); // Register the handle with the reactor. @@ -2338,8 +2337,8 @@ ACE_POSIX_AIOCB_Asynch_Transmit_File::transmit_file (ACE_HANDLE file, if (file_size == -1) ACE_ERROR_RETURN ((LM_ERROR, - "Error:%p\n", - ":Asynch_Transmit_File:Couldnt know the file size"), + "Error:%N:%l:%p\n", + "POSIX_AIOCB_Asynch_Transmit_File:filesize failed"), -1); if (bytes_to_write == 0) @@ -2371,7 +2370,7 @@ ACE_POSIX_AIOCB_Asynch_Transmit_File::transmit_file (ACE_HANDLE file, bytes_per_send, flags, act, - this->posix_aiocb_proactor_->get_handle (), + this->posix_proactor ()->get_handle (), priority), -1); @@ -2379,7 +2378,8 @@ ACE_POSIX_AIOCB_Asynch_Transmit_File::transmit_file (ACE_HANDLE file, ACE_POSIX_AIOCB_Asynch_Transmit_Handler *transmit_handler = 0; ACE_NEW_RETURN (transmit_handler, - ::ACE_POSIX_AIOCB_Asynch_Transmit_Handler (this->posix_aiocb_proactor_, result), + ::ACE_POSIX_AIOCB_Asynch_Transmit_Handler (this->posix_proactor (), + result), -1); ssize_t return_val = transmit_handler->transmit (); diff --git a/ace/POSIX_Asynch_IO.h b/ace/POSIX_Asynch_IO.h index 718bc1158fa..529f8d573ba 100644 --- a/ace/POSIX_Asynch_IO.h +++ b/ace/POSIX_Asynch_IO.h @@ -181,6 +181,10 @@ class ACE_Export ACE_POSIX_AIOCB_Asynch_Operation : public virtual ACE_POSIX_Asy // // = DESCRIPTION // +public: + ACE_POSIX_AIOCB_Proactor *posix_proactor (void) const; + // Return the underlying Proactor implementation. + protected: ACE_POSIX_AIOCB_Asynch_Operation (ACE_POSIX_AIOCB_Proactor *posix_aiocb_proactor); // Contructor. @@ -213,6 +217,10 @@ class ACE_Export ACE_POSIX_SIG_Asynch_Operation : public virtual ACE_POSIX_Async // // = DESCRIPTION // +public: + ACE_POSIX_SIG_Proactor *posix_proactor (void) const; + // Return the underlying Proactor implemetation. + protected: ACE_POSIX_SIG_Asynch_Operation (ACE_POSIX_SIG_Proactor *posix_sig_proactor); // Contructor. diff --git a/ace/POSIX_Proactor.cpp b/ace/POSIX_Proactor.cpp index 16417118e20..dae58e21e24 100644 --- a/ace/POSIX_Proactor.cpp +++ b/ace/POSIX_Proactor.cpp @@ -140,43 +140,69 @@ ACE_POSIX_Proactor::cancel_timer (ACE_Handler &handler, #endif /* 0 */ } -#if 0 int -ACE_POSIX_Proactor::handle_signal (int, siginfo_t *, ucontext_t *) +ACE_POSIX_Proactor::wake_up_dispatch_threads (void) { - // Perform a non-blocking "poll" for all the I/O events that have - // completed in the I/O completion queue. + return 0; +} - ACE_Time_Value timeout (0, 0); - int result = 0; +int +ACE_POSIX_Proactor::close_dispatch_threads (int) +{ + return 0; +} - while (1) - { - result = this->handle_events (timeout); - if (result != 0 || errno == ETIME) - break; - } +size_t +ACE_POSIX_Proactor::number_of_threads (void) const +{ + // @@ Implement it. + ACE_NOTSUP_RETURN (0); +} - // If our handle_events failed, we'll report a failure to the - // Reactor. - return result == -1 ? -1 : 0; +void +ACE_POSIX_Proactor::number_of_threads (size_t threads) +{ + // @@ Implement it. + ACE_UNUSED_ARG (threads); } -int -ACE_POSIX_Proactor::handle_close (ACE_HANDLE handle, - ACE_Reactor_Mask close_mask) +#if 0 +ACE_Proactor::Timer_Queue * +ACE_POSIX_Proactor::timer_queue (void) const { - ACE_UNUSED_ARG (close_mask); - ACE_UNUSED_ARG (handle); + return this->timer_queue_; +} - return this->close (); +void +ACE_POSIX_Proactor::timer_queue (Timer_Queue *tq) +{ + // cleanup old timer queue + if (this->delete_timer_queue_) + { + delete this->timer_queue_; + this->delete_timer_queue_ = 0; + } + + // new timer queue + if (tq == 0) + { + this->timer_queue_ = new Timer_Heap; + this->delete_timer_queue_ = 1; + } + else + { + this->timer_queue_ = tq; + this->delete_timer_queue_ = 0; + } + + // Set the proactor in the timer queue's functor + this->timer_queue_->upcall_functor ().proactor (*this); } #endif /* 0 */ ACE_HANDLE ACE_POSIX_Proactor::get_handle (void) const { - // @@ Implement this. return ACE_INVALID_HANDLE; } @@ -336,6 +362,39 @@ ACE_POSIX_Proactor::ACE_POSIX_Proactor (void) { } +#if 0 +int +ACE_POSIX_Proactor::handle_signal (int, siginfo_t *, ucontext_t *) +{ + // Perform a non-blocking "poll" for all the I/O events that have + // completed in the I/O completion queue. + + ACE_Time_Value timeout (0, 0); + int result = 0; + + while (1) + { + result = this->handle_events (timeout); + if (result != 0 || errno == ETIME) + break; + } + + // If our handle_events failed, we'll report a failure to the + // Reactor. + return result == -1 ? -1 : 0; +} + +int +ACE_POSIX_Proactor::handle_close (ACE_HANDLE handle, + ACE_Reactor_Mask close_mask) +{ + ACE_UNUSED_ARG (close_mask); + ACE_UNUSED_ARG (handle); + + return this->close (); +} +#endif /* 0 */ + void ACE_POSIX_Proactor::application_specific_code (ACE_POSIX_Asynch_Result *asynch_result, u_long bytes_transferred, @@ -358,84 +417,28 @@ ACE_POSIX_Proactor::application_specific_code (ACE_POSIX_Asynch_Result *asynch_r } } - - -int -ACE_POSIX_Proactor::wake_up_dispatch_threads (void) -{ - return 0; -} - -int -ACE_POSIX_Proactor::close_dispatch_threads (int) -{ - return 0; -} - -size_t -ACE_POSIX_Proactor::number_of_threads (void) const -{ - // @@ Implement it. - return 0; -} - -void -ACE_POSIX_Proactor::number_of_threads (size_t threads) -{ - // @@ Implement it. - ACE_UNUSED_ARG (threads); -} - -#if 0 -ACE_Proactor::Timer_Queue * -ACE_POSIX_Proactor::timer_queue (void) const -{ - return this->timer_queue_; -} - -void -ACE_POSIX_Proactor::timer_queue (Timer_Queue *tq) -{ - // cleanup old timer queue - if (this->delete_timer_queue_) - { - delete this->timer_queue_; - this->delete_timer_queue_ = 0; - } - - // new timer queue - if (tq == 0) - { - this->timer_queue_ = new Timer_Heap; - this->delete_timer_queue_ = 1; - } - else - { - this->timer_queue_ = tq; - this->delete_timer_queue_ = 0; - } - - // Set the proactor in the timer queue's functor - this->timer_queue_->upcall_functor ().proactor (*this); -} -#endif /* 0 */ - // ********************************************************************* -class ACE_Export ACE_AIO_Accept_Handler : public ACE_Handler +class ACE_Export ACE_AIOCB_Notify_Pipe_Manager : public ACE_Handler { // = TITLE - // Helper class for doing Asynch_Accept in POSIX4 systems, in - // the case of doing AIO_CONTROL_BLOCKS strategy. + // This class manages the notify pipe of the AIOCB + // Proactor. This class acts as the Handler for the + // <Asynch_Read> operations issued on the notify pipe. This + // class is very useful in implementing <Asynch_Accept> operation + // class for the <AIOCB_Proactor>. This is also useful for + // implementing <post_completion> for <AIOCB_Proactor>. // // = DESCRIPTION - // Doing Asynch_Accept in POSIX4 implementation is tricky. In - // the case of doing the things with AIO_CONTROL_BLOCKS and not - // with Real-Time Signals, this becomes even more trickier. We - // use a notifyn pipe here to implement Asynch_Accpet. This class - // will issue a <Asynch_Read> on the pipe. <Asynch_Accept> will - // send a result pointer containg all the information through - // this pipe. + // <AIOCB_Proactor> class issues a <Asynch_Read> on + // the pipe, using this class as the + // Handler. <POSIX_Asynch_Result *>'s are sent through the + // notify pipe. When <POSIX_Asynch_Result *>'s show up on the + // notify pipe, the <POSIX_AIOCB_Proactor> dispatches the + // completion of the <Asynch_Read_Stream> and calls the + // <handle_read_stream> of this class. This class calls complete + // on the <POSIX_Asynch_Result *> and thus calls the application + // handler. // Handling the MessageBlock: // We give this message block to read the result pointer through // the notify pipe. We expect that to read 4 bytes from the @@ -443,41 +446,27 @@ class ACE_Export ACE_AIO_Accept_Handler : public ACE_Handler // message block to another <accept>, we update <wr_ptr> and put // it in its initial position. public: - ACE_AIO_Accept_Handler (ACE_POSIX_AIOCB_Proactor *posix_aiocb_proactor); + ACE_AIOCB_Notify_Pipe_Manager (ACE_POSIX_AIOCB_Proactor *posix_aiocb_proactor); // Constructor. You need the posix proactor because you need to call - // things like application_specific_code etc. Having the main - // proactor is good so that life is easier when we use Asynch_Read, - // while issuing asynchronous read on the notify pipe. - - virtual ~ACE_AIO_Accept_Handler (void); + // <application_specific_code> + + virtual ~ACE_AIOCB_Notify_Pipe_Manager (void); // Destructor. - -#if 0 - void proactor (ACE_Proactor *proactor); - // Set the Proactor pointer. POSIX_AIOCB_Proactor may not have the - // Proactor pointer when it creates this object and so we have this - // method so that it can set it when it gets it. -#endif /* 0 */ - - int notify (ACE_POSIX_Asynch_Accept_Result *result); - // Send this Result to Proactor through the notification pipe. + + int notify (ACE_POSIX_Asynch_Result *result); + // Send the result pointer through the notification pipe. virtual void handle_read_stream (const ACE_Asynch_Read_Stream::Result &result); - // Read from the pipe is complete. We get the <Result> from - // Asynch_Handler. We have to do the notification here. + // This is the call back method when <Asynch_Read> from the pipe is + // complete. private: -#if 0 - ACE_Proactor *proactor_; - // The main proactor interface. -#endif /* 0 */ - ACE_POSIX_AIOCB_Proactor *posix_aiocb_proactor_; // The implementation proactor class. - + ACE_Message_Block message_block_; - // Message block to get ACE_Asynch_Accept::Result pointer from - // ACE_Asych_Accept. + // Message block to get ACE_POSIX_Asynch_Result pointer from the + // pipe. ACE_Pipe pipe_; // Pipe for the communication between Proactor and the @@ -486,11 +475,11 @@ private: ACE_POSIX_AIOCB_Asynch_Read_Stream read_stream_; // To do asynch_read on the pipe. - ACE_AIO_Accept_Handler (void); + ACE_AIOCB_Notify_Pipe_Manager (void); // Default constructor. Shouldnt be called. }; -ACE_AIO_Accept_Handler::ACE_AIO_Accept_Handler (ACE_POSIX_AIOCB_Proactor *posix_aiocb_proactor) +ACE_AIOCB_Notify_Pipe_Manager::ACE_AIOCB_Notify_Pipe_Manager (ACE_POSIX_AIOCB_Proactor *posix_aiocb_proactor) : posix_aiocb_proactor_ (posix_aiocb_proactor), message_block_ (sizeof (ACE_POSIX_Asynch_Accept_Result *)), read_stream_ (posix_aiocb_proactor) @@ -510,21 +499,21 @@ ACE_AIO_Accept_Handler::ACE_AIO_Accept_Handler (ACE_POSIX_AIOCB_Proactor *posix_ // Issue an asynch_read on the read_stream of the notify pipe. if (this->read_stream_.read (this->message_block_, - sizeof (ACE_POSIX_Asynch_Accept_Result *), + sizeof (ACE_POSIX_Asynch_Result *), 0, // ACT 0) // Priority == -1) ACE_ERROR ((LM_ERROR, "%N:%l:%p\n", - "Read from stream failed")); + "Read from pipe failed")); } -ACE_AIO_Accept_Handler::~ACE_AIO_Accept_Handler (void) +ACE_AIOCB_Notify_Pipe_Manager::~ACE_AIOCB_Notify_Pipe_Manager (void) { } int -ACE_AIO_Accept_Handler::notify (ACE_POSIX_Asynch_Accept_Result *result) +ACE_AIOCB_Notify_Pipe_Manager::notify (ACE_POSIX_Asynch_Result *result) { // Send the result pointer through the pipe. int return_val = ACE::send (this->pipe_.write_handle (), @@ -533,24 +522,21 @@ ACE_AIO_Accept_Handler::notify (ACE_POSIX_Asynch_Accept_Result *result) if (return_val != sizeof (result)) ACE_ERROR_RETURN ((LM_ERROR, "(%P %t):%p\n", - "Error:Writing on to pipe failed"), + "Error:Writing on to notify pipe failed"), -1); return 0; } void -ACE_AIO_Accept_Handler::handle_read_stream (const ACE_Asynch_Read_Stream::Result &result) +ACE_AIOCB_Notify_Pipe_Manager::handle_read_stream (const ACE_Asynch_Read_Stream::Result &result) { - // @@ - ACE_DEBUG ((LM_DEBUG, "ACE_AIO_Accept_Handler::handle_read_stream called\n")); - - // The message block actually contains the ACE_POSIX_Asynch_Accept_Result - // pointer. - ACE_POSIX_Asynch_Accept_Result *accept_result = 0; - accept_result = *(ACE_POSIX_Asynch_Accept_Result **) result.message_block ().rd_ptr (); + // The message block actually contains the ACE_POSIX_Asynch_Result + // pointer. + ACE_POSIX_Asynch_Result *asynch_result = 0; + asynch_result = *(ACE_POSIX_Asynch_Result **) result.message_block ().rd_ptr (); // Do the upcall. - this->posix_aiocb_proactor_->application_specific_code (accept_result, + this->posix_aiocb_proactor_->application_specific_code (asynch_result, 0, // No Bytes transferred. 1, // Success. 0, // Completion token. @@ -561,22 +547,22 @@ ACE_AIO_Accept_Handler::handle_read_stream (const ACE_Asynch_Read_Stream::Result if (this->message_block_.length () > 0) this->message_block_.wr_ptr (this->message_block_.rd_ptr ()); - // One accept has completed. Issue a read to handle any <accept>s in - // the future. + // One accept has completed. Issue a read to handle any + // <post_completion>s in the future. if (this->read_stream_.read (this->message_block_, - sizeof (ACE_POSIX_Asynch_Accept_Result), + sizeof (ACE_POSIX_Asynch_Result *), 0, // ACT 0) // Priority == -1) ACE_ERROR ((LM_ERROR, "%N:%l:%p\n", - "Read from stream failed")); + "Read from pipe failed")); } // ********************************************************************* ACE_POSIX_AIOCB_Proactor::ACE_POSIX_AIOCB_Proactor (void) - : aio_accept_handler_ (0), + : aiocb_notify_pipe_manager_ (0), aiocb_list_max_size_ (ACE_RTSIG_MAX), aiocb_list_cur_size_ (0) { @@ -586,8 +572,8 @@ ACE_POSIX_AIOCB_Proactor::ACE_POSIX_AIOCB_Proactor (void) // Accept Handler for aio_accept. Remember! this issues a Asynch_Read // on the notify pipe for doing the Asynch_Accept. - ACE_NEW (aio_accept_handler_, - ACE_AIO_Accept_Handler (this)); + ACE_NEW (aiocb_notify_pipe_manager_, + ACE_AIOCB_Notify_Pipe_Manager (this)); } // Destructor. @@ -612,8 +598,7 @@ ACE_POSIX_AIOCB_Proactor::handle_events (void) int ACE_POSIX_AIOCB_Proactor::post_completion (ACE_POSIX_Asynch_Result *result) { - ACE_UNUSED_ARG (result); - return 0; + return this->aiocb_notify_pipe_manager_->notify (result); } ACE_Asynch_Read_Stream_Impl * @@ -677,14 +662,6 @@ ACE_POSIX_AIOCB_Proactor::create_asynch_transmit_file (void) } int -ACE_POSIX_AIOCB_Proactor::notify_asynch_accept (ACE_POSIX_Asynch_Accept_Result* result) -{ - this->aio_accept_handler_->notify (result); - - return 0; -} - -int ACE_POSIX_AIOCB_Proactor::handle_events (unsigned long milli_seconds) { if (this->aiocb_list_cur_size_ == 0) @@ -917,7 +894,24 @@ ACE_POSIX_SIG_Proactor::handle_events (void) int ACE_POSIX_SIG_Proactor::post_completion (ACE_POSIX_Asynch_Result *result) { - ACE_UNUSED_ARG (result); + // Get this process id. + pid_t pid = ACE_OS::getpid (); + if (pid == (pid_t) -1) + ACE_ERROR_RETURN ((LM_ERROR, + "Error:(%P | %t):%p", + "<getpid> failed\n"), + -1); + + // Set the signal information. + sigval value; + value.sival_ptr = (void *) result; + + // Queue the signal. + if (sigqueue (pid, ACE_SIG_AIO, value) == -1) + ACE_ERROR_RETURN ((LM_ERROR, + "Error:(%P | %t):%p", + "<sigqueue> failed\n"), + -1); return 0; } diff --git a/ace/POSIX_Proactor.h b/ace/POSIX_Proactor.h index 014dc49babc..0f57c7942e2 100644 --- a/ace/POSIX_Proactor.h +++ b/ace/POSIX_Proactor.h @@ -44,8 +44,8 @@ class ACE_Export ACE_POSIX_Proactor : public ACE_Proactor_Impl // operations are queried on them. The other one is based on // POSIX Real Time signals. This class abstracts out the common // code needed for both the strategies. ACE_AIOCB_Proactor and - // ACE_SIG_Proactor specialize this. - + // ACE_SIG_Proactor specialize this class for each strategy. + public: virtual ~ACE_POSIX_Proactor (void); // Virtual destructor. @@ -55,8 +55,7 @@ public: virtual int register_handle (ACE_HANDLE handle, const void *completion_key); - // This method adds the <handle> to the I/O completion port. This - // function is a no-op function for Unix systems. + // This function is a no-op function for Unix systems. Returns 0. // = Timer management. @@ -107,7 +106,7 @@ public: // Post a result to the completion port of the Proactor. If errors // occur, the result will be deleted by this method. If successful, // the result will be deleted by the Proactor when the result is - // removed from the completion port. Therefore, the result should + // removed from the completion port. Therefore, the result should // have been dynamically allocated and should be orphaned by the // user once this method is called. @@ -236,7 +235,7 @@ protected: }; // Forward declarations. -class ACE_AIO_Accept_Handler; +class ACE_AIOCB_Notify_Pipe_Manager; class ACE_Export ACE_POSIX_AIOCB_Proactor : public ACE_POSIX_Proactor { @@ -249,7 +248,7 @@ class ACE_Export ACE_POSIX_AIOCB_Proactor : public ACE_POSIX_Proactor // = DESCRIPTION // - friend class ACE_AIO_Accept_Handler; + friend class ACE_AIOCB_Notify_Pipe_Manager; // Handler needs to call application specific code. friend class ACE_POSIX_AIOCB_Asynch_Operation; @@ -296,10 +295,6 @@ public: virtual ACE_Asynch_Transmit_File_Impl *create_asynch_transmit_file (void); protected: - virtual int notify_asynch_accept (ACE_POSIX_Asynch_Accept_Result* result); - // Asynch_Accept calls this function to notify an accept to the - // Proactor. - virtual int handle_events (unsigned long milli_seconds); // Dispatch a single set of events. If <milli_seconds> elapses // before any events occur, return. @@ -317,7 +312,7 @@ protected: // return 0 if yes, else return -1. If a valid ptr is passed, keep it // in a free slot. - ACE_AIO_Accept_Handler* aio_accept_handler_; + ACE_AIOCB_Notify_Pipe_Manager* aiocb_notify_pipe_manager_; // This class takes care of doing <accept> when we use // AIO_CONTROL_BLOCKS strategy. diff --git a/ace/Proactor.cpp b/ace/Proactor.cpp index e34ae8b02db..61989d99508 100644 --- a/ace/Proactor.cpp +++ b/ace/Proactor.cpp @@ -11,7 +11,6 @@ ACE_RCSID(ace, Proactor, "$Id$") // calls. #include "ace/Task_T.h" #include "ace/Log_Msg.h" -#include "ace/Object_Manager.h" #if defined (ACE_HAS_AIO_CALLS) # include "ace/POSIX_Proactor.h" diff --git a/ace/Proactor.h b/ace/Proactor.h index 0af38ff9904..fa7cc470911 100644 --- a/ace/Proactor.h +++ b/ace/Proactor.h @@ -10,8 +10,8 @@ // Proactor.h // // = AUTHOR -// Irfan Pyarali (irfan@cs.wustl.edu), -// Tim Harrison (harrison@cs.wustl.edu) and +// Irfan Pyarali <irfan@cs.wustl.edu>, +// Tim Harrison <harrison@cs.wustl.edu> and // Alexander Babu Arulanthu <alex@cs.wustl.edu> // // ============================================================================ @@ -87,7 +87,7 @@ public: virtual int register_handle (ACE_HANDLE handle, const void *completion_key); // This method adds the <handle> to the I/O completion port. This - // function is a no-op function for Unix systems. + // function is a no-op function for Unix systems and returns 0; // = Timer management. virtual long schedule_timer (ACE_Handler &handler, diff --git a/ace/Proactor_Impl.h b/ace/Proactor_Impl.h index d0a4bd17305..ea5a43ae177 100644 --- a/ace/Proactor_Impl.h +++ b/ace/Proactor_Impl.h @@ -105,17 +105,7 @@ public: // Block indefinitely until at least one event is dispatched. // Return 0 on success, non-zero (-1) on timeouts/errors and errno // is set accordingly. - -#if 0 - virtual int post_completion (ACE_Asynch_Result *result) = 0; - // Post a result to the completion port of the Proactor. If errors - // occur, the result will be deleted by this method. If successful, - // the result will be deleted by the Proactor when the result is - // removed from the completion port. Therefore, the result should - // have been dynamically allocated and should be orphaned by the - // user once this method is called. -#endif /* 0 */ - + virtual int wake_up_dispatch_threads (void) = 0; // Add wakeup dispatch threads (reinit). diff --git a/ace/WIN32_Proactor.cpp b/ace/WIN32_Proactor.cpp index fd4b7c1025e..402fb63f7ef 100644 --- a/ace/WIN32_Proactor.cpp +++ b/ace/WIN32_Proactor.cpp @@ -710,7 +710,7 @@ ACE_WIN32_Proactor::post_completion (ACE_WIN32_Asynch_Result *result) { // Grab the event associated with the Proactor HANDLE handle = this->get_handle (); - + // If Proactor event is valid, signal it if (handle != ACE_INVALID_HANDLE && handle != 0) @@ -718,9 +718,9 @@ ACE_WIN32_Proactor::post_completion (ACE_WIN32_Asynch_Result *result) // Post a completion if (::PostQueuedCompletionStatus (this->completion_port_, // completion port - 0, // number of bytes tranferred - 0, // completion key - result // overlapped + 0, // number of bytes tranferred + 0, // completion key + result // overlapped ) == FALSE) { delete result; |