Merge the parallel query partitions requests. (#1113)

The order of partition ids in the request matters. Worst case - two
parallel downloads. This is a temporary workaround.
Change the log in TaskSink to print the number of tasks pending.
Add a new method to PendingRequests to query a number of task
pending.
Fix a typo.

Resolves: OLPEDGE-2395

Signed-off-by: Mykhailo Kuchma <[email protected]>

Author: mykhailo-kuchma

olp-cpp-sdk-core/include/olp/core/client/PendingRequests.h

@@ -68,10 +68,17 @@ class CORE_API PendingRequests final {
    */
   void Remove(TaskContext task_context);
 
+  /**
+   * @brief Gets the number of tasks.
+   *
+   * @return The number of tasks pending.
+   */
+  size_t GetTaskCount() const;
+
  private:
   using ContextMap = std::unordered_set<TaskContext, TaskContextHash>;
   ContextMap task_contexts_;
-  std::mutex task_contexts_lock_;
+  mutable std::mutex task_contexts_lock_;
 };
 
 }  // namespace client

olp-cpp-sdk-core/src/client/ApiLookupClientImpl.cpp

@@ -242,13 +242,12 @@ boost::optional<OlpClient> ApiLookupClientImpl::GetCachedClient(
     std::lock_guard<std::mutex> lock(cached_clients_mutex_);
     const auto client_it = cached_clients_.find(key);
     if (client_it != cached_clients_.end()) {
-      const ClientWithExpiration& client_with_expirtation = client_it->second;
-      if (client_with_expirtation.expire_at >
-          std::chrono::steady_clock::now()) {
+      const ClientWithExpiration& client_with_expiration = client_it->second;
+      if (client_with_expiration.expire_at > std::chrono::steady_clock::now()) {
         OLP_SDK_LOG_DEBUG_F(
             kLogTag, "LookupApi(%s/%s) found in client cache, hrn='%s'",
             service.c_str(), service_version.c_str(), catalog_string_.c_str());
-        return client_with_expirtation.client;
+        return client_with_expiration.client;
       }
     }
   }

olp-cpp-sdk-core/src/client/PendingRequests.cpp

@@ -71,5 +71,10 @@ void PendingRequests::Remove(TaskContext task_context) {
   task_contexts_.erase(task_context);
 }
 
+size_t PendingRequests::GetTaskCount() const {
+  std::lock_guard<std::mutex> lock(task_contexts_lock_);
+  return task_contexts_.size();
+}
+
 }  // namespace client
 }  // namespace olp

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

@@ -34,10 +34,14 @@ TaskSink::TaskSink(std::shared_ptr<thread::TaskScheduler> task_scheduler)
       closed_(false) {}
 
 TaskSink::~TaskSink() {
-  OLP_SDK_LOG_INFO(kLogTag, "Finishing, canceling all current tasks.");
   {
     std::lock_guard<std::mutex> lock(mutex_);
     closed_ = true;
+    const auto task_count = pending_requests_->GetTaskCount();
+    if (task_count > 0) {
+      OLP_SDK_LOG_INFO_F(kLogTag, "Finishing, canceling %" PRIu64 " tasks.",
+                         static_cast<std::uint64_t>(task_count));
+    }
   }
   // CancelAllAndWait method should be called without mutex, since potentially
   // there might be new added tasks, it may result in deadlock.

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

@@ -83,34 +83,33 @@ bool VersionedLayerClientImpl::CancelPendingRequests() {
 
 client::CancellationToken VersionedLayerClientImpl::GetPartitions(
     PartitionsRequest request, PartitionsResponseCallback callback) {
-  auto catalog = catalog_;
-  auto layer_id = layer_id_;
-  auto settings = settings_;
-  auto lookup_client = lookup_client_;
-
   auto partitions_task =
-      [=](client::CancellationContext context) mutable -> PartitionsResponse {
-    if (request.GetFetchOption() == CacheWithUpdate) {
-      return {{client::ErrorCode::InvalidArgument,
-               "CacheWithUpdate option can not be used for versioned "
-               "layer"}};
+      [this](PartitionsRequest partitions_request,
+             client::CancellationContext context) -> PartitionsResponse {
+    const auto fetch_option = partitions_request.GetFetchOption();
+    if (fetch_option == CacheWithUpdate) {
+      return client::ApiError(
+          client::ErrorCode::InvalidArgument,
+          "CacheWithUpdate option can not be used for versioned layer");
     }
 
     auto version_response =
-        GetVersion(request.GetBillingTag(), request.GetFetchOption(), context);
+        GetVersion(partitions_request.GetBillingTag(), fetch_option, context);
     if (!version_response.IsSuccessful()) {
       return version_response.GetError();
     }
+
     const auto version = version_response.GetResult().GetVersion();
 
-    repository::PartitionsRepository repository(std::move(catalog), layer_id,
-                                                std::move(settings),
-                                                std::move(lookup_client));
-    return repository.GetVersionedPartitions(request, version, context);
+    repository::PartitionsRepository repository(catalog_, layer_id_, settings_,
+                                                lookup_client_);
+    return repository.GetVersionedPartitionsExtendedResponse(
+        std::move(partitions_request), version, context);
   };
 
-  return task_sink_.AddTask(std::move(partitions_task), std::move(callback),
-                            thread::NORMAL);
+  return task_sink_.AddTask(
+      std::bind(partitions_task, std::move(request), std::placeholders::_1),
+      std::move(callback), thread::NORMAL);
 }
 
 client::CancellableFuture<PartitionsResponse>

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

@@ -22,6 +22,8 @@
 #include <algorithm>
 #include <utility>
 
+#include <boost/functional/hash.hpp>
+
 #include <olp/core/client/Condition.h>
 #include <olp/core/logging/Log.h>
 #include "CatalogRepository.h"
@@ -91,6 +93,16 @@ 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 {
@@ -152,24 +164,39 @@ PartitionsRepository::GetPartitionsExtendedResponse(
   auto fetch_option = request.GetFetchOption();
   const auto key = request.CreateKey(layer_id_);
 
+  const auto catalog_str = catalog_.ToCatalogHRNString();
+
+  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) {
       OLP_SDK_LOG_DEBUG_F(kLogTag,
                           "GetPartitions found in cache, hrn='%s', key='%s'",
-                          catalog_.ToCatalogHRNString().c_str(), key.c_str());
+                          catalog_str.c_str(), key.c_str());
       return cached_partitions.get();
     } else if (fetch_option == CacheOnly) {
       OLP_SDK_LOG_INFO_F(kLogTag,
                          "GetPartitions not found in cache, hrn='%s', key='%s'",
-                         catalog_.ToCatalogHRNString().c_str(), key.c_str());
+                         catalog_str.c_str(), key.c_str());
       return {{client::ErrorCode::NotFound,
                "CacheOnly: resource not found in cache"}};
     }
   }
 
   QueryApi::PartitionsExtendedResponse response;
-  const auto& partition_ids = request.GetPartitionIds();
 
   if (partition_ids.empty()) {
     auto metadata_api = lookup_client_.LookupApi(
@@ -202,7 +229,7 @@ PartitionsRepository::GetPartitionsExtendedResponse(
   if (response.IsSuccessful() && fetch_option != OnlineOnly) {
     OLP_SDK_LOG_DEBUG_F(kLogTag,
                         "GetPartitions put to cache, hrn='%s', key='%s'",
-                        catalog_.ToCatalogHRNString().c_str(), key.c_str());
+                        catalog_str.c_str(), key.c_str());
     cache_.Put(response.GetResult(), version, expiry, is_layer_metadata);
   }
   if (!response.IsSuccessful()) {
@@ -211,7 +238,7 @@ PartitionsRepository::GetPartitionsExtendedResponse(
       OLP_SDK_LOG_WARNING_F(
           kLogTag,
           "GetPartitions 403 received, remove from cache, hrn='%s', key='%s'",
-          catalog_.ToCatalogHRNString().c_str(), key.c_str());
+          catalog_str.c_str(), key.c_str());
       cache_.Clear();
     }
   }