Alow NamedMutex to share an ApiError among threads (#1249)

It is done to prevent request repetition if the same request
in other thread have ended up with an error.
This change fixes an issue when task scheduler is overloaded
by the same requests that fails.

Relates-To: OLPSUP-15885

Signed-off-by: Mykola Malik <[email protected]>

Author: mykolamalik

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

@@ -162,10 +162,25 @@ BlobApi::DataResponse DataRepository::GetBlobData(
     }
   }
 
+  // Check if other threads have faced an error.
+  const auto optional_error = mutex.GetError();
+  if (optional_error) {
+    OLP_SDK_LOG_DEBUG_F(kLogTag,
+                        "Found error in NamedMutex, aborting, hrn='%s', "
+                        "key='%s', error='%s'",
+                        catalog_.ToCatalogHRNString().c_str(),
+                        data_handle->c_str(),
+                        optional_error->GetMessage().c_str());
+    return *optional_error;
+  }
+
   auto storage_api_lookup = lookup_client_.LookupApi(
       service, "v1", static_cast<client::FetchOptions>(fetch_option), context);
 
   if (!storage_api_lookup.IsSuccessful()) {
+    // Store an error to share it with other threads.
+    mutex.SetError(storage_api_lookup.GetError());
+
     return storage_api_lookup.GetError();
   }
 
@@ -204,6 +219,9 @@ BlobApi::DataResponse DataRepository::GetBlobData(
           catalog_.ToCatalogHRNString().c_str(), data_handle->c_str());
       repository.Clear(layer, data_handle.value());
     }
+
+    // Store an error to share it with other threads.
+    mutex.SetError(storage_response.GetError());
   }
 
   return storage_response;

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2020 HERE Europe B.V.
+ * Copyright (C) 2020-2021 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.
@@ -30,11 +30,14 @@ class NamedMutexStorage::Impl {
  public:
   std::mutex& AquireLock(const std::string& resource);
   void ReleaseLock(const std::string& resource);
+  void SetError(const std::string& resource, const client::ApiError& error);
+  boost::optional<client::ApiError> GetError(const std::string& resource);
 
  private:
   struct RefCounterMutex {
     std::mutex mutex;
     uint32_t use_count{0u};
+    boost::optional<client::ApiError> optional_error;
   };
 
   std::mutex mutex_;
@@ -61,6 +64,26 @@ void NamedMutexStorage::Impl::ReleaseLock(const std::string& resource) {
   }
 }
 
+void NamedMutexStorage::Impl::SetError(const std::string& resource,
+                                       const client::ApiError& error) {
+  std::lock_guard<std::mutex> lock(mutex_);
+  auto mutex_it = mutexes_.find(resource);
+  if (mutex_it != mutexes_.end()) {
+    mutex_it->second.optional_error = error;
+  }
+}
+
+boost::optional<client::ApiError> NamedMutexStorage::Impl::GetError(
+    const std::string& resource) {
+  std::lock_guard<std::mutex> lock(mutex_);
+  auto mutex_it = mutexes_.find(resource);
+  if (mutex_it == mutexes_.end()) {
+    return boost::none;
+  }
+
+  return mutex_it->second.optional_error;
+}
+
 NamedMutexStorage::NamedMutexStorage() : impl_(std::make_shared<Impl>()) {}
 
 std::mutex& NamedMutexStorage::AquireLock(const std::string& resource) {
@@ -71,6 +94,16 @@ void NamedMutexStorage::ReleaseLock(const std::string& resource) {
   impl_->ReleaseLock(resource);
 }
 
+void NamedMutexStorage::SetError(const std::string& resource,
+                                 const client::ApiError& error) {
+  impl_->SetError(resource, error);
+}
+
+boost::optional<client::ApiError> NamedMutexStorage::GetError(
+    const std::string& resource) {
+  return impl_->GetError(resource);
+}
+
 NamedMutex::NamedMutex(NamedMutexStorage& storage, const std::string& name)
     : storage_{storage}, name_{name}, mutex_{storage_.AquireLock(name_)} {}
 
@@ -82,6 +115,14 @@ bool NamedMutex::try_lock() { return mutex_.try_lock(); }
 
 void NamedMutex::unlock() { mutex_.unlock(); }
 
+void NamedMutex::SetError(const client::ApiError& error) {
+  storage_.SetError(name_, error);
+}
+
+boost::optional<client::ApiError> NamedMutex::GetError() {
+  return storage_.GetError(name_);
+}
+
 }  // namespace repository
 }  // namespace read
 }  // namespace dataservice

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2020 HERE Europe B.V.
+ * Copyright (C) 2020-2021 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.
@@ -24,14 +24,18 @@
 #include <string>
 #include <unordered_map>
 
+#include <olp/core/client/ApiError.h>
+#include <boost/optional.hpp>
+
 namespace olp {
 namespace dataservice {
 namespace read {
 namespace repository {
 
 /*
  * @brief A mutex storage class, used to store and access mutex primitives by
- * name, so it can be used in different places and different conditions.
+ * name, so it can be used in different places and different conditions. Has an
+ * ability to share an `ApiError` among threads that uses the same mutex.
  */
 class NamedMutexStorage {
  public:
@@ -40,14 +44,33 @@ class NamedMutexStorage {
   std::mutex& AquireLock(const std::string& resource);
   void ReleaseLock(const std::string& resource);
 
+  /**
+   * @brief Saves an error to share it among threads.
+   *
+   * @param resource A name of a mutex to store an error for.
+   * @param error An error to be stored.
+   */
+  void SetError(const std::string& resource, const client::ApiError& error);
+
+  /**
+   * @brief Gets the stored error for provided resource.
+   *
+   * @param resource A name of a mutex to get an error for.
+   *
+   * @return The stored `ApiError` instance or `boost::none` if no error has
+   * been stored for this resource.
+   */
+  boost::optional<client::ApiError> GetError(const std::string& resource);
+
  private:
   class Impl;
   std::shared_ptr<Impl> impl_;
 };
 
 /*
  * @brief A synchronization primitive that can be used to protect shared data
- * from being simultaneously accessed by multiple threads.
+ * from being simultaneously accessed by multiple threads. Has an ability to
+ * share an `ApiError` among the threads.
  */
 class NamedMutex final {
  public:
@@ -60,10 +83,27 @@ class NamedMutex final {
 
   ~NamedMutex();
 
+  // These method names are lower snake case to be able to use NamedMutex with
+  // std::unique_lock.
   void lock();
   bool try_lock();
   void unlock();
 
+  /**
+   * @brief Saves an error to share it among threads.
+   *
+   * @param error An error to be stored.
+   */
+  void SetError(const client::ApiError& error);
+
+  /**
+   * @brief Gets the stored error for this mutex.
+   *
+   * @return The stored `ApiError` instance  or `boost::none` if no error has
+   * been stored for this mutex.
+   */
+  boost::optional<client::ApiError> GetError();
+
  private:
   NamedMutexStorage& storage_;
   std::string name_;

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

@@ -30,6 +30,7 @@
 #include <olp/dataservice/read/DataRequest.h>
 #include <olp/dataservice/read/TileRequest.h>
 #include <repositories/DataRepository.h>
+#include <repositories/NamedMutex.h>
 #include <repositories/PartitionsCacheRepository.h>
 #include <repositories/PartitionsRepository.h>
 
@@ -273,6 +274,67 @@ TEST_F(DataRepositoryTest, GetBlobDataInProgressCancel) {
             olp::client::ErrorCode::Cancelled);
 }
 
+TEST_F(DataRepositoryTest, GetBlobDataSimultaniousFailedCalls) {
+  // The lookup data must be requested from the network only once
+  EXPECT_CALL(*network_mock_, Send(IsGetRequest(kUrlLookup), _, _, _, _))
+      .WillOnce(ReturnHttpResponse(olp::http::NetworkResponse().WithStatus(
+                                       olp::http::HttpStatusCode::OK),
+                                   kUrlResponseLookup));
+
+  std::promise<void> network_request_started_promise;
+  std::promise<void> finish_network_request_promise;
+
+  auto wait = [&]() {
+    network_request_started_promise.set_value();
+    finish_network_request_promise.get_future().wait();
+  };
+
+  // The blob data must be requested from the network only once
+  EXPECT_CALL(*network_mock_, Send(IsGetRequest(kUrlBlobData269), _, _, _, _))
+      .WillOnce(testing::DoAll(
+          testing::InvokeWithoutArgs(wait),
+          ReturnHttpResponse(olp::http::NetworkResponse().WithStatus(
+                                 olp::http::HttpStatusCode::REQUEST_TIMEOUT),
+                             "Timeout")));
+
+  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);
+
+  // Start first request in a separate thread
+  std::thread first_request_thread([&]() {
+    auto response =
+        repository.GetBlobData(kLayerId, kService, request, context);
+    EXPECT_FALSE(response.IsSuccessful());
+  });
+
+  // Wait untill network request processing started
+  network_request_started_promise.get_future().wait();
+
+  // Get a mutex from the storage. It guarantees that when the second thread
+  // 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);
+
+  // Start second request in a separate thread
+  std::thread second_request_thread([&]() {
+    auto response =
+        repository.GetBlobData(kLayerId, kService, request, context);
+    EXPECT_FALSE(response.IsSuccessful());
+  });
+
+  finish_network_request_promise.set_value();
+  first_request_thread.join();
+  second_request_thread.join();
+}
+
 TEST_F(DataRepositoryTest, GetVersionedDataTile) {
   EXPECT_CALL(*network_mock_, Send(IsGetRequest(kUrlLookup), _, _, _, _))
       .WillOnce(ReturnHttpResponse(olp::http::NetworkResponse().WithStatus(