Fix the data race in NamedMutex class (#1357)

The race appears when the condition variable spins on the predicate, 
while the notify is not protected with a mutex.
Additionally set the cancellation method to notify, which enables the
faster tasks cancellation.

Relates-To: OAM-1784

Signed-off-by: Mykhailo Kuchma <ext-mykhailo.kuchma@here.com>

Author: mykhailo-kuchma

olp-cpp-sdk-dataservice-read/src/repositories/NamedMutex.cpp

@@ -133,7 +133,7 @@ boost::optional<client::ApiError> NamedMutexStorage::GetError(
 }
 
 NamedMutex::NamedMutex(NamedMutexStorage& storage, const std::string& name,
-                       const client::CancellationContext& context)
+                       client::CancellationContext& context)
     : storage_{storage},
       context_{context},
       is_locked_{false},
@@ -145,10 +145,15 @@ NamedMutex::NamedMutex(NamedMutexStorage& storage, const std::string& name,
 NamedMutex::~NamedMutex() { storage_.ReleaseLock(name_); }
 
 void NamedMutex::lock() {
-  if (!context_.IsCancelled() && !try_lock()) {
+  const bool valid = context_.ExecuteOrCancelled(
+      [&]() { return client::CancellationToken([&]() { Notify(); }); });
+
+  if (valid) {
     std::unique_lock<std::mutex> unique_lock{lock_mutex_};
     lock_condition_.wait(unique_lock,
                          [&] { return context_.IsCancelled() || try_lock(); });
+    // Reset the cancel token
+    context_.ExecuteOrCancelled([&]() { return client::CancellationToken(); });
   }
 }
 
@@ -158,7 +163,7 @@ void NamedMutex::unlock() {
   if (is_locked_) {
     mutex_.unlock();
     is_locked_ = false;
-    lock_condition_.notify_all();
+    Notify();
   }
 }
 
@@ -170,6 +175,11 @@ boost::optional<client::ApiError> NamedMutex::GetError() {
   return storage_.GetError(name_);
 }
 
+void NamedMutex::Notify() {
+  std::unique_lock<std::mutex> unique_lock{lock_mutex_};
+  lock_condition_.notify_all();
+}
+
 }  // namespace repository
 }  // namespace read
 }  // namespace dataservice

olp-cpp-sdk-dataservice-read/src/repositories/NamedMutex.h

@@ -80,7 +80,7 @@ class NamedMutexStorage {
 class NamedMutex final {
  public:
   NamedMutex(NamedMutexStorage& storage, const std::string& name,
-             const client::CancellationContext& context);
+             client::CancellationContext& context);
 
   NamedMutex(const NamedMutex&) = delete;
   NamedMutex(NamedMutex&&) = delete;
@@ -111,8 +111,11 @@ class NamedMutex final {
   boost::optional<client::ApiError> GetError();
 
  private:
+  /// Notify waiting threads method.
+  void Notify();
+
   NamedMutexStorage& storage_;
-  const client::CancellationContext& context_;
+  client::CancellationContext& context_;
   bool is_locked_;
   std::string name_;
   std::mutex& mutex_;

olp-cpp-sdk-dataservice-read/tests/CMakeLists.txt

@@ -24,6 +24,7 @@ set(OLP_SDK_DATASERVICE_READ_TEST_SOURCES
     DataRepositoryTest.cpp
     JsonResultParserTest.cpp
     MetadataApiTest.cpp
+    NamedMutexTest.cpp
     ParserTest.cpp
     PartitionsCacheRepositoryTest.cpp
     PartitionsRepositoryTest.cpp

olp-cpp-sdk-dataservice-read/tests/NamedMutexTest.cpp

@@ -0,0 +1,123 @@
+/*
+ * Copyright (C) 2022 HERE Europe B.V.
+ *
+ * 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
+ *
+ *     http://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.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * License-Filename: LICENSE
+ */
+
+#include <gtest/gtest.h>
+#include <atomic>
+#include <thread>
+
+#include "repositories/NamedMutex.h"
+
+namespace {
+
+namespace client = olp::client;
+namespace repository = olp::dataservice::read::repository;
+
+TEST(NamedMutexTest, Lock) {
+  repository::NamedMutexStorage storage;
+
+  client::CancellationContext context;
+
+  repository::NamedMutex main_mutex(storage, "mutex", context);
+  main_mutex.lock();
+
+  std::atomic_int counter{0};
+
+  std::thread thread_1([&]() {
+    repository::NamedMutex mutex(storage, "mutex", context);
+    mutex.lock();
+    counter.fetch_add(1);
+    mutex.unlock();
+  });
+
+  std::thread thread_2([&]() {
+    repository::NamedMutex mutex(storage, "mutex", context);
+    mutex.lock();
+    counter.fetch_add(1);
+    mutex.unlock();
+  });
+
+  EXPECT_EQ(counter.load(), 0);
+
+  main_mutex.unlock();
+
+  thread_1.join();
+  thread_2.join();
+
+  EXPECT_EQ(counter.load(), 2);
+}
+
+TEST(NamedMutexTest, Cancel) {
+  repository::NamedMutexStorage storage;
+
+  client::CancellationContext main_context;
+  client::CancellationContext thread_1_context;
+  client::CancellationContext thread_2_context;
+
+  repository::NamedMutex main_mutex(storage, "mutex", main_context);
+  main_mutex.lock();
+
+  std::atomic_int counter{0};
+
+  std::thread thread_1([&]() {
+    repository::NamedMutex mutex(storage, "mutex", thread_1_context);
+    mutex.lock();
+    if (thread_1_context.IsCancelled()) {
+      counter.fetch_add(1);
+    }
+    mutex.unlock();
+  });
+
+  std::thread thread_2([&]() {
+    repository::NamedMutex mutex(storage, "mutex", thread_2_context);
+    mutex.lock();
+    if (thread_2_context.IsCancelled()) {
+      counter.fetch_add(1);
+    }
+    mutex.unlock();
+  });
+
+  EXPECT_EQ(counter.load(), 0);
+
+  thread_1_context.CancelOperation();
+  thread_2_context.CancelOperation();
+
+  // Give other threads time to react.
+  std::this_thread::sleep_for(std::chrono::milliseconds(100));
+
+  EXPECT_EQ(counter.load(), 2);
+
+  main_mutex.unlock();
+
+  thread_1.join();
+  thread_2.join();
+}
+
+TEST(NamedMutexTest, CancelationLifetime) {
+  repository::NamedMutexStorage storage;
+
+  client::CancellationContext main_context;
+  {
+    repository::NamedMutex main_mutex(storage, "mutex", main_context);
+    main_mutex.lock();
+    main_mutex.unlock();
+  }
+  main_context.CancelOperation();
+}
+
+}  // namespace