Allow the asynchronous logging system to safely delete its storage.

1 files changed, 56 insertions, 31 deletions

diff --git a/src/cubeb_log.cpp b/src/cubeb_log.cpp
index fa48f90..7b71150 100644
--- a/src/cubeb_log.cpp
+++ b/src/cubeb_log.cpp
@@ -25,7 +25,12 @@ const size_t CUBEB_LOG_MESSAGE_MAX_SIZE = 256;
  * messages. */
 const size_t CUBEB_LOG_MESSAGE_QUEUE_DEPTH = 40;
 /** Number of milliseconds to wait before dequeuing log messages. */
-#define CUBEB_LOG_BATCH_PRINT_INTERVAL_MS 10
+const size_t CUBEB_LOG_BATCH_PRINT_INTERVAL_MS = 10;
+
+void
+cubeb_noop_log_callback(char const * /* fmt */, ...)
+{
+}
 
 /**
  * This wraps an inline buffer, that represents a log message, that must be
@@ -65,17 +70,25 @@ public:
   void push(char const str[CUBEB_LOG_MESSAGE_MAX_SIZE])
   {
     cubeb_log_message msg(str);
-    msg_queue->enqueue(msg);
+    auto owned_queue = msg_queue.load();
+    // Check if the queue is being deallocated. If not, grab ownership. If yes,
+    // return, the message won't be logged.
+    if (!owned_queue ||
+        !msg_queue.compare_exchange_strong(owned_queue, nullptr)) {
+      return;
+    }
+    owned_queue->enqueue(msg);
+    // Return ownership.
+    msg_queue.store(owned_queue);
   }
   void run()
   {
     assert(logging_thread.get_id() == std::thread::id());
-    assert(msg_queue);
     logging_thread = std::thread([this]() {
       CUBEB_REGISTER_THREAD("cubeb_log");
       while (!shutdown_thread) {
         cubeb_log_message msg;
-        while (msg_queue->dequeue(&msg, 1)) {
+        while (msg_queue_consumer.load()->dequeue(&msg, 1)) {
           cubeb_log_internal_no_format(msg.get());
         }
         std::this_thread::sleep_for(
@@ -86,44 +99,62 @@ public:
   }
   // Tell the underlying queue the producer thread has changed, so it does not
   // assert in debug. This should be called with the thread stopped.
-  void reset_producer_thread() { msg_queue->reset_thread_ids(); }
+  void reset_producer_thread()
+  {
+    if (msg_queue) {
+      msg_queue.load()->reset_thread_ids();
+    }
+  }
   void start()
   {
-    msg_queue.reset(
-        new lock_free_queue<cubeb_log_message>(CUBEB_LOG_MESSAGE_QUEUE_DEPTH));
+    auto * queue =
+        new lock_free_queue<cubeb_log_message>(CUBEB_LOG_MESSAGE_QUEUE_DEPTH);
+    msg_queue.store(queue);
+    msg_queue_consumer.store(queue);
     shutdown_thread = false;
     run();
   }
   void stop()
   {
+    assert(((g_cubeb_log_callback == cubeb_noop_log_callback) ||
+            !g_cubeb_log_callback) &&
+           "Only call stop after logging has been disabled.");
     shutdown_thread = true;
     if (logging_thread.get_id() != std::thread::id()) {
       logging_thread.join();
       logging_thread = std::thread();
-      // This is OK, because at this point, we know the consumer has stopped
-      // consuming.
-      msg_queue->reset_thread_ids();
-      purge_queue();
-      msg_queue.reset(nullptr);
+      auto owned_queue = msg_queue.load();
+      // Check if the queue is being used. If not, grab ownership. If yes,
+      // try again shortly. At this point, the logging thread has been joined,
+      // so nothing is going to dequeue.
+      // If there is a valid pointer here, then the real-time audio thread that
+      // logs won't attempt to write into the queue, and instead drop the
+      // message.
+      while (!msg_queue.compare_exchange_weak(owned_queue, nullptr)) {
+      }
+      delete owned_queue;
+      msg_queue_consumer.store(nullptr);
     }
   }
-  void purge_queue()
+
+private:
+  cubeb_async_logger() {}
+  ~cubeb_async_logger()
   {
     assert(logging_thread.get_id() == std::thread::id() &&
-           "Only purge the async logger queue when the thread is stopped");
-    if (!msg_queue) {
-      return;
-    }
-    cubeb_log_message msg;
-    while (msg_queue->dequeue(&msg, 1)) { /* nothing */
+           (g_cubeb_log_callback == cubeb_noop_log_callback ||
+            !g_cubeb_log_callback));
+    if (msg_queue.load()) {
+      delete msg_queue.load();
     }
   }
-
-private:
-  cubeb_async_logger() {}
   /** This is quite a big data structure, but is only instantiated if the
-   * asynchronous logger is used.*/
-  std::unique_ptr<lock_free_queue<cubeb_log_message>> msg_queue;
+   * asynchronous logger is used. The two pointers point to the same object, but
+   * the first one can be temporarily null when a message is being enqueued. */
+  std::atomic<lock_free_queue<cubeb_log_message> *> msg_queue = {nullptr};
+
+  std::atomic<lock_free_queue<cubeb_log_message> *> msg_queue_consumer = {
+      nullptr};
   std::atomic<bool> shutdown_thread = {false};
   std::thread logging_thread;
 };
@@ -169,11 +200,6 @@ cubeb_async_log_reset_threads(void)
 }
 
 void
-cubeb_noop_log_callback(char const * /* fmt */, ...)
-{
-}
-
-void
 cubeb_log_set(cubeb_log_level log_level, cubeb_log_callback log_callback)
 {
   g_cubeb_log_level = log_level;
@@ -183,10 +209,9 @@ cubeb_log_set(cubeb_log_level log_level, cubeb_log_callback log_callback)
     g_cubeb_log_callback = log_callback;
     cubeb_async_logger::get().start();
   } else if (!log_callback || CUBEB_LOG_DISABLED) {
+    g_cubeb_log_callback = cubeb_noop_log_callback;
     // This returns once the thread has joined.
     cubeb_async_logger::get().stop();
-    g_cubeb_log_callback = cubeb_noop_log_callback;
-    cubeb_async_logger::get().purge_queue();
   } else {
     assert(false && "Incorrect parameters passed to cubeb_log_set");
   }