Initial implementation of CountDownLatch

PiperOrigin-RevId: 394478506

cpp/platform/impl/windows/BUILD

@@ -57,6 +57,7 @@ cc_library(
         "bluetooth_classic_server_socket.h",
         "bluetooth_classic_socket.h",
         "condition_variable.h",
+        "count_down_latch.h",
         "mutex.h",
         "server_sync.h",
         "webrtc.h",
@@ -96,6 +97,7 @@ cc_library(
         "bluetooth_classic_server_socket.cc",
         "bluetooth_classic_socket.cc",
         "condition_variable.cc",
+        "count_down_latch.cc",
         "mutex.cc",
         "platform.cc",
         "utils.cc",
@@ -107,6 +109,7 @@ cc_library(
         "bluetooth_classic_server_socket.h",
         "bluetooth_classic_socket.h",
         "condition_variable.h",
+        "count_down_latch.h",
         "mutex.h",
     ],
     compatible_with = ["//buildenv/target:non_prod"],
@@ -150,6 +153,7 @@ cc_test(
         "atomic_boolean_test.cc",
         "atomic_reference_test.cc",
         "condition_variable_test.cc",
+        "count_down_latch_test.cc",
         "crypto_test.cc",
         "input_file_test.cc",
         "mutex_test.cc",

cpp/platform/impl/windows/count_down_latch.cc

@@ -0,0 +1,77 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "platform/impl/windows/count_down_latch.h"
+
+namespace location {
+namespace nearby {
+namespace windows {
+
+// A synchronization aid that allows one or more threads to wait until a set of
+// operations being performed in other threads completes.
+//
+// Creates or opens a named or unnamed event object.
+// https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CountDownLatch.html
+CountDownLatch::CountDownLatch(int count) {
+  if (count < 0) {
+    throw(std::invalid_argument("count"));
+  }
+  // https://docs.microsoft.com/en-us/windows/win32/api/synchapi/nf-synchapi-createeventa
+  h_count_down_latch_event_ =
+      CreateEvent(NULL,               // default security attributes
+                  TRUE,               // manual-reset event
+                  FALSE,              // initial state is nonsignaled
+                  TEXT("LatchEvent")  // object name
+      );
+  count_ = count;
+}
+
+Exception CountDownLatch::Await() {
+  // Waits until the specified object is in the signaled state or the time-out
+  // interval elapses.
+  // https://docs.microsoft.com/en-us/windows/win32/api/synchapi/nf-synchapi-waitforsingleobject
+  if (WaitForSingleObject(h_count_down_latch_event_, INFINITE) ==
+      WAIT_OBJECT_0) {
+    return Exception{Exception::kSuccess};
+  }
+  return Exception{Exception::kFailed};
+}
+
+ExceptionOr<bool> CountDownLatch::Await(absl::Duration timeout) {
+  auto result = WaitForSingleObject(h_count_down_latch_event_,
+                                    absl::ToInt64Milliseconds(timeout));
+  if (result == WAIT_OBJECT_0) {
+    return ExceptionOr<bool>(true);
+  }
+
+  if (result == WAIT_TIMEOUT) {
+    return ExceptionOr<bool>{Exception::kTimeout};
+  }
+
+  return ExceptionOr<bool>{Exception::kFailed};
+}
+
+void CountDownLatch::CountDown() {
+  // Decrements (decreases by one) the value of the specified 32-bit variable as
+  // an atomic operation.
+  // https://docs.microsoft.com/en-us/windows/win32/api/winnt/nf-winnt-interlockeddecrement
+  InterlockedDecrement(&count_);
+  if (count_ == 0) {
+    SetEvent(h_count_down_latch_event_);
+  }
+}
+
+}  // namespace windows
+}  // namespace nearby
+}  // namespace location

cpp/platform/impl/windows/count_down_latch.h

@@ -15,6 +15,9 @@
 #ifndef PLATFORM_IMPL_WINDOWS_COUNT_DOWN_LATCH_H_
 #define PLATFORM_IMPL_WINDOWS_COUNT_DOWN_LATCH_H_
 
+#include <windows.h>
+#include <synchapi.h>
+
 #include "platform/api/count_down_latch.h"
 
 namespace location {
@@ -27,17 +30,24 @@ namespace windows {
 // https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CountDownLatch.html
 class CountDownLatch : public api::CountDownLatch {
  public:
-  // TODO(b/184975123): replace with real implementation.
-  ~CountDownLatch() override = default;
+  CountDownLatch(int count);
 
-  // TODO(b/184975123): replace with real implementation.
-  Exception Await() override { return Exception{}; }
-  // TODO(b/184975123): replace with real implementation.
-  ExceptionOr<bool> Await(absl::Duration timeout) override {
-    return Exception{};
+  ~CountDownLatch() override {
+    if (h_count_down_latch_event_ != NULL) {
+      CloseHandle(h_count_down_latch_event_);
+      h_count_down_latch_event_ = NULL;
+    }
   }
-  // TODO(b/184975123): replace with real implementation.
-  void CountDown() override{};
+
+  Exception Await() override;
+
+  ExceptionOr<bool> Await(absl::Duration timeout) override;
+
+  void CountDown() override;
+
+ private:
+  HANDLE h_count_down_latch_event_ = NULL;
+  uint32_t count_;
 };
 
 }  // namespace windows

cpp/platform/impl/windows/count_down_latch_test.cc

@@ -0,0 +1,180 @@
+// Copyright 2020 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "platform/impl/windows/count_down_latch.h"
+
+#include "gtest/gtest.h"
+
+class CountDownLatchTests : public testing::Test {
+ public:
+  class TestData {
+   public:
+    std::unique_ptr<location::nearby::windows::CountDownLatch>& countDownLatch;
+    LONG volatile& count;
+  };
+
+  class CountDownLatchTest {
+   public:
+    static DWORD ThreadProcCountDown(LPVOID lpParam) {
+      TestData* testData = static_cast<TestData*>(lpParam);
+
+      Sleep(1);
+
+      InterlockedIncrement(&testData->count);
+      testData->countDownLatch->CountDown();
+      return 0;
+    }
+
+    static DWORD ThreadProcAwait(LPVOID lpParam) {
+      TestData* testData = static_cast<TestData*>(lpParam);
+
+      Sleep(1);
+
+      testData->countDownLatch->Await();
+      InterlockedIncrement(&testData->count);
+
+      return 0;
+    }
+  };
+
+  CountDownLatchTests() {}
+};
+
+TEST_F(CountDownLatchTests, CountDownLatchAwaitSucceeds) {
+  // Arrange
+  LONG volatile count = 0;
+
+  std::unique_ptr<location::nearby::windows::CountDownLatch> countDownLatch =
+      std::make_unique<location::nearby::windows::CountDownLatch>(3);
+
+  HANDLE hThreads[3];
+  DWORD dwThreadID;
+
+  TestData testData{countDownLatch, count};
+
+  // Setup 3 threads
+  for (int i = 0; i < 3; i++) {
+    // TODO: More complex scenarios may require use of a parameter
+    //   to the thread procedure, such as an event per thread to
+    //   be used for synchronization.
+    hThreads[i] = CreateThread(
+        NULL,                                     // default security
+        0,                                        // default stack size
+        CountDownLatchTest::ThreadProcCountDown,  // name of the thread function
+        &testData,                                // no thread parameters
+        0,                                        // default startup flags
+        &dwThreadID);
+
+    EXPECT_TRUE(hThreads[i] != NULL);
+  }
+
+  // Act
+  location::nearby::Exception result = countDownLatch->Await();
+
+  //
+  // Assert
+  EXPECT_EQ(result.value, location::nearby::Exception::kSuccess);
+  EXPECT_EQ(count, 3);
+}
+
+TEST_F(CountDownLatchTests, CountDownLatchAwaitTimeoutTimesOut) {
+  // Arrange
+  LONG volatile count = 0;
+
+  std::unique_ptr<location::nearby::windows::CountDownLatch> countDownLatch =
+      std::make_unique<location::nearby::windows::CountDownLatch>(3);
+
+  // Act
+  location::nearby::ExceptionOr<bool> result =
+      countDownLatch->Await(absl::Milliseconds(10));
+
+  Sleep(20);
+
+  // Assert
+  EXPECT_FALSE(result.GetResult());
+  EXPECT_EQ(result.GetException().value, location::nearby::Exception::kTimeout);
+}
+
+TEST_F(CountDownLatchTests, CountDownLatchAwaitNoTimeoutSucceeds) {
+  // Arrange
+  LONG volatile count = 0;
+
+  std::unique_ptr<location::nearby::windows::CountDownLatch> countDownLatch =
+      std::make_unique<location::nearby::windows::CountDownLatch>(3);
+
+  TestData testData{countDownLatch, count};
+
+  HANDLE hThreads[3];
+  DWORD dwThreadID;
+
+  // Setup 3 threads
+  for (int i = 0; i < 3; i++) {
+    // TODO: More complex scenarios may require use of a parameter
+    //   to the thread procedure, such as an event per thread to
+    //   be used for synchronization.
+    hThreads[i] = CreateThread(
+        NULL,                                     // default security
+        0,                                        // default stack size
+        CountDownLatchTest::ThreadProcCountDown,  // name of the thread function
+        &testData,                                // no thread parameters
+        0,                                        // default startup flags
+        &dwThreadID);
+
+    EXPECT_TRUE(hThreads[i] != NULL);
+  }
+
+  WaitForMultipleObjects(3, hThreads, true, INFINITE);
+  // Act
+  location::nearby::ExceptionOr<bool> result =
+      countDownLatch->Await(absl::Milliseconds(100));
+
+  // Assert
+  EXPECT_TRUE(result.GetResult());
+  EXPECT_EQ(result.GetException().value, location::nearby::Exception::kSuccess);
+  EXPECT_EQ(count, 3);
+}
+
+TEST_F(CountDownLatchTests, CountDownLatchCountDownBeforeAwaitSucceeds) {
+  // Arrange
+  LONG volatile count = 0;
+  std::unique_ptr<location::nearby::windows::CountDownLatch> countDownLatch =
+      std::make_unique<location::nearby::windows::CountDownLatch>(1);
+
+  HANDLE hThread;
+  DWORD dwThreadID;
+
+  TestData testData{countDownLatch, count};
+
+  hThread = CreateThread(
+      NULL,                                 // default security
+      0,                                    // default stack size
+      CountDownLatchTest::ThreadProcAwait,  // name of the thread function
+      &testData,                            // no thread parameters
+      0,                                    // default startup flags
+      &dwThreadID);
+
+  EXPECT_TRUE(hThread != NULL);
+
+  // Act
+  countDownLatch->CountDown();  // This countdown occurs before the thread has a
+                                // chance to run
+
+  if (hThread != NULL) {
+    WaitForSingleObject(hThread,
+                        INFINITE);  // This will wait till the thread exits
+  }
+
+  // Assert
+  EXPECT_EQ(count, 1);
+}

cpp/platform/impl/windows/mutex.cc

@@ -23,60 +23,27 @@ namespace windows {
 //
 // https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/locks/Lock.html
 Mutex::Mutex(Mutex::Mode mode)
-    : mode_(mode), owning_thread_(std::this_thread::get_id()) {
-  InitializeCriticalSection(&critical_section_);
+    : mode_(mode) {
 }
 
 void Mutex::Lock() {
-  EnterCriticalSection(&critical_section_);
-
-  std::thread::id currentThread = std::this_thread::get_id();
-
-  if ((mode_ == Mutex::Mode::kRegular ||
-       mode_ == Mutex::Mode::kRegularNoCheck) &&
-      !locked_) {
-    mutex_actual_.lock();
-    owning_thread_ = currentThread;
-    locked_ = true;
-  }
-
-  if (mode_ == Mutex::Mode::kRecursive) {
-    if (!locked_) {
-      owning_thread_ = currentThread;
-    }
-    if (owning_thread_ == currentThread) {
-      try {
-        recursive_mutex_actual_.lock();
-      } catch ([[maybe_unused]] const std::system_error& e) {
-      // Eat the exception and fail silently, argument left for debug
-      }
-      locked_ = true;
+  if (mode_ == Mutex::Mode::kRegular || mode_ == Mutex::Mode::kRegularNoCheck) {
+    mutex_impl_.lock();
+  } else {
+    if (mode_ == Mutex::Mode::kRecursive) {
+      recursive_mutex_impl_.lock();
     }
   }
-
-  LeaveCriticalSection(&critical_section_);
 }
 
 void Mutex::Unlock() {
-  EnterCriticalSection(&critical_section_);
-
-  std::thread::id currentThread = std::this_thread::get_id();
-
   if (mode_ == Mutex::Mode::kRegular || mode_ == Mutex::Mode::kRegularNoCheck) {
-    if (currentThread == owning_thread_) {
-      mutex_actual_.unlock();
-      locked_ = false;
+    mutex_impl_.unlock();
+  } else {
+    if (mode_ == Mutex::Mode::kRecursive) {
+      recursive_mutex_impl_.unlock();
     }
   }
-
-  if (mode_ == Mutex::Mode::kRecursive) {
-    if (currentThread == owning_thread_) {
-      recursive_mutex_actual_.unlock();
-      locked_ = false;
-    }
-  }
-
-  LeaveCriticalSection(&critical_section_);
 }
 
 std::mutex& Mutex::GetWindowsMutex() { return mutex_; }

cpp/platform/impl/windows/mutex.h

@@ -49,14 +49,16 @@ class Mutex : public api::Mutex {
 
  private:
   Mutex::Mode mode_;
-  bool locked_ = false;
 
-  std::mutex mutex_actual_;
-  std::mutex& mutex_ = mutex_actual_;
-  std::recursive_mutex recursive_mutex_actual_;
-  std::recursive_mutex& recursive_mutex_ = recursive_mutex_actual_;
-  std::thread::id owning_thread_;
-  CRITICAL_SECTION critical_section_;
+  std::mutex mutex_impl_;  //  The actual mutex allocation
+  std::mutex& mutex_ =
+      mutex_impl_;  //  This is passed to other windows functions, must be by
+                    //  reference to avoid ownership problems
+  std::recursive_mutex recursive_mutex_impl_;  //  The actual mutex allocation
+  std::recursive_mutex& recursive_mutex_ =
+      recursive_mutex_impl_;  //  This is passed to other windows functions,
+                              //  must be by reference to avoid ownership
+                              //  problems
 };
 
 }  // namespace windows

cpp/platform/impl/windows/platform.cc

@@ -46,13 +46,11 @@ std::string GetPayloadPath(PayloadId payload_id) {
 }
 }  // namespace
 
-// TODO(b/184975123): replace with real implementation.
 std::unique_ptr<AtomicBoolean> ImplementationPlatform::CreateAtomicBoolean(
     bool initial_value) {
   return absl::make_unique<windows::AtomicBoolean>();
 }
 
-// TODO(b/184975123): replace with real implementation.
 std::unique_ptr<AtomicUint32> ImplementationPlatform::CreateAtomicUint32(
     std::uint32_t value) {
   return absl::make_unique<windows::AtomicUint32>();
@@ -61,15 +59,13 @@ std::unique_ptr<AtomicUint32> ImplementationPlatform::CreateAtomicUint32(
 // TODO(b/184975123): replace with real implementation.
 std::unique_ptr<CountDownLatch> ImplementationPlatform::CreateCountDownLatch(
     std::int32_t count) {
-  return absl::make_unique<windows::CountDownLatch>();
+  return absl::make_unique<windows::CountDownLatch>(count);
 }
 
-// TODO(b/184975123): replace with real implementation.
 std::unique_ptr<Mutex> ImplementationPlatform::CreateMutex(Mutex::Mode mode) {
   return absl::make_unique<windows::Mutex>(mode);
 }
 
-// TODO(b/184975123): replace with real implementation.
 std::unique_ptr<ConditionVariable>
 ImplementationPlatform::CreateConditionVariable(Mutex* mutex) {
   return absl::make_unique<location::nearby::windows::ConditionVariable>(mutex);

cpp/platform/impl/windows/utils.h

@@ -15,7 +15,7 @@
 #ifndef PLATFORM_IMPL_WINDOWS_UTILS_H_
 #define PLATFORM_IMPL_WINDOWS_UTILS_H_
 
-#include <Windows.h>
+#include <windows.h>
 #include <stdio.h>
 
 #include <string>