ChangeLogTag:Wed Aug 25 19:58:19 UTC 2010 Steve Huston <shuston@riverace.com>

3 files changed, 62 insertions, 49 deletions

diff --git a/ChangeLog b/ChangeLog
index 6da46700d9b..bfb4621f09a 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -8,6 +8,22 @@ Fri Aug 27 15:01:41 UTC 2010  Steve Huston  <shuston@riverace.com>
           failed fd for read. Thanks to Kannan Ramaswamy <kannan dot
           ramaswamy at cogcap dot com> for this information and fix.
 
+Wed Aug 25 19:58:19 UTC 2010  Steve Huston  <shuston@riverace.com>
+
+        * ace/Dev_Poll_Reactor.{h cpp}: Regarding change number 2 at
+          Fri Mar 26 14:07:55 UTC 2010  Steve Huston  <shuston@riverace.com>
+          dispatching notifies in only one thread at a time was 1) wrong
+          (TP_Reactor doesn't do this), 2) broke important existing
+          applications at a customer site. This has been fixed so that the
+          notify dispatch behavior is like ACE_TP_Reactor: when the
+          notify handle is signaled, extract one notification, release
+          the token, dispatch the notification. Thus, each thread will
+          dispatch one notification, and multiple threads can dispatch
+          them at the same time if there are multiples.
+
+          Also fixed an error where it was possible to lose notice of a
+          notification. Resolves Bugzilla #3328.
+
 Thu Jul 15 17:53:47 UTC 2010  Phil Mesnier  <mesnier_p@ociweb.com>
 
         * ace/Service_Gestalt.cpp:
diff --git a/ace/Dev_Poll_Reactor.cpp b/ace/Dev_Poll_Reactor.cpp
index 222b1d64808..feb9ff4ae3f 100644
--- a/ace/Dev_Poll_Reactor.cpp
+++ b/ace/Dev_Poll_Reactor.cpp
@@ -53,7 +53,6 @@ ACE_Dev_Poll_Reactor_Notify::ACE_Dev_Poll_Reactor_Notify (void)
 #if defined (ACE_HAS_REACTOR_NOTIFICATION_QUEUE)
   , notification_queue_ ()
 #endif  /* ACE_HAS_REACTOR_NOTIFICATION_QUEUE */
-  , dispatching_ (false)
 {
 }
 
@@ -223,12 +222,20 @@ ACE_Dev_Poll_Reactor_Notify::read_notify_pipe (ACE_HANDLE handle,
 
   bool more_messages_queued = false;
   ACE_Notification_Buffer next;
-  int result = notification_queue_.pop_next_notification (buffer,
+  int result = 1;
+  while (result == 1)
+    {
+      result = notification_queue_.pop_next_notification (buffer,
                                                           more_messages_queued,
                                                           next);
 
-  if (result <= 0)   // Nothing dequeued or error
-    return result;
+      if (result <= 0)   // Nothing dequeued or error
+        return result;
+
+      // If it's just a wake-up, toss it and see if there's anything else.
+      if (buffer.eh_ != 0)
+        break;
+    }
 
   // If there are more messages, ensure there's a byte in the pipe
   // in case the notification limit stops dequeuing notifies before
@@ -272,45 +279,10 @@ ACE_Dev_Poll_Reactor_Notify::read_notify_pipe (ACE_HANDLE handle,
 
 
 int
-ACE_Dev_Poll_Reactor_Notify::handle_input (ACE_HANDLE handle)
+ACE_Dev_Poll_Reactor_Notify::handle_input (ACE_HANDLE /*handle*/)
 {
   ACE_TRACE ("ACE_Dev_Poll_Reactor_Notify::handle_input");
-
-  {
-    ACE_MT (ACE_GUARD_RETURN (ACE_SYNCH_MUTEX, mon, this->dispatching_lock_, -1));
-    if (this->dispatching_)
-      return 0;
-    this->dispatching_ = true;
-  }
-
-  int number_dispatched = 0;
-  int result = 0;
-  ACE_Notification_Buffer buffer;
-
-  while ((result = this->read_notify_pipe (handle, buffer)) > 0)
-    {
-      // Dispatch the buffer
-      // NOTE: We count only if we made any dispatches ie. upcalls.
-      if (this->dispatch_notify (buffer) > 0)
-        ++number_dispatched;
-
-      // Bail out if we've reached the <notify_threshold_>.  Note that
-      // by default <notify_threshold_> is -1, so we'll loop until all
-      // the available notifications have been dispatched.
-      if (number_dispatched == this->max_notify_iterations_)
-        break;
-    }
-
-  if (result == -1)
-    {
-      // Reassign number_dispatched to -1 if things have gone
-      // seriously wrong.
-      number_dispatched = -1;
-    }
-
-  this->dispatching_ = false;
-
-  return number_dispatched;
+  ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("SHOULD NOT BE HERE.\n")), -1);
 }
 
 ACE_HANDLE
@@ -426,6 +398,15 @@ ACE_Dev_Poll_Reactor_Notify::dump (void) const
 #endif /* ACE_HAS_DUMP */
 }
 
+int
+ACE_Dev_Poll_Reactor_Notify::dequeue_one (ACE_Notification_Buffer &nb)
+{
+  nb.eh_ = 0;
+  nb.mask_ = 0;
+  return this->read_notify_pipe (this->notify_handle (), nb);
+}
+
+
 // -----------------------------------------------------------------
 
 ACE_Dev_Poll_Reactor::Handler_Repository::Handler_Repository (void)
@@ -1309,6 +1290,22 @@ ACE_Dev_Poll_Reactor::dispatch_io_event (Token_Guard &guard)
 #endif /* ACE_HAS_DEV_POLL */
 
       int status = 0;   // gets callback status, below.
+
+      // Dispatch notifies directly. The notify dispatcher locates a
+      // notification then releases the token prior to dispatching it.
+      // NOTE: If notify_handler_->dispatch_one() returns a fail condition
+      // it has not releases the guard. Else, it has.
+      if (eh == this->notify_handler_)
+        {
+          ACE_Notification_Buffer b;
+          status =
+            dynamic_cast<ACE_Dev_Poll_Reactor_Notify *>(notify_handler_)->dequeue_one (b);
+          if (status == -1)
+            return status;
+          guard.release_token ();
+          return notify_handler_->dispatch_notify (b);
+        }
+
       {
         // Modify the reference count in an exception-safe way.
         // Note that eh could be the notify handler. It's not strictly
@@ -1327,9 +1324,6 @@ ACE_Dev_Poll_Reactor::dispatch_io_event (Token_Guard &guard)
         // back with either 0 or < 0.
         status = this->upcall (eh, callback, handle);
 
-        if (eh == this->notify_handler_)
-          return status;
-
         // If the callback returned 0, epoll-based needs to resume the
         // suspended handler but dev/poll doesn't.
         // The epoll case is optimized to not acquire the token in order
diff --git a/ace/Dev_Poll_Reactor.h b/ace/Dev_Poll_Reactor.h
index 878e7a8f78c..d741e1d7d77 100644
--- a/ace/Dev_Poll_Reactor.h
+++ b/ace/Dev_Poll_Reactor.h
@@ -206,6 +206,14 @@ public:
   /// Dump the state of an object.
   virtual void dump (void) const;
 
+  /// Method called by ACE_Dev_Poll_Reactor to obtain one notification.
+  /// THIS METHOD MUST BE CALLED WITH THE REACTOR TOKEN HELD!
+  ///
+  /// @return -1 on error, else 0 and @arg nb has the notify to
+  ///            dispatch. Note that the contained event handler may be
+  ///            0 if there were only wake-ups (no handlers to dispatch).
+  int dequeue_one (ACE_Notification_Buffer &nb);
+
 protected:
 
   /**
@@ -245,11 +253,6 @@ protected:
    */
   ACE_Notification_Queue notification_queue_;
 #endif /* ACE_HAS_REACTOR_NOTIFICATION_QUEUE */
-
-  /// Lock and flag to say whether we're already dispatching notifies.
-  /// Purpose is to only dispatch notifies from one thread at a time.
-  ACE_SYNCH_MUTEX dispatching_lock_;
-  volatile bool dispatching_;
 };
 
 // ---------------------------------------------------------------------