MB-31402: Back off logging of stuck connections

Change logging of stuck connections during deletion to:

   1. Don't log anything the first 30 seconds. Then
      dump the state of all stuck connections.
   2. Don't og anything for the next 30 seconds. Then
      dump the state of all stuck connections which
      changed since the previous dump.
   3. goto 2.

Change-Id: Ia0520ae28bb0f6b799f52c9e8902607c95d9485b
Reviewed-on: http://review.couchbase.org/103254
Tested-by: Build Bot <[email protected]>
Reviewed-by: Dave Rigby <[email protected]>

Author: trondn

daemon/connections.cc

@@ -64,26 +64,19 @@ static Connection* allocate_connection(SOCKET sfd,
 static void release_connection(Connection* c);
 
 /** External functions *******************************************************/
-int signal_idle_clients(FrontEndThread* me, int bucket_idx, bool logging) {
-    // We've got a situation right now where we're seeing that
-    // some of the connections is "stuck". Let's dump all
-    // information until we solve the bug.
-    logging = true;
 
+int signal_idle_clients(FrontEndThread& me) {
     int connected = 0;
-    std::lock_guard<std::mutex> lock(connections.mutex);
-    for (auto* c : connections.conns) {
-        if (c->getThread() == me) {
-            ++connected;
-            if (bucket_idx == -1 || c->getBucketIndex() == bucket_idx) {
-                if (!c->signalIfIdle() && logging) {
-                    auto details = c->toJSON().dump();
-                    LOG_INFO("Worker thread {}: {}", me->index, details);
+    const auto index = me.index;
+
+    iterate_thread_connections(
+            &me, [index, &connected](Connection& connection) {
+                ++connected;
+                if (!connection.signalIfIdle()) {
+                    auto details = connection.toJSON().dump();
+                    LOG_INFO("Worker thread {}: {}", index, details);
                 }
-            }
-        }
-    }
-
+            });
     return connected;
 }
 

daemon/connections.h

@@ -27,6 +27,7 @@
 
 struct FrontEndThread;
 class ListeningPort;
+class Bucket;
 
 /* Destroy all connections and reset connection management */
 void destroy_connections();
@@ -82,17 +83,15 @@ Connection* conn_new(const SOCKET sfd,
 ListeningPort *get_listening_port_instance(const in_port_t port);
 
 /**
- * Signal (set writable) all idle clients bound to either a specific
- * bucket specified by its index, or any bucket (specified as -1).
- * Due to the threading model we're only going to look at the clients
- * connected to the thread represented by me.
+ * Signal all of the idle clients in the system.
  *
- * @param me the thread to inspect
- * @param bucket_idx the bucket we'd like to signal (set to -1 to signal
- *                   all buckets)
- * @return the number of client connections bound to this thread.
+ * Due to the locking model we need to have the FrontEndThread locked
+ * when calling the method.
+ *
+ * @param me The connections to inspect must be bound to this thread
+ * @return The number of clients connections bound to this thread
  */
-int signal_idle_clients(FrontEndThread* me, int bucket_idx, bool logging);
+int signal_idle_clients(FrontEndThread& me);
 
 /**
  * Iterate over all of the connections and call the callback function

daemon/front_end_thread.h

@@ -124,17 +124,6 @@ struct FrontEndThread {
      */
     Subdoc::Operation subdoc_op;
 
-    /**
-     * When we're deleting buckets we need to disconnect idle
-     * clients. This variable is incremented for every delete bucket
-     * thread running and decremented when it's done. When this
-     * variable is set we'll try to look through all connections and
-     * update them with a write event if they're in an "idle"
-     * state. That should cause them to be rescheduled and cause the
-     * client to disconnect.
-     */
-    int deleting_buckets = 0;
-
     /**
      * Shared validator used by all connections serviced by this thread
      * when they need to validate a JSON document
@@ -147,4 +136,3 @@ struct FrontEndThread {
 
 void notify_thread(FrontEndThread& thread);
 void notify_dispatcher();
-void notify_thread_bucket_deletion(FrontEndThread& me);

daemon/memcached.cc

@@ -765,7 +765,7 @@ void event_handler(evutil_socket_t fd, short which, void *arg) {
 
     if (memcached_shutdown) {
         // Someone requested memcached to shut down.
-        if (signal_idle_clients(thr, -1, false) == 0) {
+        if (signal_idle_clients(*thr) == 0) {
             cb_assert(thr != nullptr);
             LOG_INFO("Stopping worker thread {}", thr->index);
             c->eventBaseLoopbreak();
@@ -781,7 +781,7 @@ void event_handler(evutil_socket_t fd, short which, void *arg) {
     if (memcached_shutdown) {
         // Someone requested memcached to shut down. If we don't have
         // any connections bound to this thread we can just shut down
-        int connected = signal_idle_clients(thr, -1, true);
+        int connected = signal_idle_clients(*thr);
         if (connected == 0) {
             LOG_INFO("Stopping worker thread {}", thr->index);
             event_base_loopbreak(thr->base);
@@ -1789,22 +1789,6 @@ void CreateBucketThread::run()
     task->makeRunnable();
 }
 
-void notify_thread_bucket_deletion(FrontEndThread& me) {
-    for (size_t ii = 0; ii < all_buckets.size(); ++ii) {
-        bool destroy = false;
-        {
-            std::lock_guard<std::mutex> guard(all_buckets[ii].mutex);
-            if (all_buckets[ii].state == BucketState::Destroying) {
-                destroy = true;
-            }
-        }
-
-        if (destroy) {
-            signal_idle_clients(&me, gsl::narrow<int>(ii), false);
-        }
-    }
-}
-
 void DestroyBucketThread::destroy() {
     ENGINE_ERROR_CODE ret = ENGINE_KEY_ENOENT;
     std::unique_lock<std::mutex> all_bucket_lock(buckets_lock);
@@ -1861,58 +1845,114 @@ void DestroyBucketThread::destroy() {
 
     perform_callbacks(ON_DELETE_BUCKET, nullptr, &all_buckets[idx]);
 
-    LOG_INFO("{} Delete bucket [{}]. Wait for clients to disconnect",
-             connection_id,
-             name);
-
     /* If this thread is connected to the requested bucket... release it */
     if (connection != nullptr && idx == size_t(connection->getBucketIndex())) {
         disassociate_bucket(*connection);
     }
 
-    /* Let all of the worker threads start invalidating connections */
-    threads_initiate_bucket_deletion();
-
+    // Wait until all users disconnected...
     auto& bucket = all_buckets[idx];
-
-    /* Wait until all users disconnected... */
     {
         std::unique_lock<std::mutex> guard(bucket.mutex);
+        if (bucket.clients > 0) {
+            LOG_INFO("{} Delete bucket [{}]. Wait for {} clients to disconnect",
+                     connection_id,
+                     name,
+                     bucket.clients);
 
-        while (bucket.clients > 0) {
-            LOG_INFO(
-                    "{} Delete bucket [{}]. Still waiting: {} clients "
-                    "connected",
-                    connection_id.c_str(),
-                    name.c_str(),
-                    bucket.clients);
-            /* drop the lock and notify the worker threads */
+            // Signal clients bound to the bucket before waiting
             guard.unlock();
-            threads_notify_bucket_deletion();
+            iterate_all_connections([&bucket](Connection& connection) {
+                if (&connection.getBucket() == &bucket) {
+                    connection.signalIfIdle();
+                }
+            });
             guard.lock();
-
-            bucket.cond.wait_for(guard,
-                                 std::chrono::milliseconds(1000),
-                                 [&bucket] { return bucket.clients == 0; });
         }
-    }
 
-    /* Tell the worker threads to stop trying to invalidating connections */
-    threads_complete_bucket_deletion();
+        using std::chrono::seconds;
+        using std::chrono::steady_clock;
 
-    /*
-     * We cannot call assert_no_assocations(idx) because it iterates
-     * over all connections and calls c->getBucketIndex().  The problem
-     * is that a worker thread can call associate_initial_bucket() or
-     * associate_bucket() at the same time.  This could lead to a call
-     * to c->setBucketIndex(0) (the "no bucket"), which although safe,
-     * raises a threadsanitizer warning.
-
-     * Note, if associate_bucket() attempts to associate a connection
-     * with a bucket that has been destroyed, or is in the process of
-     * being destroyed, the association will fail because
-     * BucketState != Ready.  See associate_bucket() for more details.
-     */
+        auto nextLog = steady_clock::now() + seconds(30);
+        nlohmann::json prevDump;
+
+        // We need to disconnect all of the clients before we can delete the
+        // bucket. We try to log stuff while we're waiting in order to
+        // know why stuff isn't complete. We'll do it in the following way:
+        //
+        //       1. Don't log anything the first 30 seconds. Then
+        //          dump the state of all stuck connections.
+        //       2. Don't og anything for the next 30 seconds. Then
+        //          dump the state of all stuck connections which
+        //          changed since the previous dump.
+        //       3. goto 2.
+        //
+        while (bucket.clients > 0) {
+            bucket.cond.wait_for(guard, seconds(1), [&bucket] {
+                return bucket.clients == 0;
+            });
+
+            if (bucket.clients == 0) {
+                break;
+            }
+
+            if (steady_clock::now() < nextLog) {
+                guard.unlock();
+                iterate_all_connections([&bucket](Connection& connection) {
+                    if (&connection.getBucket() == &bucket) {
+                        connection.signalIfIdle();
+                    }
+                });
+                guard.lock();
+                continue;
+            }
+
+            nextLog = steady_clock::now() + seconds(30);
+
+            // drop the lock and notify the worker threads
+            guard.unlock();
+
+            nlohmann::json json;
+            iterate_all_connections([&bucket, &json](Connection& connection) {
+                if (&connection.getBucket() == &bucket) {
+                    if (!connection.signalIfIdle()) {
+                        json[std::to_string(connection.getId())] =
+                                connection.toJSON();
+                    }
+                }
+            });
+
+            // remove all connections which didn't change
+            for (auto it = prevDump.begin(); it != prevDump.end(); ++it) {
+                auto entry = json.find(it.key());
+                if (entry != json.end()) {
+                    auto old = it.value().dump();
+                    auto current = entry->dump();
+                    if (old == current) {
+                        json.erase(entry);
+                    }
+                }
+            }
+
+            prevDump = std::move(json);
+            if (prevDump.empty()) {
+                LOG_INFO(
+                        R"({} Delete bucket [{}]. Still waiting: {} clients connected (state is unchanged).)",
+                        connection_id,
+                        name,
+                        bucket.clients);
+            } else {
+                LOG_INFO(
+                        R"({} Delete bucket [{}]. Still waiting: {} clients connected: {})",
+                        connection_id,
+                        name,
+                        bucket.clients,
+                        prevDump.dump());
+            }
+
+            guard.lock();
+        }
+    }
 
     LOG_INFO(
             "{} Delete bucket [{}]. Shut down the bucket", connection_id, name);
@@ -1923,7 +1963,7 @@ void DestroyBucketThread::destroy() {
              connection_id,
              name);
 
-    /* Clean up the stats... */
+    // Clean up the stats...
     threadlocal_stats_reset(bucket.stats);
 
     // Clear any registered event handlers

daemon/memcached.h

@@ -110,10 +110,6 @@ const char* get_server_version(void);
  */
 void update_topkeys(const Cookie& cookie);
 
-void threads_notify_bucket_deletion();
-void threads_complete_bucket_deletion();
-void threads_initiate_bucket_deletion();
-
 SERVER_HANDLE_V1* get_server_api();
 
 void shutdown_server();

daemon/thread.cc

@@ -278,7 +278,7 @@ static void thread_libevent_process(evutil_socket_t fd, short, void* arg) {
             return;
         }
 
-        if (signal_idle_clients(&me, -1, false) == 0) {
+        if (signal_idle_clients(me) == 0) {
             LOG_INFO("Stopping worker thread {}", me.index);
             event_base_loopbreak(me.base);
             return;
@@ -342,22 +342,14 @@ static void thread_libevent_process(evutil_socket_t fd, short, void* arg) {
         run_event_loop(c, EV_READ | EV_WRITE);
     }
 
-    /*
-     * I could look at all of the connection objects bound to dying buckets
-     */
-    if (me.deleting_buckets) {
-        notify_thread_bucket_deletion(me);
-    }
-
     if (memcached_shutdown) {
         // Someone requested memcached to shut down. If we don't have
         // any connections bound to this thread we can just shut down
-        auto connected = signal_idle_clients(&me, -1, true);
+        auto connected = signal_idle_clients(me);
         if (connected == 0) {
             LOG_INFO("Stopping worker thread {}", me.index);
             event_base_loopbreak(me.base);
         } else {
-            // @todo Change loglevel once MB-16255 is resolved
             LOG_INFO("Waiting for {} connected clients on worker thread {}",
                      connected,
                      me.index);
@@ -515,32 +507,6 @@ FrontEndThread::~FrontEndThread() {
     }
 }
 
-void threads_notify_bucket_deletion() {
-    for (auto& thr : threads) {
-        notify_thread(thr);
-    }
-}
-
-void threads_complete_bucket_deletion() {
-    for (auto& thr : threads) {
-        TRACE_LOCKGUARD_TIMED(thr.mutex,
-                              "mutex",
-                              "threads_complete_bucket_deletion::threadLock",
-                              SlowMutexThreshold);
-        thr.deleting_buckets--;
-    }
-}
-
-void threads_initiate_bucket_deletion() {
-    for (auto& thr : threads) {
-        TRACE_LOCKGUARD_TIMED(thr.mutex,
-                              "mutex",
-                              "threads_initiate_bucket_deletion::threadLock",
-                              SlowMutexThreshold);
-        thr.deleting_buckets++;
-    }
-}
-
 void notify_thread(FrontEndThread& thread) {
     if (cb::net::send(thread.notify[1], "", 1, 0) != 1 &&
         !cb::net::is_blocking(cb::net::get_socket_error())) {