Fix the data race in TaskSink class (#1091)

The data race could happen when the instance is destroyed while the
a new task is being added and already passed the check condition.

Switch StreamLayerClientImpl, CatalogClientImpl to TaskSink.
Remove ExecuteOrSchedule helpers.

Resolves: OLPSUP-12083

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

Author: mykhailo-kuchma

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

@@ -28,7 +28,6 @@
 
 #include "Common.h"
 #include "repositories/CatalogRepository.h"
-#include "repositories/ExecuteOrSchedule.inl"
 
 namespace olp {
 namespace dataservice {
@@ -42,25 +41,17 @@ CatalogClientImpl::CatalogClientImpl(client::HRN catalog,
                                      client::OlpClientSettings settings)
     : catalog_(std::move(catalog)),
       settings_(std::move(settings)),
-      lookup_client_(catalog_, settings_) {
+      lookup_client_(catalog_, settings_),
+      task_sink_(settings_.task_scheduler) {
   if (!settings_.cache) {
     settings_.cache = client::OlpClientSettingsFactory::CreateDefaultCache({});
   }
-
-  // to avoid capturing task scheduler inside a task, we need a copy of settings
-  // without the scheduler
-  task_scheduler_ = std::move(settings_.task_scheduler);
-
-  pending_requests_ = std::make_shared<client::PendingRequests>();
-}
-
-CatalogClientImpl::~CatalogClientImpl() {
-  pending_requests_->CancelAllAndWait();
 }
 
 bool CatalogClientImpl::CancelPendingRequests() {
   OLP_SDK_LOG_TRACE(kLogTag, "CancelPendingRequests");
-  return pending_requests_->CancelAll();
+  task_sink_.CancelTasks();
+  return true;
 }
 
 client::CancellationToken CatalogClientImpl::GetCatalog(
@@ -77,8 +68,8 @@ client::CancellationToken CatalogClientImpl::GetCatalog(
       return repository.GetCatalog(request, std::move(context));
     };
 
-    return AddTask(task_scheduler_, pending_requests_,
-                   std::move(get_catalog_task), std::move(callback));
+    return task_sink_.AddTask(std::move(get_catalog_task), std::move(callback),
+                              thread::NORMAL);
   };
 
   return ScheduleFetch(std::move(schedule_get_catalog), std::move(request),
@@ -111,8 +102,8 @@ client::CancellationToken CatalogClientImpl::GetLatestVersion(
       return repository.GetLatestVersion(request, std::move(context));
     };
 
-    return AddTask(task_scheduler_, pending_requests_,
-                   std::move(get_latest_version_task), std::move(callback));
+    return task_sink_.AddTask(std::move(get_latest_version_task),
+                              std::move(callback), thread::NORMAL);
   };
 
   return ScheduleFetch(std::move(schedule_get_latest_version),
@@ -142,8 +133,8 @@ client::CancellationToken CatalogClientImpl::ListVersions(
     return repository.GetVersionsList(request, std::move(context));
   };
 
-  return AddTask(task_scheduler_, pending_requests_,
-                 std::move(versions_list_task), std::move(callback));
+  return task_sink_.AddTask(std::move(versions_list_task), std::move(callback),
+                            thread::NORMAL);
 }
 
 client::CancellableFuture<VersionsResponse> CatalogClientImpl::ListVersions(

olp-cpp-sdk-dataservice-read/src/CatalogClientImpl.h

@@ -31,6 +31,8 @@
 #include <olp/dataservice/read/Types.h>
 #include <olp/dataservice/read/VersionsRequest.h>
 
+#include "TaskSink.h"
+
 namespace olp {
 namespace client {
 class OlpClient;
@@ -48,8 +50,6 @@ class CatalogClientImpl final {
  public:
   CatalogClientImpl(client::HRN catalog, client::OlpClientSettings settings);
 
-  ~CatalogClientImpl();
-
   bool CancelPendingRequests();
 
   client::CancellationToken GetCatalog(CatalogRequest request,
@@ -72,9 +72,8 @@ class CatalogClientImpl final {
  private:
   client::HRN catalog_;
   client::OlpClientSettings settings_;
-  std::shared_ptr<thread::TaskScheduler> task_scheduler_;
-  std::shared_ptr<client::PendingRequests> pending_requests_;
   client::ApiLookupClient lookup_client_;
+  TaskSink task_sink_;
 };
 
 }  // namespace read

olp-cpp-sdk-dataservice-read/src/Common.h

@@ -23,7 +23,6 @@
 #include <olp/core/client/PendingRequests.h>
 #include <olp/core/thread/TaskScheduler.h>
 #include <olp/dataservice/read/FetchOptions.h>
-#include "repositories/ExecuteOrSchedule.inl"
 
 namespace olp {
 namespace dataservice {
@@ -62,59 +61,6 @@ inline client::CancellationToken ScheduleFetch(TaskScheduler&& schedule_task,
                        std::forward<Callback>(callback));
 }
 
-/*
- * @brief Common function used to wrap a lambda function and a callback that
- * consumes the function result with a TaskContext class and schedule this to a
- * task scheduler.
- * @param task_scheduler Task scheduler instance.
- * @param pending_requests PendingRequests instance that tracks current
- * requests.
- * @param task Function that will be executed.
- * @param callback Function that will consume task output.
- * @param priority Priority of the task.
- * @param args Additional agrs to pass to TaskContext.
- * @return CancellationToken used to cancel the operation.
- */
-template <typename Function, typename Callback, typename... Args>
-inline client::CancellationToken AddTaskWithPriority(
-    const std::shared_ptr<thread::TaskScheduler>& task_scheduler,
-    const std::shared_ptr<client::PendingRequests>& pending_requests,
-    Function task, Callback callback, uint32_t priority, Args&&... args) {
-  auto context = client::TaskContext::Create(
-      std::move(task), std::move(callback), std::forward<Args>(args)...);
-  pending_requests->Insert(context);
-
-  repository::ExecuteOrSchedule(task_scheduler,
-                                [=] {
-                                  context.Execute();
-                                  pending_requests->Remove(context);
-                                },
-                                priority);
-
-  return context.CancelToken();
-}
-
-/*
- * @brief Common function used to wrap a lambda function and a callback that
- * consumes the function result with a TaskContext class and schedule this to a
- * task scheduler with NORMAL priority.
- * @param task_scheduler Task scheduler instance.
- * @param pending_requests PendingRequests instance that tracks current
- * requests.
- * @param task Function that will be executed.
- * @param callback Function that will consume task output.
- * @param args Additional agrs to pass to TaskContext.
- * @return CancellationToken used to cancel the operation.
- */
-template <typename Function, typename Callback, typename... Args>
-inline client::CancellationToken AddTask(
-    const std::shared_ptr<thread::TaskScheduler>& task_scheduler,
-    const std::shared_ptr<client::PendingRequests>& pending_requests,
-    Function task, Callback callback, Args&&... args) {
-  return AddTaskWithPriority(task_scheduler, pending_requests, task, callback,
-                             thread::NORMAL, std::forward<Args>(args)...);
-}
-
 }  // namespace read
 }  // namespace dataservice
 }  // namespace olp

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

@@ -31,7 +31,6 @@
 #include "Common.h"
 #include "generated/api/BlobApi.h"
 #include "generated/api/StreamApi.h"
-#include "repositories/ExecuteOrSchedule.inl"
 
 namespace olp {
 namespace dataservice {
@@ -60,20 +59,19 @@ StreamLayerClientImpl::StreamLayerClientImpl(client::HRN catalog,
     : catalog_(std::move(catalog)),
       layer_id_(std::move(layer_id)),
       settings_(std::move(settings)),
-      pending_requests_(std::make_shared<client::PendingRequests>()),
-      lookup_client_(catalog_, settings_) {
+      lookup_client_(catalog_, settings_),
+      task_sink_(settings_.task_scheduler) {
   if (!settings_.cache) {
     settings_.cache = client::OlpClientSettingsFactory::CreateDefaultCache({});
   }
 }
 
-StreamLayerClientImpl::~StreamLayerClientImpl() {
-  pending_requests_->CancelAllAndWait();
-}
+StreamLayerClientImpl::~StreamLayerClientImpl() = default;
 
 bool StreamLayerClientImpl::CancelPendingRequests() {
   OLP_SDK_LOG_TRACE(kLogTag, "CancelPendingRequests");
-  return pending_requests_->CancelAll();
+  task_sink_.CancelTasks();
+  return true;
 }
 
 client::CancellationToken StreamLayerClientImpl::Subscribe(
@@ -148,8 +146,8 @@ client::CancellationToken StreamLayerClientImpl::Subscribe(
     return subscripton_id;
   };
 
-  return AddTask(settings_.task_scheduler, pending_requests_,
-                 std::move(subscribe_task), std::move(callback));
+  return task_sink_.AddTask(std::move(subscribe_task), std::move(callback),
+                            thread::NORMAL);
 }
 
 client::CancellableFuture<SubscribeResponse> StreamLayerClientImpl::Subscribe(
@@ -216,8 +214,8 @@ client::CancellationToken StreamLayerClientImpl::Unsubscribe(
     return subscription_id;
   };
 
-  return AddTask(settings_.task_scheduler, pending_requests_,
-                 std::move(unsubscribe_task), std::move(callback));
+  return task_sink_.AddTask(std::move(unsubscribe_task), std::move(callback),
+                            thread::NORMAL);
 }
 
 client::CancellableFuture<UnsubscribeResponse>
@@ -265,8 +263,8 @@ client::CancellationToken StreamLayerClientImpl::GetData(
     return blob_response;
   };
 
-  return AddTask(settings_.task_scheduler, pending_requests_,
-                 std::move(get_data_task), std::move(callback));
+  return task_sink_.AddTask(std::move(get_data_task), std::move(callback),
+                            thread::NORMAL);
 }
 
 client::CancellableFuture<DataResponse> StreamLayerClientImpl::GetData(
@@ -362,8 +360,8 @@ client::CancellationToken StreamLayerClientImpl::Poll(
     return res;
   };
 
-  return AddTask(settings_.task_scheduler, pending_requests_,
-                 std::move(poll_task), std::move(callback));
+  return task_sink_.AddTask(std::move(poll_task), std::move(callback),
+                            thread::NORMAL);
 }
 
 client::CancellableFuture<PollResponse> StreamLayerClientImpl::Poll() {
@@ -422,8 +420,8 @@ client::CancellationToken StreamLayerClientImpl::Seek(
     return res;
   };
 
-  return AddTask(settings_.task_scheduler, pending_requests_,
-                 std::move(seek_task), std::move(callback));
+  return task_sink_.AddTask(std::move(seek_task), std::move(callback),
+                            thread::NORMAL);
 }
 client::CancellableFuture<SeekResponse> StreamLayerClientImpl::Seek(
     SeekRequest request) {

olp-cpp-sdk-dataservice-read/src/StreamLayerClientImpl.h

@@ -31,6 +31,8 @@
 #include <olp/dataservice/read/Types.h>
 #include <olp/dataservice/read/model/Messages.h>
 
+#include "TaskSink.h"
+
 namespace olp {
 namespace client {
 class OlpClient;
@@ -93,10 +95,10 @@ class StreamLayerClientImpl {
   client::HRN catalog_;
   std::string layer_id_;
   client::OlpClientSettings settings_;
-  std::shared_ptr<client::PendingRequests> pending_requests_;
   std::mutex mutex_;
   std::unique_ptr<StreamLayerClientContext> client_context_;
   client::ApiLookupClient lookup_client_;
+  TaskSink task_sink_;
 };
 
 }  // namespace read

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

@@ -0,0 +1,82 @@
+/*
+ * Copyright (C) 2020 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 "TaskSink.h"
+
+#include <olp/core/logging/Log.h>
+
+namespace olp {
+namespace dataservice {
+namespace read {
+namespace {
+constexpr auto kLogTag = "TaskSink";
+}
+
+TaskSink::TaskSink(std::shared_ptr<thread::TaskScheduler> task_scheduler)
+    : task_scheduler_(std::move(task_scheduler)),
+      pending_requests_(std::make_shared<client::PendingRequests>()),
+      closed_(false) {}
+
+TaskSink::~TaskSink() {
+  OLP_SDK_LOG_INFO(kLogTag, "Finishing, canceling all current tasks.");
+  {
+    std::lock_guard<std::mutex> lock(mutex_);
+    closed_ = true;
+  }
+  // CancelAllAndWait method should be called without mutex, since potentially
+  // there might be new added tasks, it may result in deadlock.
+  pending_requests_->CancelAllAndWait();
+}
+
+void TaskSink::CancelTasks() { pending_requests_->CancelAll(); }
+
+bool TaskSink::AddTaskImpl(client::TaskContext task, uint32_t priority) {
+  if (task_scheduler_) {
+    return ScheduleTask(std::move(task), priority);
+  } else {
+    ExecuteTask(std::move(task));
+    return true;
+  }
+}
+
+bool TaskSink::ScheduleTask(client::TaskContext task, uint32_t priority) {
+  std::lock_guard<std::mutex> lock(mutex_);
+  if (closed_) {
+    OLP_SDK_LOG_WARNING(
+        kLogTag, "Attempt to add a task when the sink is already closed");
+    return false;
+  }
+
+  pending_requests_->Insert(task);
+  auto pending_requests = pending_requests_;
+  task_scheduler_->ScheduleTask(
+      [=] {
+        task.Execute();
+        pending_requests->Remove(task);
+      },
+      priority);
+
+  return true;
+}
+
+void TaskSink::ExecuteTask(client::TaskContext task) { task.Execute(); }
+
+}  // namespace read
+}  // namespace dataservice
+}  // namespace olp