Add check for IsCancelled in NamedMutex (#1354)

Pass CancellationContext to NamedMutex. Check IsCancelled before
locking mutex. Add GetBlobDataCancelParralellRequest test.

Resolves: OLPSUP-20250
Signed-off-by: Yevhen Krasilnyk <[email protected]>

Signed-off-by: Yevhen Krasilnyk <[email protected]>

Author: Smartkras

olp-cpp-sdk-dataservice-read/src/ApiClientLookup.cpp

@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2019-2021 HERE Europe B.V.
+ * Copyright (C) 2019-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.
@@ -59,7 +59,8 @@ ApiClientLookup::ApiClientResponse ApiClientLookup::LookupApi(
     std::string service_version, FetchOptions options,
     client::OlpClientSettings settings, repository::NamedMutexStorage storage) {
   // This mutex is required to avoid concurrent requests to online.
-  repository::NamedMutex mutex(storage, catalog.ToString());
+  repository::NamedMutex mutex(storage, catalog.ToString(),
+                               cancellation_context);
   std::unique_lock<repository::NamedMutex> lock(mutex, std::defer_lock);
 
   // If we are not planning to go online or access the cache, do not lock.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2019-2021 HERE Europe B.V.
+ * Copyright (C) 2019-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.
@@ -158,7 +158,8 @@ BlobApi::DataResponse DataRepository::GetBlobData(
     return client::ApiError::PreconditionFailed("Data handle is missing");
   }
 
-  NamedMutex mutex(storage_, catalog_.ToString() + layer + *data_handle);
+  NamedMutex mutex(storage_, catalog_.ToString() + layer + *data_handle,
+                   context);
   std::unique_lock<NamedMutex> lock(mutex, std::defer_lock);
 
   // If we are not planning to go online or access the cache, do not lock.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2020-2021 HERE Europe B.V.
+ * Copyright (C) 2020-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.
@@ -19,6 +19,8 @@
 
 #include "NamedMutex.h"
 
+#include <thread>
+
 #include <olp/core/porting/make_unique.h>
 
 namespace olp {
@@ -104,16 +106,31 @@ boost::optional<client::ApiError> NamedMutexStorage::GetError(
   return impl_->GetError(resource);
 }
 
-NamedMutex::NamedMutex(NamedMutexStorage& storage, const std::string& name)
-    : storage_{storage}, name_{name}, mutex_{storage_.AquireLock(name_)} {}
+NamedMutex::NamedMutex(NamedMutexStorage& storage, const std::string& name,
+                       const client::CancellationContext& context)
+    : storage_{storage},
+      context_{context},
+      is_locked_{false},
+      name_{name},
+      mutex_{storage_.AquireLock(name_)} {}
 
 NamedMutex::~NamedMutex() { storage_.ReleaseLock(name_); }
 
-void NamedMutex::lock() { mutex_.lock(); }
+void NamedMutex::lock() {
+  while (!is_locked_ && !context_.IsCancelled()) {
+    is_locked_ = mutex_.try_lock();
+    std::this_thread::yield();
+  }
+}
 
-bool NamedMutex::try_lock() { return mutex_.try_lock(); }
+bool NamedMutex::try_lock() { return is_locked_ = mutex_.try_lock(); }
 
-void NamedMutex::unlock() { mutex_.unlock(); }
+void NamedMutex::unlock() {
+  if (is_locked_) {
+    mutex_.unlock();
+    is_locked_ = false;
+  }
+}
 
 void NamedMutex::SetError(const client::ApiError& error) {
   storage_.SetError(name_, error);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2020-2021 HERE Europe B.V.
+ * Copyright (C) 2020-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.
@@ -25,6 +25,7 @@
 #include <unordered_map>
 
 #include <olp/core/client/ApiError.h>
+#include <olp/core/client/CancellationContext.h>
 #include <boost/optional.hpp>
 
 namespace olp {
@@ -74,7 +75,8 @@ class NamedMutexStorage {
  */
 class NamedMutex final {
  public:
-  NamedMutex(NamedMutexStorage& storage, const std::string& name);
+  NamedMutex(NamedMutexStorage& storage, const std::string& name,
+             const client::CancellationContext& context);
 
   NamedMutex(const NamedMutex&) = delete;
   NamedMutex(NamedMutex&&) = delete;
@@ -106,6 +108,8 @@ class NamedMutex final {
 
  private:
   NamedMutexStorage& storage_;
+  const client::CancellationContext& context_;
+  bool is_locked_;
   std::string name_;
   std::mutex& mutex_;
 };

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

@@ -242,7 +242,8 @@ PartitionsRepository::GetPartitionsExtendedResponse(
       partition_ids.empty() ? "" : HashPartitions(partition_ids);
   const auto version_str = version ? std::to_string(*version) : "";
 
-  NamedMutex mutex(storage_, catalog_str + layer_id_ + version_str + detail);
+  NamedMutex mutex(storage_, catalog_str + layer_id_ + version_str + detail,
+                   context);
   std::unique_lock<NamedMutex> lock(mutex, std::defer_lock);
 
   // If we are not planning to go online or access the cache, do not lock.
@@ -343,7 +344,7 @@ PartitionsResponse PartitionsRepository::GetPartitionById(
   const auto request_key =
       catalog_.ToString() + request.CreateKey(layer_id_, version);
 
-  NamedMutex mutex(storage_, request_key);
+  NamedMutex mutex(storage_, request_key, context);
   std::unique_lock<repository::NamedMutex> lock(mutex, std::defer_lock);
 
   // If we are not planning to go online or access the cache, do not lock.
@@ -431,7 +432,7 @@ QuadTreeIndexResponse PartitionsRepository::GetQuadTreeIndexForTile(
       catalog_.ToCatalogHRNString(), layer_id_, root_tile_key, version,
       kAggregateQuadTreeDepth);
 
-  NamedMutex mutex(storage_, quad_cache_key);
+  NamedMutex mutex(storage_, quad_cache_key, context);
   std::unique_lock<NamedMutex> lock(mutex, std::defer_lock);
 
   // If we are not planning to go online or access the cache, do not lock.

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

@@ -201,7 +201,7 @@ client::NetworkStatistics PrefetchTilesRepository::LoadAggregatedSubQuads(
       const auto quad_cache_key = cache::KeyGenerator::CreateQuadTreeKey(
           catalog_str_, layer_id_, root, version, kMaxQuadTreeIndexDepth);
 
-      NamedMutex mutex(storage_, quad_cache_key);
+      NamedMutex mutex(storage_, quad_cache_key, context);
       std::unique_lock<NamedMutex> lock(mutex);
 
       if (!cache_repository_.ContainsTree(root, kMaxQuadTreeIndexDepth,
@@ -229,7 +229,7 @@ SubQuadsResponse PrefetchTilesRepository::GetVersionedSubQuads(
   const auto quad_cache_key = cache::KeyGenerator::CreateQuadTreeKey(
       catalog_str_, layer_id_, tile, version, kMaxQuadTreeIndexDepth);
 
-  NamedMutex mutex(storage_, quad_cache_key);
+  NamedMutex mutex(storage_, quad_cache_key, context);
   std::lock_guard<NamedMutex> lock(mutex);
 
   if (cache_repository_.Get(tile, depth, version, quad_tree)) {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2019-2021 HERE Europe B.V.
+ * Copyright (C) 2019-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.
@@ -19,6 +19,10 @@
 
 #include <gtest/gtest.h>
 
+#include <chrono>
+#include <future>
+#include <thread>
+
 #include <matchers/NetworkUrlMatchers.h>
 #include <mocks/NetworkMock.h>
 #include <olp/core/cache/CacheSettings.h>
@@ -321,7 +325,7 @@ TEST_F(DataRepositoryTest, GetBlobDataSimultaniousFailedCalls) {
   // accuares the mutex, the stored error will not be cleaned up in scope of
   // ReleaseLock call from the first thread
   olp::dataservice::read::repository::NamedMutex mutex(
-      storage, hrn.ToString() + kService + kUrlBlobDataHandle);
+      storage, hrn.ToString() + kService + kUrlBlobDataHandle, context);
 
   // Start second request in a separate thread
   std::thread second_request_thread([&]() {
@@ -523,4 +527,77 @@ TEST_F(DataRepositoryTest, GetVersionedDataTileReturnEmpty) {
   ASSERT_EQ(response.GetError().GetMessage(),
             "Failed to parse quad tree response");
 }
+
+TEST_F(DataRepositoryTest, GetBlobDataCancelParralellRequest) {
+  EXPECT_CALL(*network_mock_, Send(IsGetRequest(kUrlLookup), _, _, _, _))
+      .WillRepeatedly(
+          ReturnHttpResponse(olp::http::NetworkResponse().WithStatus(
+                                 olp::http::HttpStatusCode::OK),
+                             kUrlResponseLookup));
+
+  constexpr auto wait_time = std::chrono::seconds(1);
+  auto wait = [&wait_time]() { std::this_thread::sleep_for(wait_time); };
+
+  EXPECT_CALL(*network_mock_, Send(IsGetRequest(kUrlBlobData269), _, _, _, _))
+      .WillRepeatedly(testing::DoAll(
+          testing::InvokeWithoutArgs(wait),
+          ReturnHttpResponse(olp::http::NetworkResponse().WithStatus(
+                                 olp::http::HttpStatusCode::OK),
+                             "Some Data")));
+
+  olp::client::CancellationContext context;
+  olp::dataservice::read::repository::NamedMutexStorage storage;
+
+  olp::dataservice::read::DataRequest request;
+  request.WithDataHandle(kUrlBlobDataHandle);
+
+  olp::client::HRN hrn(GetTestCatalog());
+  ApiLookupClient lookup_client(hrn, *settings_);
+  DataRepository repository(hrn, *settings_, lookup_client, storage);
+
+  std::promise<void> first_thread_finished_promise;
+  std::promise<void> second_thread_finished_promise;
+
+  // Start first request in a separate thread
+  std::thread first_request_thread([&]() {
+    auto response =
+        repository.GetBlobData(kLayerId, kService, request, context);
+
+    EXPECT_FALSE(response);
+    EXPECT_EQ(response.GetError().GetErrorCode(),
+              olp::client::ErrorCode::Cancelled);
+
+    first_thread_finished_promise.set_value();
+  });
+
+  // Start second request in a separate thread
+  std::thread second_request_thread([&]() {
+    auto response =
+        repository.GetBlobData(kLayerId, kService, request, context);
+
+    EXPECT_FALSE(response);
+    EXPECT_EQ(response.GetError().GetErrorCode(),
+              olp::client::ErrorCode::Cancelled);
+
+    second_thread_finished_promise.set_value();
+  });
+
+  const auto start = std::chrono::high_resolution_clock::now();
+
+  // Start threads w/o wait for them to finish
+  first_request_thread.detach();
+  second_request_thread.detach();
+
+  // Cancel operation should immeadeately finish both requests
+  context.CancelOperation();
+
+  // Wait until threads are finished
+  first_thread_finished_promise.get_future().wait();
+  second_thread_finished_promise.get_future().wait();
+
+  const auto end = std::chrono::high_resolution_clock::now();
+
+  // Compare time spanding for waiting for threads to finish
+  EXPECT_LT(end - start, wait_time);
+}
 }  // namespace