Tasking module (NVIDIA#5436)

# Tasking module
`dali::tasking` is a task execution engine which supports inter-task dependencies, semaphores and data passing.
Main concepts of the tasking API are:
* Task - a single-use object that encapsulates a function; it can depend on zero or more Waitable objects
* Waitable - an object for which a task can wait
* Scheduler - manages tasks and resolves their readiness state

## Main components
A `Task` is also `Waitable` - that is, a task can depend on the completion of other task(s).
There are two kinds of `Waitable` objects - completion events and releasable objects.
A `CompletionEvent` (e.g. Task) is an object which remains acquirable after it's "completed". Performing Acquire operation on it does not alter the state of the object.
A `Semaphore` can be externally released (it's `Releasable`), but `Acquire` lowers the semaphore count, so the number of tasks that can acquire a semaphore before it's released is limited.

Acquisition of `Waitable` objects on behalf of waiting tasks is the duty of the `Scheduler`. By contrast, `Release` operation on `Releasable` objects can be performed in any context.

## Data passing
Data can be passed between tasks. It's wrapped in a `TaskResult` type which stores either the value as `std::any` or the error as `std::exception_ptr`. An attempt to access the value with an exception present results in the exception being rethrown.
A `Task` wraps a function taking 0 or 1 argument (in the latter case it's this-like pointer to the `Task`) and returning a value of values.
Upon creation of the task, the caller defines the number of return values. There are two options:
- Scalar return value - in which case the return value can be any objects convertible to `std::any`. If the task returns `std::any`, it's stripped and the stored value is copied.
- Multiple return values - in this case the return value of the function must be either a collection or a `tuple`.
A `Task` can `Subscribe` to the result of a producer task (which has to be done before the producer is submitted to the scheduler). The result can be obtained by calling `task->GetInputValue<T>(index)`

## Obtaining results
When a task is added to the scheduler, a `TaskFuture` object can be obtained. `Scheduler::AddSilentTask` should be used if the result of the task is not needed. Calling `TaskFuture::Value` will wait for the task to complete and return the value produced by the task (or rethrow the exception).

## Error handling
All errors thrown by the task functions are stored in the task results. An attempt to obtain results of a task that threw an exception will result in the exception being rethrown.

---------

Signed-off-by: Michal Zientkiewicz <[email protected]>

Author: mzient

Doxyfile

@@ -177,7 +177,7 @@ SHORT_NAMES            = NO
 # description.)
 # The default value is: NO.
 
-JAVADOC_AUTOBRIEF      = NO
+JAVADOC_AUTOBRIEF      = YES
 
 # If the QT_AUTOBRIEF tag is set to YES then doxygen will interpret the first
 # line (until the first dot) of a Qt-style comment as the brief description. If

dali/core/CMakeLists.txt

@@ -1,4 +1,4 @@
-# Copyright (c) 2017-2018, NVIDIA CORPORATION. All rights reserved.
+# Copyright (c) 2017-2024, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -16,6 +16,7 @@ project(dali_core CUDA CXX C)
 
 add_subdirectory(mm)
 add_subdirectory(os)
+add_subdirectory(exec)
 
 # Get all the source files
 collect_headers(DALI_INST_HDRS PARENT_SCOPE)

dali/core/exec/CMakeLists.txt

@@ -0,0 +1,20 @@
+# Copyright (c) 2024, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+#
+# 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.
+
+add_subdirectory(tasking)
+
+# Get all the source files
+collect_headers(DALI_INST_HDRS PARENT_SCOPE)
+collect_sources(DALI_CORE_SRCS PARENT_SCOPE)
+collect_test_sources(DALI_CORE_TEST_SRCS PARENT_SCOPE)

dali/core/exec/tasking/CMakeLists.txt

@@ -0,0 +1,18 @@
+# Copyright (c) 2024, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+#
+# 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.
+
+# Get all the source files
+collect_headers(DALI_INST_HDRS PARENT_SCOPE)
+collect_sources(DALI_CORE_SRCS PARENT_SCOPE)
+collect_test_sources(DALI_CORE_TEST_SRCS PARENT_SCOPE)

dali/core/exec/tasking/scheduler.cc

@@ -0,0 +1,108 @@
+// Copyright (c) 2024, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+//
+// 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.
+
+#include <cassert>
+#include <mutex>
+#include <iostream>
+#include "dali/core/exec/tasking/scheduler.h"
+
+namespace dali::tasking {
+
+bool Scheduler::AcquireAllAndMoveToReady(SharedTask &task) noexcept {
+  assert(task->state_ <= TaskState::Pending);
+
+  // All or nothing - first we check that all preconditions are met
+  for (auto &w : task->preconditions_)
+    if (!w->IsAcquirable())
+      return false;  // at least one unmet
+  // If they are, we acquire them - this must succeed
+  for (auto &w : task->preconditions_)
+    if (!w->TryAcquire(task)) {
+      std::cerr
+          << "Internal error - resource acquisition failed for a resource known to be available"
+          << std::endl;
+      std::abort();
+    }
+
+  task->preconditions_.clear();
+  task->state_ = TaskState::Ready;
+  pending_.Remove(task);
+  ready_.push(std::move(task));
+  return true;
+}
+
+void Scheduler::Notify(Waitable *w) {
+  bool is_completion_event = dynamic_cast<CompletionEvent *>(w) != nullptr;
+  bool is_task = is_completion_event && dynamic_cast<Task *>(w);
+
+  int new_ready = 0;
+  {
+    std::lock_guard g(mtx_);
+    if (is_task)
+      task_done_.notify_all();
+
+    SmallVector<SharedTask, 8> waiting;
+    int n = w->waiting_.size();
+    waiting.reserve(n);
+    for (int i = 0; i < n; i++)
+      waiting.emplace_back(w->waiting_[i]);
+
+    for (auto &task : waiting) {
+      // If the waitable is a completion event, it will never become unacquirable again.
+      // Otherwise, we have to re-check it.
+      if (!is_completion_event && !w->IsAcquirable())
+        break;
+      if (task->Ready())
+        continue;
+
+      // If the task has only one precondition or the waitable is a completion event,
+      // then we can just try to acquire that waitable on behalf of the task.
+      // A completion event, once complete, is never un-completed and all waiting threads
+      // will be able to acquire it. This menas that we can eagerly acquire it without risking
+      // deadlocks. This imposes less overhead than re-checking all preconditions each time.
+      if (is_completion_event ||
+          (task->preconditions_.size() == 1 && task->preconditions_.begin()->get() == w)) {
+        // try acquire - the only way this can fail is that the task was
+        // re-checked in another thread and marked as ready...
+        if (!w->TryAcquire(task)) {
+          assert(task->preconditions_.size() != 1 || task->preconditions_.begin()->get() != w);
+          continue;  // ... if so, nothing to do
+        }
+        auto it = std::find_if(task->preconditions_.begin(), task->preconditions_.end(),
+                               [w](auto &pre) { return pre.get() == w; });
+        assert(it != task->preconditions_.end());
+        task->preconditions_.erase(it);
+        if (task->Ready()) {
+          pending_.Remove(task);
+          task->state_ = TaskState::Ready;
+          ready_.push(std::move(task));
+          new_ready++;
+          // OK, the task is ready, we're done with it
+          continue;
+        }
+      }
+
+      if (AcquireAllAndMoveToReady(task))
+        new_ready++;
+    }
+  }
+
+
+  if (new_ready == 1)
+    this->task_ready_.notify_one();
+  else if (new_ready > 1)
+    this->task_ready_.notify_all();
+}
+
+}  // namespace dali::tasking