MB-25822: Create separate mutex for pending_io

This is a recreate of commit 0800742 against our spock branch. Lots
of refactoring has occurred between spock to vulcan, meaning a simple
cherry-pick of the original patch was not easy.

This patch recreates the original intent and switches the pending-io
from a list to std::map (rather than the original switch from std::set
to std::map)

The LIBEVENT_THREAD::mutex currently guards the whole state of the
worker thread - as such it is acquired by the worker thread when it
starts processing event(s) for a connection, and only unlocked when it
finishes with that connection.

The state guarded includes the pending_io set; which is (primarily):

a) Read at the start of the event loop (to re-register connections in
   libevent).

b) Written by notify_io_complete when background threads want to
   wakeup frontend threads.

Profiling of the LIBEVENT_THREAD::mutex wait/lock times shows that
callers of notify_io_complete (notably ConnManager and DCP
ConnNotifier task) can end up waiting for long periods (~100ms in
local testing) to simply add a connection to the pending set.

To improve this, add a separate mutex which is only used for the
pending_io set. This also requires a change to the data structure for
the pending_io - we need to additionally record the pending status
code against each connection.

This is because previously this status (aiostat) was updated with
LIBEVENT_THREAD::mutex held; therefore no worker thread could possibly
be running. With the separate mutex this is no longer the case and we
cannot simply write to aiostat as it may be being accessed by the
worker. We obviously don't want to have to acquire
LIBEVENT_THREAD::mutex to update it as that introduces the same
problem. Instead we record the pending status with the connection in
pending_io; and when we come to process pending_io in the worker
thread (which already holds LIBEVENT_THREAD::mutex) we update aiostat
with the pending status before running it's event loop).

Change-Id: Ifad49f12d2249b92d4ffd3a761e353bb85d0d2d5
Reviewed-on: http://review.couchbase.org/94350
Well-Formed: Build Bot <[email protected]>
Reviewed-by: Trond Norbye <[email protected]>
Tested-by: Build Bot <[email protected]>

Author: jimwwalker

daemon/connections.cc

@@ -297,12 +297,18 @@ void conn_close(McbpConnection *c) {
     if (thread == nullptr) {
         throw std::logic_error("conn_close: unable to obtain non-NULL thread from connection");
     }
-    /* remove from pending-io list */
-    if (settings.getVerbose() > 1 && list_contains(thread->pending_io, c)) {
-        LOG_WARNING(c,
-                    "Current connection was in the pending-io list.. Nuking it");
+
+    // remove from pending-io list
+    {
+        std::lock_guard<std::mutex> lock(*thread->pending_io.mutex);
+        /* remove from pending-io list */
+        if (settings.getVerbose() > 1 && thread->pending_io.map->count(c)) {
+            LOG_WARNING(c,
+                        "Current connection was in the pending-io list.. "
+                        "Nuking it");
+        }
+        thread->pending_io.map->erase(c);
     }
-    thread->pending_io = list_remove(thread->pending_io, c);
 
     conn_cleanup(c);
 

daemon/memcached.cc

@@ -1025,6 +1025,16 @@ void event_handler(evutil_socket_t fd, short which, void *arg) {
         return;
     }
 
+    // Remove the list from the list of pending io's (in case the
+    // object was scheduled to run in the dispatcher before the
+    // callback for the worker thread is executed.
+    //
+    {
+        std::lock_guard<std::mutex> lock(*thr->pending_io.mutex);
+        thr->pending_io.map->erase(c);
+    }
+
+
     LOCK_THREAD(thr);
     if (memcached_shutdown) {
         // Someone requested memcached to shut down.
@@ -1036,13 +1046,6 @@ void event_handler(evutil_socket_t fd, short which, void *arg) {
         }
     }
 
-    /*
-     * Remove the list from the list of pending io's (in case the
-     * object was scheduled to run in the dispatcher before the
-     * callback for the worker thread is executed.
-     */
-    thr->pending_io = list_remove(thr->pending_io, c);
-
     /* sanity */
     cb_assert(fd == c->getSocketDescriptor());
 
@@ -1684,7 +1687,7 @@ static ENGINE_ERROR_CODE release_cookie(const void *void_cookie) {
     }
 
     Connection *c = cookie->connection;
-    int notify;
+    int notify = 0;
     LIBEVENT_THREAD *thr;
 
     thr = c->getThread();
@@ -1698,7 +1701,7 @@ static ENGINE_ERROR_CODE release_cookie(const void *void_cookie) {
      * pending IO and have the system retry the operation for the
      * connection
      */
-    notify = add_conn_to_pending_io_list(c);
+    notify = add_conn_to_pending_io_list(c, ENGINE_SUCCESS);
     UNLOCK_THREAD(thr);
 
     /* kick the thread in the butt */

daemon/memcached.h

@@ -72,14 +72,28 @@ class Connection;
 class ConnectionQueue;
 
 struct LIBEVENT_THREAD {
+    /**
+     * Pending IO requests for this thread. Maps each pending Connection to
+     * the IO status to be notified.
+     */
+    using PendingIoMap = std::unordered_map<Connection*, ENGINE_ERROR_CODE>;
+
     cb_thread_t thread_id;      /* unique ID of this thread */
     struct event_base *base;    /* libevent handle this thread uses */
     struct event notify_event;  /* listen event for notify pipe */
     SOCKET notify[2];           /* notification pipes */
     ConnectionQueue *new_conn_queue; /* queue of new connections to handle */
-    cb_mutex_t mutex;      /* Mutex to lock protect access to the pending_io */
+    cb_mutex_t mutex;      /* Mutex for access to this structure */
     bool is_locked;
-    Connection *pending_io;    /* List of connection with pending async io ops */
+
+    struct {
+        // Note: these are pointers as this struct is calloc'd and we need to
+        // ensure correct construct/destruct of these objects on all platforms
+        // which is much simpler if we new/delete these two objects
+        std::mutex* mutex; /// Mutex for access to the PendingIoMap
+        PendingIoMap* map; /// map of connection with pending async io ops
+    } pending_io;
+
     int index;                  /* index of this thread in the threads array */
     ThreadType type;      /* Type of IO this thread processes */
 
@@ -171,7 +185,7 @@ Connection *list_remove(Connection *h, Connection *n);
 
 bool load_extension(const char *soname, const char *config);
 
-int add_conn_to_pending_io_list(Connection *c);
+int add_conn_to_pending_io_list(Connection* c, ENGINE_ERROR_CODE status);
 
 /* connection state machine */
 bool conn_listening(ListenConnection *c);

daemon/thread.cc

@@ -205,6 +205,9 @@ static void setup_thread(LIBEVENT_THREAD *me) {
 
     cb_mutex_initialize(&me->mutex);
 
+    me->pending_io.mutex = new std::mutex();
+    me->pending_io.map = new LIBEVENT_THREAD::PendingIoMap();
+
     // Initialize threads' sub-document parser / handler
     me->subdoc_op = subdoc_op_alloc();
 
@@ -236,16 +239,6 @@ static void worker_libevent(void *arg) {
     ERR_remove_state(0);
 }
 
-static int number_of_pending(Connection *c, Connection *list) {
-    int rv = 0;
-    for (; list; list = list->getNext()) {
-        if (list == c) {
-            rv ++;
-        }
-    }
-    return rv;
-}
-
 static void drain_notification_channel(evutil_socket_t fd)
 {
     int nread;
@@ -309,14 +302,19 @@ static void thread_libevent_process(evutil_socket_t fd, short which, void *arg)
 
     dispatch_new_connections(me);
 
+    LIBEVENT_THREAD::PendingIoMap pending;
+    {
+        std::lock_guard<std::mutex> lock(*me->pending_io.mutex);
+        me->pending_io.map->swap(pending);
+    }
+
     LOCK_THREAD(me);
-    Connection* pending = me->pending_io;
-    me->pending_io = NULL;
-    while (pending != NULL) {
-        Connection *c = pending;
+
+    for (auto io : pending) {
+        auto* c = io.first;
+        auto status = io.second;
+
         cb_assert(me == c->getThread());
-        pending = pending->getNext();
-        c->setNext(nullptr);
 
         auto *mcbp = dynamic_cast<McbpConnection*>(c);
         if (mcbp != nullptr) {
@@ -332,6 +330,7 @@ static void thread_libevent_process(evutil_socket_t fd, short which, void *arg)
              * from the context of the notification pipe, so just let it
              * run one time to set up the correct mask in libevent
              */
+            mcbp->setAiostat(status);
             mcbp->setNumEvents(1);
         }
         run_event_loop(c, EV_READ|EV_WRITE);
@@ -364,23 +363,6 @@ static void thread_libevent_process(evutil_socket_t fd, short which, void *arg)
 
 extern volatile rel_time_t current_time;
 
-static bool has_cycle(Connection *c) {
-    Connection *slowNode, *fastNode1, *fastNode2;
-
-    if (!c) {
-        return false;
-    }
-
-    slowNode = fastNode1 = fastNode2 = c;
-    while (slowNode && (fastNode1 = fastNode2->getNext()) && (fastNode2 = fastNode1->getNext())) {
-        if (slowNode == fastNode1 || slowNode == fastNode2) {
-            return true;
-        }
-        slowNode = slowNode->getNext();
-    }
-    return false;
-}
-
 bool list_contains(Connection *haystack, Connection *needle) {
     for (; haystack; haystack = haystack->getNext()) {
         if (needle == haystack) {
@@ -406,17 +388,6 @@ Connection * list_remove(Connection *haystack, Connection *needle) {
     return haystack;
 }
 
-static void enlist_conn(Connection *c, Connection **list) {
-    LIBEVENT_THREAD *thr = c->getThread();
-    cb_assert(list == &thr->pending_io);
-    cb_assert(!list_contains(thr->pending_io, c));
-    cb_assert(c->getNext() == nullptr);
-    c->setNext(*list);
-    *list = c;
-    cb_assert(list_contains(*list, c));
-    cb_assert(!has_cycle(*list));
-}
-
 void notify_io_complete(const void *void_cookie, ENGINE_ERROR_CODE status)
 {
     if (void_cookie == nullptr) {
@@ -445,10 +416,7 @@ void notify_io_complete(const void *void_cookie, ENGINE_ERROR_CODE status)
     LOG_DEBUG(NULL, "Got notify from %u, status 0x%x",
               connection->getId(), status);
 
-    LOCK_THREAD(thr);
-    reinterpret_cast<McbpConnection*>(connection)->setAiostat(status);
-    notify = add_conn_to_pending_io_list(connection);
-    UNLOCK_THREAD(thr);
+    notify = add_conn_to_pending_io_list(connection, status);
 
     /* kick the thread in the butt */
     if (notify) {
@@ -578,6 +546,8 @@ void threads_cleanup(void)
         subdoc_op_free(threads[ii].subdoc_op);
         delete threads[ii].validator;
         delete threads[ii].new_conn_queue;
+        delete threads[ii].pending_io.mutex;
+        delete threads[ii].pending_io.map;
     }
 
     cb_free(thread_ids);
@@ -620,15 +590,13 @@ void notify_thread(LIBEVENT_THREAD *thread) {
     }
 }
 
-int add_conn_to_pending_io_list(Connection *c) {
-    int notify = 0;
+int add_conn_to_pending_io_list(Connection* c, ENGINE_ERROR_CODE status) {
     auto thread = c->getThread();
-    if (number_of_pending(c, thread->pending_io) == 0) {
-        if (thread->pending_io == NULL) {
-            notify = 1;
-        }
-        enlist_conn(c, &thread->pending_io);
+    std::pair<LIBEVENT_THREAD::PendingIoMap::iterator, bool> result;
+    {
+        std::lock_guard<std::mutex> lock(*thread->pending_io.mutex);
+        result = thread->pending_io.map->emplace(c, status);
     }
-
+    int notify = result.second ? 1 : 0;
     return notify;
 }