ThreadPool: tighten task contracts and validation

Clarify cancellation, queued-task ownership, duplicate enqueue, and worker-thread
wait/shutdown restrictions in the thread pool documentation.

Reject NaN task priorities by falling back to the default priority, and preserve
the full previous affinity mask returned by SetCurrentThreadAffinity().

Author: TheMostDiligent

Common/interface/ThreadPool.h

@@ -90,10 +90,16 @@ DILIGENT_BEGIN_INTERFACE(IAsyncTask, IObject)
     VIRTUAL ASYNC_TASK_STATUS METHOD(Run)(THIS_
                                      Uint32 ThreadId) PURE;
 
-    /// Cancel the task, if possible.
-    
-    /// If the task is running, the task implementation should
-    /// abort the task execution, if possible.
+    /// Requests task cancellation, if possible.
+
+    /// This is a cooperative cancellation request. If the task is running, the
+    /// task implementation should observe the request and abort execution, if
+    /// possible. Calling this method does not remove the task from a thread
+    /// pool queue and does not guarantee that a task in the
+    /// Diligent::ASYNC_TASK_STATUS_NOT_STARTED state immediately transitions to
+    /// Diligent::ASYNC_TASK_STATUS_CANCELLED.
+    ///
+    /// To cancel a queued task that has not started, use IThreadPool::RemoveTask().
     VIRTUAL void METHOD(Cancel)(THIS) PURE;
 
     /// Sets the task status, see Diligent::ASYNC_TASK_STATUS.
@@ -103,11 +109,14 @@ DILIGENT_BEGIN_INTERFACE(IAsyncTask, IObject)
     /// Gets the task status, see Diligent::ASYNC_TASK_STATUS.
     VIRTUAL ASYNC_TASK_STATUS METHOD(GetStatus)(THIS) CONST PURE;
 
-    /// Sets the task priorirty.
+    /// Sets the task priority.
+
+    /// NaN priority is invalid. Implementations will report an error and use
+    /// the default priority 0 instead.
     VIRTUAL void METHOD(SetPriority)(THIS_
                                     float fPriority) PURE;
 
-    /// Returns the task priorirty.
+    /// Returns the task priority.
     VIRTUAL float METHOD(GetPriority)(THIS) CONST PURE;
 
     /// Checks if the task is finished (i.e. cancelled or complete).
@@ -175,6 +184,9 @@ DILIGENT_BEGIN_INTERFACE(IThreadPool, IObject)
     ///
     /// Thread pool will keep a strong reference to the task,
     /// so an application is free to release it after enqueuing.
+    /// A task object represents one scheduled execution and must not be
+    /// enqueued again until that execution has finished or the task has been
+    /// removed from the queue.
     ///
     /// The thread pool does not keep strong references to prerequisite tasks.
     /// Prerequisites are tracked weakly; if a prerequisite object expires before
@@ -220,6 +232,11 @@ DILIGENT_BEGIN_INTERFACE(IThreadPool, IObject)
     ///
     /// \return    true if the task was successfully removed from the queue,
     ///            and false otherwise.
+    ///
+    /// If the task is found in the queue, it is marked as
+    /// Diligent::ASYNC_TASK_STATUS_CANCELLED. This unblocks callers waiting in
+    /// IAsyncTask::WaitForCompletion(). Running tasks are not removed; call
+    /// IAsyncTask::Cancel() to request cooperative cancellation of a running task.
     VIRTUAL bool METHOD(RemoveTask)(THIS_
                                     IAsyncTask* pTask) PURE;
 
@@ -230,6 +247,13 @@ DILIGENT_BEGIN_INTERFACE(IThreadPool, IObject)
     /// tasks in the quque are finished and the queue is empty.
     /// An application is responsible to make sure that all tasks
     /// will finish eventually.
+    ///
+    /// \warning This method must not be called from a task running in this
+    ///          thread pool: the task is counted as running until it returns,
+    ///          while this method waits for all running tasks to finish.
+    ///
+    /// \warning Deadlock may also occur if all worker threads block waiting for
+    ///          work that requires those same worker threads to make progress.
     VIRTUAL void METHOD(WaitForAllTasks)(THIS) PURE;
 
 
@@ -247,6 +271,9 @@ DILIGENT_BEGIN_INTERFACE(IThreadPool, IObject)
     /// threads exit, so this is a graceful drain-and-stop operation rather than
     /// an immediate cancellation of queued work.
     ///
+    /// \warning This method must not be called from a worker thread of this
+    ///          pool because it joins all worker threads, including the caller.
+    ///
     /// Enqueuing tasks after calling this method is an error.
     VIRTUAL void METHOD(StopThreads)(THIS) PURE;
 

Common/interface/ThreadPool.hpp

@@ -32,6 +32,7 @@
 #include "ThreadPool.h"
 
 #include <atomic>
+#include <cmath>
 #include <condition_variable>
 #include <functional>
 #include <mutex>
@@ -87,7 +88,7 @@ class AsyncTaskBase : public ObjectBase<IAsyncTask>
     explicit AsyncTaskBase(IReferenceCounters* pRefCounters,
                            float               fPriority = 0) noexcept :
         TBase{pRefCounters},
-        m_fPriority{fPriority}
+        m_fPriority{ValidatePriority(fPriority)}
     {
     }
     virtual ~AsyncTaskBase() = 0;
@@ -156,7 +157,7 @@ class AsyncTaskBase : public ObjectBase<IAsyncTask>
 
     virtual void DILIGENT_CALL_TYPE SetPriority(float fPriority) override final
     {
-        m_fPriority.store(fPriority);
+        m_fPriority.store(ValidatePriority(fPriority));
     }
 
     virtual float DILIGENT_CALL_TYPE GetPriority() const override final
@@ -195,6 +196,16 @@ class AsyncTaskBase : public ObjectBase<IAsyncTask>
     std::atomic<bool> m_bSafelyCancel{false};
 
 private:
+    static float ValidatePriority(float fPriority) noexcept
+    {
+        if (std::isnan(fPriority))
+        {
+            LOG_ERROR_MESSAGE("Task priority must not be NaN. Using the default priority 0.");
+            return 0;
+        }
+        return fPriority;
+    }
+
     mutable std::mutex              m_StatusMtx;
     mutable std::condition_variable m_StatusChangedCond;
     std::atomic<float>              m_fPriority{0};

Common/src/ThreadPool.cpp

@@ -28,6 +28,7 @@
 
 #include <algorithm>
 #include <mutex>
+#include <string>
 #include <thread>
 #include <map>
 #include <vector>
@@ -411,7 +412,7 @@ Uint64 PinWorkerThread(Uint32 ThreadId, Uint64 AllowedCoresMask)
     VERIFY_EXPR(AffinityMask != 0);
     Uint32 WorkerCore = PlatformMisc::GetLSB(AffinityMask);
     VERIFY_EXPR(WorkerCore < NumCores);
-    Uint64 PrevMask = PlatformMisc::SetCurrentThreadAffinity(Uint64{1} << WorkerCore) != 0;
+    Uint64 PrevMask = PlatformMisc::SetCurrentThreadAffinity(Uint64{1} << WorkerCore);
     if (PrevMask == 0)
     {
         LOG_WARNING_MESSAGE("Failed to pin worker thread ", ThreadId, " to core ", WorkerCore);

Tests/DiligentCoreTest/src/Common/ThreadPoolTest.cpp

@@ -32,6 +32,7 @@
 #include <chrono>
 #include <cmath>
 #include <future>
+#include <limits>
 #include <memory>
 #include <stdexcept>
 #include <thread>
@@ -426,6 +427,33 @@ TEST(Common_ThreadPool, WaitUntilRunningNotifiesAllWaitersWhenCancelled)
         "WaitUntilRunning did not wake all waiters when the task was cancelled");
 }
 
+TEST(Common_ThreadPool, NaNTaskPriorityUsesDefault)
+{
+    const float NaN = std::numeric_limits<float>::quiet_NaN();
+
+    // Expected behavior: NaN priority supplied at construction is rejected
+    // because NaN cannot be used as a strict weak ordering key in the queue.
+    {
+        Testing::TestingEnvironment::ErrorScope ExpectedErrors{"Task priority must not be NaN"};
+        RefCntAutoPtr<DummyTask>                pTask;
+        pTask = MakeNewRCObj<DummyTask>()(NaN);
+        EXPECT_EQ(pTask->GetPriority(), 0.f);
+    }
+
+    RefCntAutoPtr<DummyTask> pTask;
+    pTask = MakeNewRCObj<DummyTask>()();
+    pTask->SetPriority(10.f);
+    EXPECT_EQ(pTask->GetPriority(), 10.f);
+
+    // Expected behavior: NaN priority supplied through SetPriority() is
+    // rejected and the task falls back to the default priority.
+    {
+        Testing::TestingEnvironment::ErrorScope ExpectedErrors{"Task priority must not be NaN"};
+        pTask->SetPriority(NaN);
+    }
+    EXPECT_EQ(pTask->GetPriority(), 0.f);
+}
+
 TEST(Common_ThreadPool, RemoveTask)
 {
     constexpr Uint32 NumThreads = 4;