MB-32687: Backport signalIfIdle refactor

signalIfIdle used to update the event mask in libevent to add
a write event to make the socket available. Libevent would then
pick up that socket next time it went down to the OS to do a poll
on the sockets. The problem is that when we try to update the
event mask for a socket libevent will try to acquire the mutex
used by the event base in order to do that. If the code was
called from another worker thread it would hold it's own lock
on the event base (and we're subject for a deadlock).

The new implementation of signalIfIdle push the connection
onto a list of connections to signal for the worker thread,
before it notifies the worker thread by writing to the
notification pipe for the worker thread. The callback for
the worker thread swaps out the list of connections, and
runs the state machine for each of the connections in there.

Change-Id: Ic0ba5de83b8f924a8b7457c65d0e91fbf7927f7d
Reviewed-on: http://review.couchbase.org/105714
Tested-by: Build Bot <[email protected]>
Well-Formed: Build Bot <[email protected]>
Reviewed-by: Jim Walker <[email protected]>

Author: trondn

daemon/connection.cc

@@ -1467,25 +1467,10 @@ void Connection::propagateDisconnect() const {
 
 void Connection::signalIfIdle(bool logbusy, size_t workerthread) {
     if (!isEwouldblock() && stateMachine.isIdleState()) {
-        // Raise a 'fake' write event to ensure the connection has an
-        // event delivered (for example if its sendQ is full).
-        if (!registered_in_libevent) {
-            ev_flags = EV_READ | EV_WRITE | EV_PERSIST;
-            if (!registerEvent()) {
-                LOG_WARNING(
-                        "{}: Connection::signalIfIdle: Unable to "
-                        "registerEvent.  Setting state to conn_closing",
-                        getId());
-                setState(McbpStateMachine::State::closing);
-            }
-        } else if (!updateEvent(EV_READ | EV_WRITE | EV_PERSIST)) {
-            LOG_WARNING(
-                    "{}: Connection::signalIfIdle: Unable to "
-                    "updateEvent.  Setting state to conn_closing",
-                    getId());
-            setState(McbpStateMachine::State::closing);
-        }
-        event_active(&event, EV_WRITE, 0);
+        auto* thr = getThread();
+        LOG_WARNING("Add notificaiton of {}", uint64_t(this));
+        thr->notification.push(this);
+        notify_thread(*thr);
     } else if (logbusy) {
         unique_cJSON_ptr json(toJSON());
         auto details = to_string(json, false);

daemon/connections.cc

@@ -223,6 +223,8 @@ void conn_close(Connection& connection) {
     if (thread == nullptr) {
         throw std::logic_error("conn_close: unable to obtain non-NULL thread from connection");
     }
+
+    thread->notification.remove(&connection);
     // remove from pending-io list
     {
         std::lock_guard<std::mutex> lock(thread->pending_io.mutex);

daemon/memcached.cc

@@ -907,6 +907,9 @@ void event_handler(evutil_socket_t fd, short which, void *arg) {
         thr->pending_io.map.erase(c);
     }
 
+    // Remove the connection from the notification list if it's there
+    thr->notification.remove(c);
+
     TRACE_LOCKGUARD_TIMED(thr->mutex,
                           "mutex",
                           "event_handler::threadLock",

daemon/memcached.h

@@ -132,6 +132,18 @@ struct LIBEVENT_THREAD {
         PendingIoMap map;
     } pending_io;
 
+    /// A list of connections to signal if they're idle
+    class NotificationList {
+    public:
+        void push(Connection* c);
+        void remove(Connection* c);
+        void swap(std::vector<Connection*>& other);
+
+    protected:
+        std::mutex mutex;
+        std::vector<Connection*> connections;
+    } notification;
+
     /// index of this thread in the threads array
     size_t index = 0;
 

daemon/thread.cc

@@ -65,6 +65,32 @@ void ConnectionQueue::push(std::unique_ptr<ConnectionQueueItem> item) {
 
 static LIBEVENT_THREAD dispatcher_thread;
 
+void LIBEVENT_THREAD::NotificationList::push(Connection* c) {
+    std::lock_guard<std::mutex> lock(mutex);
+    auto iter = std::find(connections.begin(), connections.end(), c);
+    if (iter == connections.end()) {
+        try {
+            connections.push_back(c);
+        } catch (const std::bad_alloc&) {
+            // Just ignore and hopefully we'll be able to signal it at a later
+            // time.
+        }
+    }
+}
+
+void LIBEVENT_THREAD::NotificationList::remove(Connection* c) {
+    std::lock_guard<std::mutex> lock(mutex);
+    auto iter = std::find(connections.begin(), connections.end(), c);
+    if (iter != connections.end()) {
+        connections.erase(iter);
+    }
+}
+
+void LIBEVENT_THREAD::NotificationList::swap(std::vector<Connection*>& other) {
+    std::lock_guard<std::mutex> lock(mutex);
+    connections.swap(other);
+}
+
 /*
  * Each libevent instance has a wakeup pipe, which other threads
  * can use to signal that they've put a new connection on its queue.
@@ -282,9 +308,22 @@ static void thread_libevent_process(evutil_socket_t fd, short which, void *arg)
                           "thread_libevent_process::threadLock",
                           SlowMutexThreshold);
 
+    std::vector<Connection*> notify;
+    me.notification.swap(notify);
+
     for (auto io : pending) {
         auto* c = io.first;
         auto status = io.second;
+
+        // Remove from the notify list if it's there as we don't
+        // want to run them twice
+        {
+            auto iter = std::find(notify.begin(), notify.end(), c);
+            if (iter != notify.end()) {
+                notify.erase(iter);
+            }
+        }
+
         if (c->getSocketDescriptor() != INVALID_SOCKET &&
             !c->isRegisteredInLibevent()) {
             /* The socket may have been shut down while we're looping */
@@ -302,6 +341,12 @@ static void thread_libevent_process(evutil_socket_t fd, short which, void *arg)
         run_event_loop(c, EV_READ | EV_WRITE);
     }
 
+    // Notify the connections we haven't notified yet
+    for (auto c : notify) {
+        c->setNumEvents(1);
+        run_event_loop(c, EV_READ | EV_WRITE);
+    }
+
     /*
      * I could look at all of the connection objects bound to dying buckets
      */