Merge "Add a Reader-Writer Lock"

libwvdrmengine/Android.mk

@@ -96,6 +96,7 @@ CORE_SRC_DIR := cdm/core/src
 LOCAL_SRC_FILES := \
     $(CORE_SRC_DIR)/properties.cpp \
     $(UTIL_SRC_DIR)/platform.cpp \
+    $(UTIL_SRC_DIR)/rw_lock.cpp \
     $(UTIL_SRC_DIR)/string_conversions.cpp \
     $(UTIL_SRC_DIR)/clock.cpp \
     $(UTIL_SRC_DIR)/file_store.cpp \

libwvdrmengine/build_and_run_all_unit_tests.sh

@@ -109,6 +109,7 @@ try_adb_push oemcrypto_test
 try_adb_push policy_engine_constraints_unittest
 try_adb_push policy_engine_unittest
 try_adb_push request_license_test
+try_adb_push rw_lock_test
 try_adb_push service_certificate_unittest
 try_adb_push timer_unittest
 try_adb_push usage_table_header_unittest

libwvdrmengine/cdm/core/test/rw_lock_test.cpp

@@ -0,0 +1,293 @@
+// Copyright 2019 Google LLC. All Rights Reserved. This file and proprietary
+// source code may only be used and distributed under the Widevine Master
+// License Agreement.
+
+#include <atomic>
+#include <chrono>
+#include <condition_variable>
+#include <future>
+#include <mutex>
+#include <thread>
+
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+
+#include "rw_lock.h"
+
+using std::this_thread::sleep_for;
+
+namespace wvcdm {
+
+namespace {
+
+const size_t kFleetSize = 20;
+
+const auto kShortSnooze = std::chrono::milliseconds(10);
+// By convention, we assume any other threads put into flight will have
+// finished their minimal actions after 1 second.
+const auto kDefaultTimeout = std::chrono::seconds(1);
+
+struct ConditionVariablePair {
+  std::mutex mutex;
+  std::condition_variable condition_variable;
+};
+
+template <typename Duration>
+bool WaitForTrue(std::atomic<bool>* value, Duration timeout) {
+  std::chrono::steady_clock clock;
+  auto start = clock.now();
+  while (!*value) {
+    if (clock.now() - start >= timeout) return false;
+    sleep_for(kShortSnooze);
+  }
+  return true;
+}
+
+}  // namespace
+
+TEST(SharedMutexUnitTests, ReadersDoNotBlockReaders) {
+  shared_mutex mutex_under_test;
+
+  // Used to signal the primary reader thread to stop holding the lock.
+  ConditionVariablePair delay_primary;
+  // Used to indicate if the primary thread is waiting. (and thus holding the
+  // mutex)
+  std::atomic<bool> primary_is_waiting(false);
+
+  // Start a thread that takes the shared_mutex as a reader and holds it until
+  // signalled to stop.
+  std::future<void> primary_reader = std::async(std::launch::async, [&] {
+    shared_lock<shared_mutex> auto_lock(mutex_under_test);
+
+    ASSERT_FALSE(primary_is_waiting);
+    primary_is_waiting = true;
+
+    std::unique_lock<std::mutex> lock(delay_primary.mutex);
+    delay_primary.condition_variable.wait(lock);
+
+    primary_is_waiting = false;
+  });
+
+  // Wait for the primary thread to be waiting.
+  ASSERT_TRUE(WaitForTrue(&primary_is_waiting, kDefaultTimeout));
+
+  // Start a bunch of threads that take the shared_mutex as a reader and
+  // immediately release it. These should not be blocked.
+  std::vector<std::future<void>> reader_fleet;
+
+  for (size_t i = 0; i < kFleetSize; ++i) {
+    reader_fleet.push_back(std::async(std::launch::async, [&] {
+      shared_lock<shared_mutex> auto_lock(mutex_under_test);
+      EXPECT_TRUE(primary_is_waiting);
+      sleep_for(kShortSnooze);
+    }));
+  }
+
+  // The readers fleet should all complete successfully.
+  for (std::future<void>& reader : reader_fleet) {
+    EXPECT_EQ(std::future_status::ready, reader.wait_for(kDefaultTimeout));
+  }
+
+  // Verify the primary reader's state and signal it to finish.
+  EXPECT_TRUE(primary_is_waiting);
+  delay_primary.condition_variable.notify_one();
+  EXPECT_EQ(std::future_status::ready,
+            primary_reader.wait_for(kDefaultTimeout));
+}
+
+TEST(SharedMutexUnitTests, ReadersBlockWriters) {
+  shared_mutex mutex_under_test;
+
+  // Used to signal the reader thread to stop holding the lock.
+  ConditionVariablePair delay_reader;
+  // Used to indicate if the reader thread is waiting. (and thus holding the
+  // mutex)
+  std::atomic<bool> reader_is_waiting(false);
+
+  // Start a thread that takes the shared_mutex as a reader and holds it until
+  // signalled to stop.
+  std::future<void> reader = std::async(std::launch::async, [&] {
+    shared_lock<shared_mutex> auto_lock(mutex_under_test);
+
+    ASSERT_FALSE(reader_is_waiting);
+    reader_is_waiting = true;
+
+    std::unique_lock<std::mutex> lock(delay_reader.mutex);
+    delay_reader.condition_variable.wait(lock);
+
+    reader_is_waiting = false;
+  });
+
+  // Wait for the reader thread to be waiting.
+  ASSERT_TRUE(WaitForTrue(&reader_is_waiting, kDefaultTimeout));
+
+  // Start a thread that takes the shared_mutex as a writer and immediately
+  // releases it. This should be blocked by the reader above.
+  std::future<void> writer = std::async(std::launch::async, [&] {
+    std::unique_lock<shared_mutex> auto_lock(mutex_under_test);
+    EXPECT_FALSE(reader_is_waiting);
+    sleep_for(kShortSnooze);
+  });
+
+  // The writer should not be complete still, because it is waiting on the
+  // shared_mutex.
+  EXPECT_NE(std::future_status::ready, writer.wait_for(kDefaultTimeout));
+
+  // Verify the reader is still waiting and signal it to finish.
+  EXPECT_TRUE(reader_is_waiting);
+  delay_reader.condition_variable.notify_one();
+
+  // Make sure the reader actually finishes.
+  EXPECT_EQ(std::future_status::ready, reader.wait_for(kDefaultTimeout));
+
+  // The writer should now be able to complete as well.
+  EXPECT_EQ(std::future_status::ready, writer.wait_for(kDefaultTimeout));
+}
+
+TEST(SharedMutexUnitTests, WritersBlockReaders) {
+  shared_mutex mutex_under_test;
+
+  // Used to signal the writer thread to stop holding the lock.
+  ConditionVariablePair delay_writer;
+  // Used to indicate if the writer thread is waiting. (and thus holding the
+  // mutex)
+  std::atomic<bool> writer_is_waiting(false);
+
+  // Start a thread that takes the shared_mutex as a writer and holds it until
+  // signalled to stop.
+  std::future<void> writer = std::async(std::launch::async, [&] {
+    std::unique_lock<shared_mutex> auto_lock(mutex_under_test);
+
+    ASSERT_FALSE(writer_is_waiting);
+    writer_is_waiting = true;
+
+    std::unique_lock<std::mutex> lock(delay_writer.mutex);
+    delay_writer.condition_variable.wait(lock);
+
+    writer_is_waiting = false;
+  });
+
+  // Wait for the writer thread to be waiting.
+  ASSERT_TRUE(WaitForTrue(&writer_is_waiting, kDefaultTimeout));
+
+  // Start a thread that takes the shared_mutex as a reader and immediately
+  // releases it. This should be blocked by the writer above.
+  std::future<void> reader = std::async(std::launch::async, [&] {
+    shared_lock<shared_mutex> auto_lock(mutex_under_test);
+    EXPECT_FALSE(writer_is_waiting);
+    sleep_for(kShortSnooze);
+  });
+
+  // The reader should not be complete still, because it is waiting on the
+  // shared_mutex.
+  EXPECT_NE(std::future_status::ready, reader.wait_for(kDefaultTimeout));
+
+  // Verify the writer is still waiting and signal it to finish.
+  EXPECT_TRUE(writer_is_waiting);
+  delay_writer.condition_variable.notify_one();
+
+  // Make sure the writer actually finishes.
+  EXPECT_EQ(std::future_status::ready, writer.wait_for(kDefaultTimeout));
+
+  // The reader should now be able to complete as well.
+  EXPECT_EQ(std::future_status::ready, reader.wait_for(kDefaultTimeout));
+}
+
+TEST(SharedMutexUnitTests, WritersBlockWriters) {
+  shared_mutex mutex_under_test;
+
+  // Used to signal the primary writer thread to stop holding the lock.
+  ConditionVariablePair delay_primary;
+  // Used to indicate if the primary thread is waiting. (and thus holding the
+  // mutex)
+  std::atomic<bool> primary_is_waiting(false);
+
+  // Start a thread that takes the shared_mutex as a writer and holds it until
+  // signalled to stop.
+  std::future<void> primary = std::async(std::launch::async, [&] {
+    std::unique_lock<shared_mutex> auto_lock(mutex_under_test);
+
+    ASSERT_FALSE(primary_is_waiting);
+    primary_is_waiting = true;
+
+    std::unique_lock<std::mutex> lock(delay_primary.mutex);
+    delay_primary.condition_variable.wait(lock);
+
+    primary_is_waiting = false;
+  });
+
+  // Wait for the primary thread to be waiting.
+  ASSERT_TRUE(WaitForTrue(&primary_is_waiting, kDefaultTimeout));
+
+  // Start a thread that takes the shared_mutex as a writer and immediately
+  // releases it. This should be blocked by the primary writer above.
+  std::future<void> secondary = std::async(std::launch::async, [&] {
+    std::unique_lock<shared_mutex> auto_lock(mutex_under_test);
+    EXPECT_FALSE(primary_is_waiting);
+    sleep_for(kShortSnooze);
+  });
+
+  // The secondary writer should not be complete still, because it is waiting on
+  // the shared_mutex.
+  EXPECT_NE(std::future_status::ready, secondary.wait_for(kDefaultTimeout));
+
+  // Verify the primary is still waiting and signal it to finish.
+  EXPECT_TRUE(primary_is_waiting);
+  delay_primary.condition_variable.notify_one();
+
+  // Make sure the primary writer actually finished.
+  EXPECT_EQ(std::future_status::ready, primary.wait_for(kDefaultTimeout));
+
+  // The secondary writer should now be complete as well.
+  EXPECT_EQ(std::future_status::ready, secondary.wait_for(kDefaultTimeout));
+}
+
+TEST(SharedMutexUnitTests, NonConcurrentLocksDoNotBlock) {
+  shared_mutex mutex_under_test;
+
+  for (size_t i = 0; i < kFleetSize / 2; ++i) {
+    // Start a thread that takes the shared_mutex as a reader and immediately
+    // releases it.
+    std::future<void> reader = std::async(std::launch::async, [&] {
+      shared_lock<shared_mutex> auto_lock(mutex_under_test);
+      sleep_for(kShortSnooze);
+    });
+
+    ASSERT_EQ(std::future_status::ready, reader.wait_for(kDefaultTimeout));
+
+    // Start a thread that takes the shared_mutex as a writer and immediately
+    // releases it.
+    std::future<void> writer = std::async(std::launch::async, [&] {
+      std::unique_lock<shared_mutex> auto_lock(mutex_under_test);
+      sleep_for(kShortSnooze);
+    });
+
+    ASSERT_EQ(std::future_status::ready, writer.wait_for(kDefaultTimeout));
+  }
+}
+
+TEST(SharedMutexUnitTests, LargeVolumeOfThreadsSortsItselfOutEventually) {
+  shared_mutex mutex_under_test;
+
+  // Start a lot of threads that try to hold the lock at once.
+  std::vector<std::future<void>> fleet;
+
+  for (size_t i = 0; i < kFleetSize / 2; ++i) {
+    fleet.push_back(std::async(std::launch::async, [&] {
+      std::unique_lock<shared_mutex> auto_lock(mutex_under_test);
+      sleep_for(kShortSnooze);
+    }));
+
+    fleet.push_back(std::async(std::launch::async, [&] {
+      shared_lock<shared_mutex> auto_lock(mutex_under_test);
+      sleep_for(kShortSnooze);
+    }));
+  }
+
+  // The fleet should all complete eventually.
+  for (std::future<void>& future : fleet) {
+    EXPECT_EQ(std::future_status::ready, future.wait_for(kDefaultTimeout));
+  }
+}
+
+}  // namespace wvcdm

libwvdrmengine/cdm/test/Android.mk

@@ -107,6 +107,10 @@ test_src_dir := .
 test_main := ../core/test/test_main.cpp
 include $(LOCAL_PATH)/integration-test.mk
 
+test_name := rw_lock_test
+test_src_dir := ../core/test
+include $(LOCAL_PATH)/integration-test.mk
+
 test_name := service_certificate_unittest
 test_src_dir := ../core/test
 include $(LOCAL_PATH)/unit-test.mk

libwvdrmengine/cdm/util/include/rw_lock.h

@@ -0,0 +1,65 @@
+// Copyright 2019 Google LLC. All Rights Reserved. This file and proprietary
+// source code may only be used and distributed under the Widevine Master
+// License Agreement.
+
+#ifndef WVCDM_UTIL_RW_LOCK_H_
+#define WVCDM_UTIL_RW_LOCK_H_
+
+#include <stdint.h>
+
+#include <condition_variable>
+#include <mutex>
+
+#include "disallow_copy_and_assign.h"
+#include "util_common.h"
+
+namespace wvcdm {
+
+// A simple reader-writer mutex implementation that mimics the one from C++17
+class CORE_UTIL_EXPORT shared_mutex {
+ public:
+  shared_mutex() : reader_count_(0), has_writer_(false) {}
+  ~shared_mutex();
+
+  // These methods take the mutex as a reader. They do not fulfill the
+  // SharedMutex requirement from the C++14 STL, but they fulfill enough of it
+  // to be used with |shared_lock| below.
+  void lock_shared();
+  void unlock_shared();
+
+  // These methods take the mutex as a writer. They fulfill the Mutex
+  // requirement from the C++11 STL so that this mutex can be used with
+  // |std::unique_lock|.
+  void lock() { lock_implementation(false); }
+  bool try_lock() { return lock_implementation(true); }
+  void unlock();
+
+ private:
+  bool lock_implementation(bool abort_if_unavailable);
+
+  uint32_t reader_count_;
+  bool has_writer_;
+
+  std::mutex mutex_;
+  std::condition_variable condition_variable_;
+
+  CORE_DISALLOW_COPY_AND_ASSIGN(shared_mutex);
+};
+
+// A simple reader lock implementation that mimics the one from C++14
+template <typename Mutex>
+class shared_lock {
+ public:
+  explicit shared_lock(Mutex& lock) : lock_(&lock) { lock_->lock_shared(); }
+  explicit shared_lock(Mutex* lock) : lock_(lock) { lock_->lock_shared(); }
+  ~shared_lock() { lock_->unlock_shared(); }
+
+ private:
+  Mutex* lock_;
+
+  CORE_DISALLOW_COPY_AND_ASSIGN(shared_lock);
+};
+
+}  // namespace wvcdm
+
+#endif  // WVCDM_UTIL_RW_LOCK_H_

libwvdrmengine/cdm/util/src/rw_lock.cpp

@@ -0,0 +1,60 @@
+// Copyright 2019 Google LLC. All Rights Reserved. This file and proprietary
+// source code may only be used and distributed under the Widevine Master
+// License Agreement.
+
+#include "rw_lock.h"
+
+#include "log.h"
+
+namespace wvcdm {
+
+shared_mutex::~shared_mutex() {
+  if (reader_count_ > 0) {
+    LOGE("shared_mutex destroyed with active readers!");
+  }
+  if (has_writer_) {
+    LOGE("shared_mutex destroyed with an active writer!");
+  }
+}
+
+void shared_mutex::lock_shared() {
+  std::unique_lock<std::mutex> lock(mutex_);
+
+  while (has_writer_) {
+    condition_variable_.wait(lock);
+  }
+
+  ++reader_count_;
+}
+
+void shared_mutex::unlock_shared() {
+  std::unique_lock<std::mutex> lock(mutex_);
+
+  --reader_count_;
+
+  if (reader_count_ == 0) {
+    condition_variable_.notify_all();
+  }
+}
+
+bool shared_mutex::lock_implementation(bool abort_if_unavailable) {
+  std::unique_lock<std::mutex> lock(mutex_);
+
+  while (reader_count_ > 0 || has_writer_) {
+    if (abort_if_unavailable) return false;
+    condition_variable_.wait(lock);
+  }
+
+  has_writer_ = true;
+  return true;
+}
+
+void shared_mutex::unlock() {
+  std::unique_lock<std::mutex> lock(mutex_);
+
+  has_writer_ = false;
+
+  condition_variable_.notify_all();
+}
+
+}  // namespace wvcdm

libwvdrmengine/run_all_unit_tests.sh

@@ -113,6 +113,7 @@ adb_shell_run license_keys_unittest
 adb_shell_run license_unittest
 adb_shell_run policy_engine_constraints_unittest
 adb_shell_run policy_engine_unittest
+adb_shell_run rw_lock_test
 adb_shell_run service_certificate_unittest
 adb_shell_run timer_unittest
 adb_shell_run usage_table_header_unittest