Preemptive cancellation of TaskContext jobs

- Enable pre-execution cancellation in TaskContext by registering a
  cancellation token in constructor that invokes PreExecuteCancel()
- Clear cancellation token in ~TaskContextImpl to prevent dangling refs
- Fix CancellationContext::CancelOperation() to swap out the token
  before calling Cancel(), avoiding re-entrancy issues
- Update BlockingCancel test to cover pre-execution cancellation

Relates-To: HERESDK-12253
Signed-off-by: Mykhailo Diachenko <ext-mykhailo.z.diachenko@here.com>

Author: diachenko-mischa

olp-cpp-sdk-core/include/olp/core/client/CancellationContext.inl

@@ -19,6 +19,7 @@
 
 #pragma once
 
+#include <utility>
 namespace olp {
 namespace client {
 
@@ -58,9 +59,10 @@ inline void CancellationContext::CancelOperation() {
     return;
   }
 
-  impl_->sub_operation_cancel_token_.Cancel();
-  impl_->sub_operation_cancel_token_ = CancellationToken();
+  auto token = CancellationToken();
+  std::swap(token, impl_->sub_operation_cancel_token_);
   impl_->is_cancelled_ = true;
+  token.Cancel();
 }
 
 inline bool CancellationContext::IsCancelled() const {

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

@@ -193,9 +193,19 @@ class CORE_API TaskContext {
         : execute_func_(std::move(execute_func)),
           callback_(std::move(callback)),
           context_(std::move(context)),
-          state_{State::PENDING} {}
+          state_{State::PENDING} {
+      context_.ExecuteOrCancelled(
+          [&]() -> CancellationToken {
+            return CancellationToken([&] { PreExecuteCancel(); });
+          },
+          []() {});
+    }
 
-    ~TaskContextImpl() override{};
+    ~TaskContextImpl() override {
+      // Overwrite the token to prevent dangling reference to this.
+      context_.ExecuteOrCancelled(
+          []() -> CancellationToken { return CancellationToken(); }, []() {});
+    };
 
     /**
      * @brief Checks for the cancellation, executes the task, and calls
@@ -249,6 +259,40 @@ class CORE_API TaskContext {
       state_.store(State::COMPLETED);
     }
 
+    void PreExecuteCancel() {
+      State expected_state = State::PENDING;
+
+      if (!state_.compare_exchange_strong(expected_state, State::IN_PROGRESS)) {
+        return;
+      }
+
+      // Moving the user callback and function guarantee that they are
+      // executed exactly once
+      ExecuteFunc function = nullptr;
+      UserCallback callback = nullptr;
+
+      {
+        std::lock_guard<std::mutex> lock(mutex_);
+        function = std::move(execute_func_);
+        callback = std::move(callback_);
+      }
+
+      Response user_response =
+          client::ApiError(client::ErrorCode::Cancelled, "Cancelled");
+
+      if (callback) {
+        callback(std::move(user_response));
+      }
+
+      // Resources need to be released before the notification, else lambas
+      // would have captured resources like network or `TaskScheduler`.
+      function = nullptr;
+      callback = nullptr;
+
+      condition_.Notify();
+      state_.store(State::COMPLETED);
+    }
+
     /**
      * @brief Cancels the operation and waits for the notification.
      *

olp-cpp-sdk-core/tests/client/TaskContextTest.cpp

@@ -136,19 +136,29 @@ TEST(TaskContextTest, BlockingCancel) {
 
   Callback callback = [&](Response r) { response = std::move(r); };
 
-  TaskContext context = TaskContext::Create(func, callback);
+  {
+    SCOPED_TRACE("Pre-exec cancellation");
+    TaskContext context = TaskContext::Create(func, callback);
+    EXPECT_TRUE(context.BlockingCancel(std::chrono::seconds(0)));
+    EXPECT_FALSE(response.IsSuccessful());
+    EXPECT_EQ(response.GetError().GetErrorCode(), ErrorCode::Cancelled);
+  }
 
-  EXPECT_FALSE(context.BlockingCancel(std::chrono::seconds(0)));
+  {
+    SCOPED_TRACE("Pre-exec cancellation race");
+    response = Response{};
+    TaskContext context = TaskContext::Create(func, callback);
 
-  std::thread cancel_thread([&]() { EXPECT_TRUE(context.BlockingCancel()); });
+    std::thread cancel_thread([&]() { EXPECT_TRUE(context.BlockingCancel()); });
 
-  std::thread execute_thread([&]() { context.Execute(); });
+    std::thread execute_thread([&]() { context.Execute(); });
 
-  execute_thread.join();
-  cancel_thread.join();
+    execute_thread.join();
+    cancel_thread.join();
 
-  EXPECT_FALSE(response.IsSuccessful());
-  EXPECT_EQ(response.GetError().GetErrorCode(), ErrorCode::Cancelled);
+    EXPECT_FALSE(response.IsSuccessful());
+    EXPECT_EQ(response.GetError().GetErrorCode(), ErrorCode::Cancelled);
+  }
 }
 
 TEST(TaskContextTest, BlockingCancelIsWaiting) {