log: broadcast cond signals

We were using a single cond, and only signalling one waiter.  That means
that if the flusher and several logging threads are waiting, and we hit
a limit, we the logger could signal another logger instead of the flusher,
and we could deadlock.

Similarly, if the flusher empties the queue, it might signal only a single
logger, and that logger could re-signal the flusher, and the other logger
could wait forever.

Intead, break the single cond into two: one for loggers, and one for the
flusher.  Always signal the (one) flusher, and always broadcast to all
loggers.

Backport: bobtail, argonaut
Signed-off-by: Sage Weil <sage@inktank.com>
Reviewed-by: Dan Mick <dan.mick@inktank.com>
(cherry picked from commit 813787af3dbb99e42f481af670c4bb0e254e4432)

2 files changed, 14 insertions, 8 deletions

diff --git a/src/log/Log.cc b/src/log/Log.cc
index 55c831d66fd..57bfc10a9d7 100644
--- a/src/log/Log.cc
+++ b/src/log/Log.cc
@@ -51,7 +51,10 @@ Log::Log(SubsystemMap *s)
   ret = pthread_mutex_init(&m_queue_mutex, NULL);
   assert(ret == 0);
 
-  ret = pthread_cond_init(&m_cond, NULL);
+  ret = pthread_cond_init(&m_cond_loggers, NULL);
+  assert(ret == 0);
+
+  ret = pthread_cond_init(&m_cond_flusher, NULL);
   assert(ret == 0);
 
   // kludge for prealloc testing
@@ -73,7 +76,8 @@ Log::~Log()
   pthread_spin_destroy(&m_lock);
   pthread_mutex_destroy(&m_queue_mutex);
   pthread_mutex_destroy(&m_flush_mutex);
-  pthread_cond_destroy(&m_cond);
+  pthread_cond_destroy(&m_cond_loggers);
+  pthread_cond_destroy(&m_cond_flusher);
 }
 
 
@@ -139,10 +143,10 @@ void Log::submit_entry(Entry *e)
 
   // wait for flush to catch up
   while (m_new.m_len > m_max_new)
-    pthread_cond_wait(&m_cond, &m_queue_mutex);
+    pthread_cond_wait(&m_cond_loggers, &m_queue_mutex);
 
   m_new.enqueue(e);
-  pthread_cond_signal(&m_cond);
+  pthread_cond_signal(&m_cond_flusher);
   pthread_mutex_unlock(&m_queue_mutex);
 }
 
@@ -169,7 +173,7 @@ void Log::flush()
   pthread_mutex_lock(&m_queue_mutex);
   EntryQueue t;
   t.swap(m_new);
-  pthread_cond_signal(&m_cond);
+  pthread_cond_broadcast(&m_cond_loggers);
   pthread_mutex_unlock(&m_queue_mutex);
   _flush(&t, &m_recent, false);
 
@@ -278,7 +282,8 @@ void Log::stop()
   assert(is_started());
   pthread_mutex_lock(&m_queue_mutex);
   m_stop = true;
-  pthread_cond_signal(&m_cond);
+  pthread_cond_signal(&m_cond_flusher);
+  pthread_cond_broadcast(&m_cond_loggers);
   pthread_mutex_unlock(&m_queue_mutex);
   join();
 }
@@ -294,7 +299,7 @@ void *Log::entry()
       continue;
     }
 
-    pthread_cond_wait(&m_cond, &m_queue_mutex);
+    pthread_cond_wait(&m_cond_flusher, &m_queue_mutex);
   }
   pthread_mutex_unlock(&m_queue_mutex);
   flush();
diff --git a/src/log/Log.h b/src/log/Log.h
index ab83a7c86e4..f6a27dc5b37 100644
--- a/src/log/Log.h
+++ b/src/log/Log.h
@@ -24,7 +24,8 @@ class Log : private Thread
   pthread_spinlock_t m_lock;
   pthread_mutex_t m_queue_mutex;
   pthread_mutex_t m_flush_mutex;
-  pthread_cond_t m_cond;
+  pthread_cond_t m_cond_loggers;
+  pthread_cond_t m_cond_flusher;
 
   EntryQueue m_new;    ///< new entries
   EntryQueue m_recent; ///< recent (less new) entries we've already written at low detail