dep/ACE_wrappers/ace/POSIX_CB_Proactor.cpp

   1 // $Id: POSIX_CB_Proactor.cpp 80826 2008-03-04 14:51:23Z wotte $
   2
   3 #include "ace/POSIX_CB_Proactor.h"
   4
   5 #if defined (ACE_HAS_AIO_CALLS) && !defined (ACE_HAS_BROKEN_SIGEVENT_STRUCT)
   6
   7 #include "ace/Task_T.h"
   8 #include "ace/Log_Msg.h"
   9 #include "ace/Object_Manager.h"
  10 #include "ace/OS_NS_sys_time.h"
  11
  12 ACE_RCSID (ace,
  13            POSIX_CB_Proactor,
  14            "$Id: POSIX_CB_Proactor.cpp 80826 2008-03-04 14:51:23Z wotte $")
  15
  16 ACE_BEGIN_VERSIONED_NAMESPACE_DECL
  17
  18 ACE_POSIX_CB_Proactor::ACE_POSIX_CB_Proactor (size_t max_aio_operations)
  19   : ACE_POSIX_AIOCB_Proactor (max_aio_operations,
  20                               ACE_POSIX_Proactor::PROACTOR_CB),
  21     sema_ ((unsigned int) 0)
  22 {
  23   // we should start pseudo-asynchronous accept task
  24   // one per all future acceptors
  25
  26   this->get_asynch_pseudo_task ().start ();
  27 }
  28
  29 // Destructor.
  30 ACE_POSIX_CB_Proactor::~ACE_POSIX_CB_Proactor (void)
  31 {
  32   this->close ();
  33 }
  34
  35 ACE_POSIX_Proactor::Proactor_Type
  36 ACE_POSIX_CB_Proactor::get_impl_type (void)
  37 {
  38   return PROACTOR_CB;
  39 }
  40
  41 void ACE_POSIX_CB_Proactor::aio_completion_func (sigval cb_data)
  42 {
  43   ACE_POSIX_CB_Proactor * impl = static_cast<ACE_POSIX_CB_Proactor *> (cb_data.sival_ptr);
  44   if ( impl != 0 )
  45     impl->notify_completion (0);
  46 }
  47
  48 #if defined (ACE_HAS_SIG_C_FUNC)
  49 extern "C" void
  50 ACE_POSIX_CB_Proactor_aio_completion (sigval cb_data)
  51 {
  52   ACE_POSIX_CB_Proactor::aio_completion_func (cb_data);
  53 }
  54 #endif /* ACE_HAS_SIG_C_FUNC */
  55
  56 int
  57 ACE_POSIX_CB_Proactor::handle_events (ACE_Time_Value &wait_time)
  58 {
  59   // Decrement <wait_time> with the amount of time spent in the method
  60   ACE_Countdown_Time countdown (&wait_time);
  61   return this->handle_events_i (wait_time.msec ());
  62 }
  63
  64 int
  65 ACE_POSIX_CB_Proactor::handle_events (void)
  66 {
  67   return this->handle_events_i (ACE_INFINITE);
  68 }
  69
  70 int
  71 ACE_POSIX_CB_Proactor::notify_completion (int sig_num)
  72 {
  73   ACE_UNUSED_ARG (sig_num);
  74
  75   return this->sema_.release();
  76 }
  77
  78
  79 ssize_t
  80 ACE_POSIX_CB_Proactor::allocate_aio_slot (ACE_POSIX_Asynch_Result *result)
  81 {
  82   ssize_t slot = ACE_POSIX_AIOCB_Proactor::allocate_aio_slot (result);
  83   if (slot == -1)
  84     return -1;
  85
  86   // setup OS notification methods for this aio
  87   // @@ TODO: This gets the completion method back to this proactor to
  88   // find the completed aiocb. It would be so much better to not only get
  89   // the proactor, but the aiocb as well.
  90 #if defined(__sgi)
  91   result->aio_sigevent.sigev_notify = SIGEV_CALLBACK;
  92   result->aio_sigevent.sigev_func   = aio_completion_func ;
  93 #else
  94   result->aio_sigevent.sigev_notify = SIGEV_THREAD;
  95 #  if defined (ACE_HAS_SIG_C_FUNC)
  96   result->aio_sigevent.sigev_notify_function =
  97     ACE_POSIX_CB_Proactor_aio_completion;
  98 #  else
  99   result->aio_sigevent.sigev_notify_function = aio_completion_func;
 100 #  endif /* ACE_HAS_SIG_C_FUNC */
 101   result->aio_sigevent.sigev_notify_attributes = 0;
 102 #endif /* __sgi */
 103
 104   result->aio_sigevent.sigev_value.sival_ptr = this ;
 105
 106   return slot;
 107 }
 108
 109 int
 110 ACE_POSIX_CB_Proactor::handle_events_i (u_long milli_seconds)
 111 {
 112
 113   int result_wait=0;
 114
 115   // Wait for the signals.
 116   if (milli_seconds == ACE_INFINITE)
 117     {
 118       result_wait = this->sema_.acquire();
 119     }
 120   else
 121     {
 122       // Wait for <milli_seconds> amount of time.
 123       ACE_Time_Value abs_time = ACE_OS::gettimeofday ()
 124                               + ACE_Time_Value (0, milli_seconds * 1000);
 125
 126       result_wait = this->sema_.acquire(abs_time);
 127     }
 128
 129   // Check for errors
 130   // but let continue work in case of errors
 131   // we should check "post_completed" queue
 132   if (result_wait == -1)
 133     {
 134       int const lerror = errno;
 135       if (lerror != ETIME &&   // timeout
 136           lerror != EINTR )    // interrupted system call
 137         ACE_ERROR ((LM_ERROR,
 138                     "%N:%l:(%P | %t)::%p\n",
 139                     "ACE_POSIX_CB_Proactor::handle_events:"
 140                     "semaphore acquire failed"
 141                   ));
 142     }
 143
 144   size_t index = 0;          // start index to scan aiocb list
 145   size_t count = this->aiocb_list_max_size_;  // max number to iterate
 146
 147   int error_status = 0;
 148   size_t return_status = 0;
 149
 150   int ret_aio = 0;
 151   int ret_que = 0;
 152
 153   for (;; ret_aio++)
 154     {
 155       ACE_POSIX_Asynch_Result * asynch_result =
 156           this->find_completed_aio (error_status,
 157                                     return_status,
 158                                     index,
 159                                     count);
 160
 161       if (asynch_result == 0)
 162           break;
 163
 164       // Call the application code.
 165       this->application_specific_code (asynch_result,
 166                                        return_status, // Bytes transferred.
 167                                        0,             // No completion key.
 168                                        error_status); // Error
 169      }
 170
 171   // process post_completed results
 172   ret_que = this->process_result_queue ();
 173
 174   // Uncomment this  if you want to test
 175   // and research the behavior of you system
 176   // ACE_DEBUG ((LM_DEBUG,
 177   //            "(%t) NumAIO=%d NumQueue=%d\n",
 178   //             ret_aio, ret_que));
 179
 180   return ret_aio + ret_que > 0 ? 1 : 0;
 181 }
 182
 183 ACE_END_VERSIONED_NAMESPACE_DECL
 184
 185 #endif /* ACE_HAS_AIO_CALLS && !ACE_HAS_BROKEN_SIGEVENT_STRUCT */