event/epoll: Add back socket for polling of events immediately after

             reading the entire rpc message from the wire

Currently socket is added back for future events after higher layers
(rpc, xlators etc) have processed the message. If message processing
involves signficant delay (as in writev replies processed by Erasure
Coding), performance takes hit. Hence this patch modifies
transport/socket to add back the socket for polling of events
immediately after reading the entire rpc message, but before
notification to higher layers.

credits: Thanks to "Kotresh Hiremath Ravishankar"
         <khiremat@redhat.com> for assitance in fixing a regression in
         bitrot caused by this patch.

Change-Id: I04b6b9d0b51a1cfb86ecac3c3d87a5f388cf5800
BUG: 1448364
Signed-off-by: Raghavendra G <rgowdapp@redhat.com>
Reviewed-on: https://review.gluster.org/15036
CentOS-regression: Gluster Build System <jenkins@build.gluster.org>
NetBSD-regression: NetBSD Build System <jenkins@build.gluster.org>
Smoke: Gluster Build System <jenkins@build.gluster.org>
Reviewed-by: Amar Tumballi <amarts@redhat.com>

1 files changed, 52 insertions, 29 deletions

diff --git a/libglusterfs/src/event-epoll.c b/libglusterfs/src/event-epoll.c
index e2b40602e7a..4b76cc96fd3 100644
--- a/libglusterfs/src/event-epoll.c
+++ b/libglusterfs/src/event-epoll.c
@@ -569,38 +569,11 @@ pre_unlock:
         if (!handler)
 		goto out;
 
-	ret = handler (fd, idx, data,
+	ret = handler (fd, idx, gen, data,
 		       (event->events & (EPOLLIN|EPOLLPRI)),
 		       (event->events & (EPOLLOUT)),
 		       (event->events & (EPOLLERR|EPOLLHUP)));
 
-	LOCK (&slot->lock);
-	{
-		slot->in_handler--;
-
-		if (gen != slot->gen) {
-			/* event_unregister() happened while we were
-			   in handler()
-			*/
-			gf_msg_debug ("epoll", 0, "generation bumped on idx=%d"
-                                      " from gen=%d to slot->gen=%d, fd=%d, "
-				      "slot->fd=%d", idx, gen, slot->gen, fd,
-                                      slot->fd);
-			goto post_unlock;
-		}
-
-		/* This call also picks up the changes made by another
-		   thread calling event_select_on_epoll() while this
-		   thread was busy in handler()
-		*/
-                if (slot->in_handler == 0) {
-                        event->events = slot->events;
-                        ret = epoll_ctl (event_pool->fd, EPOLL_CTL_MOD,
-                                         fd, event);
-                }
-	}
-post_unlock:
-	UNLOCK (&slot->lock);
 out:
 	event_slot_unref (event_pool, slot, idx);
 
@@ -891,6 +864,55 @@ event_pool_destroy_epoll (struct event_pool *event_pool)
         return ret;
 }
 
+static int
+event_handled_epoll (struct event_pool *event_pool, int fd, int idx, int gen)
+{
+        struct event_slot_epoll *slot  = NULL;
+        struct epoll_event epoll_event = {0, };
+        struct event_data *ev_data     = (void *)&epoll_event.data;
+        int                ret         = 0;
+
+	slot = event_slot_get (event_pool, idx);
+
+        assert (slot->fd == fd);
+
+	LOCK (&slot->lock);
+	{
+		slot->in_handler--;
+
+		if (gen != slot->gen) {
+			/* event_unregister() happened while we were
+			   in handler()
+			*/
+			gf_msg_debug ("epoll", 0, "generation bumped on idx=%d"
+                                      " from gen=%d to slot->gen=%d, fd=%d, "
+				      "slot->fd=%d", idx, gen, slot->gen, fd,
+                                      slot->fd);
+			goto post_unlock;
+		}
+
+		/* This call also picks up the changes made by another
+		   thread calling event_select_on_epoll() while this
+		   thread was busy in handler()
+		*/
+                if (slot->in_handler == 0) {
+                        epoll_event.events = slot->events;
+                        ev_data->idx = idx;
+                        ev_data->gen = gen;
+
+                        ret = epoll_ctl (event_pool->fd, EPOLL_CTL_MOD,
+                                         fd, &epoll_event);
+                }
+	}
+post_unlock:
+	UNLOCK (&slot->lock);
+
+        event_slot_unref (event_pool, slot, idx);
+
+        return ret;
+}
+
+
 struct event_ops event_ops_epoll = {
         .new                       = event_pool_new_epoll,
         .event_register            = event_register_epoll,
@@ -899,7 +921,8 @@ struct event_ops event_ops_epoll = {
         .event_unregister_close    = event_unregister_close_epoll,
         .event_dispatch            = event_dispatch_epoll,
         .event_reconfigure_threads = event_reconfigure_threads_epoll,
-        .event_pool_destroy        = event_pool_destroy_epoll
+        .event_pool_destroy        = event_pool_destroy_epoll,
+        .event_handled             = event_handled_epoll,
 };
 
 #endif