Bug 1803765 - Update libcubeb to revision f4b583c. r=cubeb-reviewers,kinetik

Differential Revision: https://phabricator.services.mozilla.com/D163720

media/libcubeb/moz.yaml

@@ -9,8 +9,8 @@ origin:
   description: "Cross platform audio library"
   url: https://github.com/mozilla/cubeb
   license: ISC
-  release: 93d1fa3fccdc22da37aa59f67b213591797db369 (2022-09-22T21:52:54Z).
-  revision: 93d1fa3fccdc22da37aa59f67b213591797db369
+  release: f4b583c04591ec4d9781bef2f4e034fcb9e2dce7 (2022-12-12T15:45:21Z).
+  revision: f4b583c04591ec4d9781bef2f4e034fcb9e2dce7
 
 vendoring:
   url: https://github.com/mozilla/cubeb

media/libcubeb/src/cubeb.c

@@ -351,6 +351,8 @@ cubeb_destroy(cubeb * context)
   }
 
   context->ops->destroy(context);
+
+  cubeb_set_log_callback(CUBEB_LOG_DISABLED, NULL);
 }
 
 int

media/libcubeb/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,47 +70,93 @@ 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()
   {
-    std::thread([this]() {
+    assert(logging_thread.get_id() == std::thread::id());
+    logging_thread = std::thread([this]() {
       CUBEB_REGISTER_THREAD("cubeb_log");
-      while (true) {
+      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());
         }
-#ifdef _WIN32
-        Sleep(CUBEB_LOG_BATCH_PRINT_INTERVAL_MS);
-#else
-        timespec sleep_duration = sleep_for;
-        timespec remainder;
-        do {
-          if (nanosleep(&sleep_duration, &remainder) == 0 || errno != EINTR) {
-            break;
-          }
-          sleep_duration = remainder;
-        } while (remainder.tv_sec || remainder.tv_nsec);
-#endif
+        std::this_thread::sleep_for(
+            std::chrono::milliseconds(CUBEB_LOG_BATCH_PRINT_INTERVAL_MS));
       }
       CUBEB_UNREGISTER_THREAD();
-    }).detach();
+    });
   }
   // 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()
+  {
+    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();
+      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);
+    }
+  }
 
 private:
-#ifndef _WIN32
-  const struct timespec sleep_for = {
-      CUBEB_LOG_BATCH_PRINT_INTERVAL_MS / 1000,
-      (CUBEB_LOG_BATCH_PRINT_INTERVAL_MS % 1000) * 1000 * 1000};
-#endif
-  cubeb_async_logger() : msg_queue(CUBEB_LOG_MESSAGE_QUEUE_DEPTH) { run(); }
+  cubeb_async_logger() {}
+  ~cubeb_async_logger()
+  {
+    assert(logging_thread.get_id() == std::thread::id() &&
+           (g_cubeb_log_callback == cubeb_noop_log_callback ||
+            !g_cubeb_log_callback));
+    if (msg_queue.load()) {
+      delete msg_queue.load();
+    }
+  }
   /** This is quite a big data structure, but is only instantiated if the
-   * asynchronous logger is used.*/
-  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;
 };
 
 void
@@ -115,8 +166,8 @@ cubeb_log_internal(char const * file, uint32_t line, char const * fmt, ...)
   va_start(args, fmt);
   char msg[CUBEB_LOG_MESSAGE_MAX_SIZE];
   vsnprintf(msg, CUBEB_LOG_MESSAGE_MAX_SIZE, fmt, args);
-  g_cubeb_log_callback.load()("%s:%d:%s", file, line, msg);
   va_end(args);
+  g_cubeb_log_callback.load()("%s:%d:%s", file, line, msg);
 }
 
 void
@@ -152,7 +203,18 @@ void
 cubeb_log_set(cubeb_log_level log_level, cubeb_log_callback log_callback)
 {
   g_cubeb_log_level = log_level;
-  g_cubeb_log_callback = log_callback;
+  // Once a callback has a been set, `g_cubeb_log_callback` is never set back to
+  // nullptr, to prevent a TOCTOU race between checking the pointer
+  if (log_callback && log_level != CUBEB_LOG_DISABLED) {
+    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();
+  } else {
+    assert(false && "Incorrect parameters passed to cubeb_log_set");
+  }
 }
 
 cubeb_log_level
@@ -164,5 +226,8 @@ cubeb_log_get_level()
 cubeb_log_callback
 cubeb_log_get_callback()
 {
+  if (g_cubeb_log_callback == cubeb_noop_log_callback) {
+    return nullptr;
+  }
   return g_cubeb_log_callback;
 }

media/libcubeb/src/cubeb_log.h

@@ -33,9 +33,9 @@ extern "C" {
 void
 cubeb_log_set(cubeb_log_level log_level, cubeb_log_callback log_callback);
 cubeb_log_level
-cubeb_log_get_level();
+cubeb_log_get_level(void);
 cubeb_log_callback
-cubeb_log_get_callback();
+cubeb_log_get_callback(void);
 void
 cubeb_log_internal_no_format(const char * msg);
 void
@@ -54,14 +54,14 @@ cubeb_async_log_reset_threads(void);
 
 #define LOG_INTERNAL(level, fmt, ...)                                          \
   do {                                                                         \
-    if (cubeb_log_get_level() <= level && cubeb_log_get_callback()) {          \
+    if (cubeb_log_get_level() >= level && cubeb_log_get_callback()) {          \
       cubeb_log_internal(__FILENAME__, __LINE__, fmt, ##__VA_ARGS__);          \
     }                                                                          \
   } while (0)
 
 #define ALOG_INTERNAL(level, fmt, ...)                                         \
   do {                                                                         \
-    if (cubeb_log_get_level() <= level && cubeb_log_get_callback()) {          \
+    if (cubeb_log_get_level() >= level && cubeb_log_get_callback()) {          \
       cubeb_async_log(fmt, ##__VA_ARGS__);                                     \
     }                                                                          \
   } while (0)

media/libcubeb/src/cubeb_wasapi.cpp

@@ -21,6 +21,7 @@
 #include <stdint.h>
 #include <stdio.h>
 #include <stdlib.h>
+#include <timeapi.h>
 #include <vector>
 #include <windef.h>
 #include <windows.h>
@@ -97,6 +98,8 @@ namespace {
 
 const int64_t LATENCY_NOT_AVAILABLE_YET = -1;
 
+const DWORD DEVICE_CHANGE_DEBOUNCE_MS = 250;
+
 struct com_heap_ptr_deleter {
   void operator()(void * ptr) const noexcept { CoTaskMemFree(ptr); }
 };
@@ -696,7 +699,8 @@ public:
   }
 
   wasapi_endpoint_notification_client(HANDLE event, ERole role)
-      : ref_count(1), reconfigure_event(event), role(role)
+      : ref_count(1), reconfigure_event(event), role(role),
+        last_device_change(timeGetTime())
   {
   }
 
@@ -705,17 +709,32 @@ public:
   HRESULT STDMETHODCALLTYPE OnDefaultDeviceChanged(EDataFlow flow, ERole role,
                                                    LPCWSTR device_id)
   {
-    LOG("endpoint: Audio device default changed.");
+    LOG("endpoint: Audio device default changed flow=%d role=%d "
+        "new_device_id=%ws.",
+        flow, role, device_id);
 
     /* we only support a single stream type for now. */
-    if (flow != eRender && role != this->role) {
+    if (flow != eRender || role != this->role) {
       return S_OK;
     }
 
-    BOOL ok = SetEvent(reconfigure_event);
-    if (!ok) {
-      LOG("endpoint: SetEvent on reconfigure_event failed: %lx",
-          GetLastError());
+    DWORD last_change_ms = timeGetTime() - last_device_change;
+    bool same_device = default_device_id && device_id &&
+                       wcscmp(default_device_id.get(), device_id) == 0;
+    LOG("endpoint: Audio device default changed last_change=%u same_device=%d",
+        last_change_ms, same_device);
+    if (last_change_ms > DEVICE_CHANGE_DEBOUNCE_MS || !same_device) {
+      if (device_id) {
+        default_device_id.reset(_wcsdup(device_id));
+      } else {
+        default_device_id.reset();
+      }
+      BOOL ok = SetEvent(reconfigure_event);
+      LOG("endpoint: Audio device default changed: trigger reconfig");
+      if (!ok) {
+        LOG("endpoint: SetEvent on reconfigure_event failed: %lx",
+            GetLastError());
+      }
     }
 
     return S_OK;
@@ -754,6 +773,8 @@ private:
   LONG ref_count;
   HANDLE reconfigure_event;
   ERole role;
+  std::unique_ptr<const wchar_t[]> default_device_id;
+  DWORD last_device_change;
 };
 
 namespace {
@@ -2802,7 +2823,7 @@ wasapi_stream_init(cubeb * context, cubeb_stream ** stream,
 
   *stream = stm.release();
 
-  LOG("Stream init succesfull (%p)", *stream);
+  LOG("Stream init successful (%p)", *stream);
   return CUBEB_OK;
 }
 
@@ -2813,20 +2834,18 @@ close_wasapi_stream(cubeb_stream * stm)
 
   stm->stream_reset_lock.assert_current_thread_owns();
 
-  stm->output_client = nullptr;
-  stm->render_client = nullptr;
-
-  stm->input_client = nullptr;
-  stm->capture_client = nullptr;
-
-  stm->output_device = nullptr;
-  stm->input_device = nullptr;
-
 #ifdef CUBEB_WASAPI_USE_IAUDIOSTREAMVOLUME
   stm->audio_stream_volume = nullptr;
 #endif
-
   stm->audio_clock = nullptr;
+  stm->render_client = nullptr;
+  stm->output_client = nullptr;
+  stm->output_device = nullptr;
+
+  stm->capture_client = nullptr;
+  stm->input_client = nullptr;
+  stm->input_device = nullptr;
+
   stm->total_frames_written += static_cast<UINT64>(
       round(stm->frames_written *
             stream_to_mix_samplerate_ratio(stm->output_stream_params,

media/libcubeb/test/test_audio.cpp

@@ -26,11 +26,6 @@
 using namespace std;
 
 #define MAX_NUM_CHANNELS 32
-
-#if !defined(M_PI)
-#define M_PI 3.14159265358979323846
-#endif
-
 #define VOLUME 0.2
 
 float get_frequency(int channel_index)
@@ -242,3 +237,7 @@ TEST(cubeb, run_channel_rate_test)
     }
   }
 }
+
+
+#undef MAX_NUM_CHANNELS
+#undef VOLUME

media/libcubeb/test/test_deadlock.cpp

@@ -258,3 +258,5 @@ TEST(cubeb, run_deadlock_test)
 
   cubeb_stream_stop(stream);
 }
+
+#undef CALL_THREAD_KILLER

media/libcubeb/test/test_device_changed_callback.cpp

@@ -107,3 +107,10 @@ TEST(cubeb, device_changed_callbacks)
 
   cubeb_stream_destroy(stream);
 }
+
+#undef SAMPLE_FREQUENCY
+#undef STREAM_FORMAT
+#undef INPUT_CHANNELS
+#undef INPUT_LAYOUT
+#undef OUTPUT_CHANNELS
+#undef OUTPUT_LAYOUT

media/libcubeb/test/test_duplex.cpp

@@ -337,3 +337,10 @@ TEST(cubeb, one_duplex_one_input)
 
   ASSERT_FALSE(duplex_stream_state.invalid_audio_value.load());
 }
+
+#undef SAMPLE_FREQUENCY
+#undef STREAM_FORMAT
+#undef INPUT_CHANNELS
+#undef INPUT_LAYOUT
+#undef OUTPUT_CHANNELS
+#undef OUTPUT_LAYOUT

media/libcubeb/test/test_logging.cpp

@@ -0,0 +1,193 @@
+/*
+ * Copyright © 2016 Mozilla Foundation
+ *
+ * This program is made available under an ISC-style license.  See the
+ * accompanying file LICENSE for details.
+ */
+
+/* cubeb_logging test  */
+#include "gtest/gtest.h"
+#if !defined(_XOPEN_SOURCE)
+#define _XOPEN_SOURCE 600
+#endif
+#include "cubeb/cubeb.h"
+#include "cubeb_log.h"
+#include <atomic>
+#include <math.h>
+#include <memory>
+#include <stdio.h>
+#include <stdlib.h>
+#include <thread>
+
+#include "common.h"
+
+#define PRINT_LOGS_TO_STDERR 0
+
+std::atomic<uint32_t> log_statements_received = {0};
+std::atomic<uint32_t> data_callback_call_count = {0};
+
+void
+test_logging_callback(char const * fmt, ...)
+{
+  log_statements_received++;
+#if PRINT_LOGS_TO_STDERR == 1
+  char buf[1024];
+  va_list argslist;
+  va_start(argslist, fmt);
+  vsnprintf(buf, 1024, fmt, argslist);
+  fprintf(stderr, "%s\n", buf);
+  va_end(argslist);
+#endif // PRINT_LOGS_TO_STDERR
+}
+
+long
+data_cb_load(cubeb_stream * stream, void * user, const void * inputbuffer,
+             void * outputbuffer, long nframes)
+{
+  data_callback_call_count++;
+  return nframes;
+}
+
+void
+state_cb(cubeb_stream * stream, void * /*user*/, cubeb_state state)
+{
+  if (stream == NULL)
+    return;
+
+  switch (state) {
+  case CUBEB_STATE_STARTED:
+    fprintf(stderr, "stream started\n");
+    break;
+  case CUBEB_STATE_STOPPED:
+    fprintf(stderr, "stream stopped\n");
+    break;
+  case CUBEB_STATE_DRAINED:
+    fprintf(stderr, "stream drained\n");
+    break;
+  default:
+    fprintf(stderr, "unknown stream state %d\n", state);
+  }
+
+  return;
+}
+
+// Waits for at least one audio callback to have occured.
+void
+wait_for_audio_callback()
+{
+  uint32_t audio_callback_index =
+      data_callback_call_count.load(std::memory_order_acquire);
+  while (audio_callback_index ==
+         data_callback_call_count.load(std::memory_order_acquire)) {
+    delay(100);
+  }
+}
+
+TEST(cubeb, logging)
+{
+  cubeb * ctx;
+  cubeb_stream * stream;
+  cubeb_stream_params output_params;
+  int r;
+  uint32_t latency_frames = 0;
+
+  cubeb_set_log_callback(CUBEB_LOG_NORMAL, test_logging_callback);
+
+  r = common_init(&ctx, "Cubeb logging test");
+  ASSERT_EQ(r, CUBEB_OK) << "Error initializing cubeb library";
+
+  std::unique_ptr<cubeb, decltype(&cubeb_destroy)> cleanup_cubeb_at_exit(
+      ctx, cubeb_destroy);
+
+  output_params.format = CUBEB_SAMPLE_FLOAT32LE;
+  output_params.rate = 48000;
+  output_params.channels = 2;
+  output_params.layout = CUBEB_LAYOUT_STEREO;
+  output_params.prefs = CUBEB_STREAM_PREF_NONE;
+
+  r = cubeb_get_min_latency(ctx, &output_params, &latency_frames);
+  ASSERT_EQ(r, CUBEB_OK) << "Could not get minimal latency";
+
+  r = cubeb_stream_init(ctx, &stream, "Cubeb logging", NULL, NULL, NULL,
+                        &output_params, latency_frames, data_cb_load, state_cb,
+                        NULL);
+  ASSERT_EQ(r, CUBEB_OK) << "Error initializing cubeb stream";
+
+  std::unique_ptr<cubeb_stream, decltype(&cubeb_stream_destroy)>
+      cleanup_stream_at_exit(stream, cubeb_stream_destroy);
+
+  ASSERT_NE(log_statements_received.load(std::memory_order_acquire), 0u);
+
+  cubeb_set_log_callback(CUBEB_LOG_DISABLED, nullptr);
+  log_statements_received.store(0, std::memory_order_release);
+
+  // This is synchronous and we'll receive log messages on all backends that we
+  // test
+  cubeb_stream_start(stream);
+
+  ASSERT_EQ(log_statements_received.load(std::memory_order_acquire), 0u);
+
+  cubeb_set_log_callback(CUBEB_LOG_VERBOSE, test_logging_callback);
+
+  wait_for_audio_callback();
+
+  ASSERT_NE(log_statements_received.load(std::memory_order_acquire), 0u);
+
+  bool log_callback_set = true;
+  uint32_t iterations = 100;
+  while (iterations--) {
+    wait_for_audio_callback();
+
+    if (!log_callback_set) {
+      ASSERT_EQ(log_statements_received.load(std::memory_order_acquire), 0u);
+      // Set a logging callback, start logging
+      cubeb_set_log_callback(CUBEB_LOG_VERBOSE, test_logging_callback);
+      log_callback_set = true;
+    } else {
+      // Disable the logging callback, stop logging.
+      ASSERT_NE(log_statements_received.load(std::memory_order_acquire), 0u);
+      cubeb_set_log_callback(CUBEB_LOG_DISABLED, nullptr);
+      log_statements_received.store(0, std::memory_order_release);
+      // Disabling logging should flush any log message -- wait a bit and check
+      // that this is true.
+      ASSERT_EQ(log_statements_received.load(std::memory_order_acquire), 0u);
+      log_callback_set = false;
+    }
+  }
+
+  cubeb_stream_stop(stream);
+}
+
+TEST(cubeb, logging_stress)
+{
+  cubeb_set_log_callback(CUBEB_LOG_NORMAL, test_logging_callback);
+
+  std::atomic<bool> thread_done = {false};
+
+  auto t = std::thread([&thread_done]() {
+    uint32_t count = 0;
+    do {
+      while (rand() % 10) {
+        ALOG("Log message #%d!", count++);
+      }
+    } while (count < 1e4);
+    thread_done.store(true);
+  });
+
+  bool enabled = true;
+  while (!thread_done.load()) {
+    if (enabled) {
+      cubeb_set_log_callback(CUBEB_LOG_DISABLED, nullptr);
+      enabled = false;
+    } else {
+      cubeb_set_log_callback(CUBEB_LOG_NORMAL, test_logging_callback);
+      enabled = true;
+    }
+  }
+
+  cubeb_set_log_callback(CUBEB_LOG_DISABLED, nullptr);
+
+  t.join();
+
+  ASSERT_TRUE(true);
+}

media/libcubeb/test/test_overload_callback.cpp

@@ -90,3 +90,6 @@ TEST(cubeb, overload_callback)
   delay(500);
   cubeb_stream_stop(stream);
 }
+
+#undef SAMPLE_FREQUENCY
+#undef STREAM_FORMAT

media/libcubeb/test/test_record.cpp

@@ -114,3 +114,6 @@ TEST(cubeb, record)
   ASSERT_FALSE(stream_state.invalid_audio_value.load());
 #endif
 }
+
+#undef SAMPLE_FREQUENCY
+#undef STREAM_FORMAT

media/libcubeb/test/test_resampler.cpp

@@ -53,7 +53,7 @@ const uint32_t max_chunks = 30; /* ms */
 const uint32_t chunk_increment = 10;
 #endif
 
-#define DUMP_ARRAYS
+// #define DUMP_ARRAYS
 #ifdef DUMP_ARRAYS
 /**
  * Files produced by dump(...) can be converted to .wave files using:
@@ -1084,3 +1084,6 @@ TEST(cubeb, individual_methods) {
   ASSERT_EQ(frames_needed2, 0u);
 }
 
+
+#undef NOMINMAX
+#undef DUMP_ARRAYS

media/libcubeb/test/test_ring_buffer.cpp

@@ -225,3 +225,5 @@ TEST(cubeb, ring_buffer)
 
   test_reset_api();
 }
+
+#undef NOMINMAX

media/libcubeb/test/test_sanity.cpp

@@ -690,3 +690,9 @@ TEST(cubeb, stable_devid)
   ASSERT_EQ(r, CUBEB_OK);
   cubeb_destroy(ctx);
 }
+
+#undef STREAM_RATE
+#undef STREAM_LATENCY
+#undef STREAM_CHANNELS
+#undef STREAM_LAYOUT
+#undef STREAM_FORMAT

media/libcubeb/test/test_tone.cpp

@@ -119,3 +119,6 @@ TEST(cubeb, tone)
 
   ASSERT_TRUE(user_data->position.load());
 }
+
+#undef SAMPLE_FREQUENCY
+#undef STREAM_FORMAT