Revert NamedMutex removal (#1174)

Merging same requests inside OlpClient::CallApi
    does not prevent running a new request
    between end of the previous call & putting its
    result into a cache.
    For concurrent cases it leads to multiple parallel
    requests and network mock issues in tests.

    As a quick fix NamedMutexes are reverted for
    methods where they had been used before.

    Relates-To: OLPEDGE-2417

Signed-off-by: Iuliia Moroz <[email protected]>

Author: jmoro3

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

@@ -28,6 +28,7 @@
 #include <olp/core/logging/Log.h>
 #include "CatalogRepository.h"
 #include "DataCacheRepository.h"
+#include "NamedMutex.h"
 #include "PartitionsCacheRepository.h"
 #include "PartitionsRepository.h"
 #include "generated/api/BlobApi.h"
@@ -134,6 +135,14 @@ BlobApi::DataResponse DataRepository::GetBlobData(
   repository::DataCacheRepository repository(
       catalog_, settings_.cache, settings_.default_cache_expiration);
 
+  NamedMutex mutex(catalog_.ToString() + layer + *data_handle);
+  std::unique_lock<NamedMutex> lock(mutex, std::defer_lock);
+
+  // If we are not planning to go online or access the cache, do not lock.
+  if (fetch_option != CacheOnly && fetch_option != OnlineOnly) {
+    lock.lock();
+  }
+
   if (fetch_option != OnlineOnly && fetch_option != CacheWithUpdate) {
     auto cached_data = repository.Get(layer, data_handle.value());
     if (cached_data) {

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

@@ -27,6 +27,7 @@
 #include <olp/core/client/Condition.h>
 #include <olp/core/logging/Log.h>
 #include "CatalogRepository.h"
+#include "NamedMutex.h"
 #include "generated/api/MetadataApi.h"
 #include "generated/api/QueryApi.h"
 #include "olp/dataservice/read/CatalogRequest.h"
@@ -92,6 +93,15 @@ repository::PartitionResponse FindPartition(
 
   return std::move(aggregated_partition);
 }
+
+std::string HashPartitions(
+    const read::PartitionsRequest::PartitionIds& partitions) {
+  size_t seed = 0;
+  for (const auto& partition : partitions) {
+    boost::hash_combine(seed, partition);
+  }
+  return std::to_string(seed);
+}
 }  // namespace
 
 namespace olp {
@@ -157,6 +167,18 @@ PartitionsRepository::GetPartitionsExtendedResponse(
 
   const auto& partition_ids = request.GetPartitionIds();
 
+  // Temporary workaround for merging the same requests. Should be removed after
+  // OlpClient could handle that.
+  const auto detail =
+      partition_ids.empty() ? "" : HashPartitions(partition_ids);
+  NamedMutex mutex(catalog_str + layer_id_ + detail);
+  std::unique_lock<NamedMutex> lock(mutex, std::defer_lock);
+
+  // If we are not planning to go online or access the cache, do not lock.
+  if (fetch_option != CacheOnly && fetch_option != OnlineOnly) {
+    lock.lock();
+  }
+
   if (fetch_option != OnlineOnly && fetch_option != CacheWithUpdate) {
     auto cached_partitions = cache_.Get(request, version);
     if (cached_partitions) {
@@ -243,6 +265,14 @@ PartitionsResponse PartitionsRepository::GetPartitionById(
   const auto request_key =
       catalog_.ToString() + request.CreateKey(layer_id_, version);
 
+  NamedMutex mutex(request_key);
+  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.
+  if (fetch_option != CacheOnly && fetch_option != OnlineOnly) {
+    lock.lock();
+  }
+
   std::chrono::seconds timeout{settings_.retry_settings.timeout};
   const auto key = request.CreateKey(layer_id_, version);
 
@@ -318,6 +348,14 @@ QuadTreeIndexResponse PartitionsRepository::GetQuadTreeIndexForTile(
   const auto& root_tile_key = tile_key.ChangedLevelBy(-kAggregateQuadTreeDepth);
   const auto root_tile_here = root_tile_key.ToHereTile();
 
+  NamedMutex mutex(catalog_.ToString() + layer_id_ + root_tile_here + "Index");
+  std::unique_lock<NamedMutex> lock(mutex, std::defer_lock);
+
+  // If we are not planning to go online or access the cache, do not lock.
+  if (fetch_option != CacheOnly && fetch_option != OnlineOnly) {
+    lock.lock();
+  }
+
   // Look for QuadTree covering the tile in the cache
   if (fetch_option != OnlineOnly && fetch_option != CacheWithUpdate) {
     read::QuadTreeIndex cached_tree;