vtkThreadedCallbackQueue.h

Go to the documentation of this file.

// SPDX-FileCopyrightText: Copyright (c) Ken Martin, Will Schroeder, Bill Lorensen
// SPDX-License-Identifier: BSD-3-Clause
 
#ifndef vtkThreadedCallbackQueue_h
#define vtkThreadedCallbackQueue_h
 
#include "vtkCommonCoreModule.h" // For export macro
#include "vtkObject.h"
#include "vtkSmartPointer.h" // For vtkSmartPointer
 
#include <atomic>             // For atomic_bool
#include <condition_variable> // For condition variable
#include <deque>              // For deque
#include <functional>         // For greater
#include <memory>             // For unique_ptr
#include <mutex>              // For mutex
#include <thread>             // For thread
#include <unordered_map>      // For unordered_map
#include <unordered_set>      // For unordered_set
#include <vector>             // For vector
 
#if !defined(__WRAP__)
 
VTK_ABI_NAMESPACE_BEGIN
 
class VTKCOMMONCORE_EXPORT vtkThreadedCallbackQueue : public vtkObject
{
private:
  template <class FT>
  struct Signature;

  template <class T, class DummyT = std::nullptr_t>
  struct Dereference
  {
    struct Type;
  };

  template <class T>
  using DereferencedType = typename std::decay<typename Dereference<T>::Type>::type;

  template <class FT>
  using InvokeResult = typename Signature<DereferencedType<FT>>::InvokeResult;

  template <class ReturnT, bool IsLValueReference = std::is_lvalue_reference<ReturnT>::value>
  class ReturnValueWrapper
  {
    class ReturnLValueRef;
    class ReturnConstLValueRef;
  };
 
public:
  static vtkThreadedCallbackQueue* New();
  vtkTypeMacro(vtkThreadedCallbackQueue, vtkObject);
  void PrintSelf(ostream& os, vtkIndent indent) override;
 
  vtkThreadedCallbackQueue();

  ~vtkThreadedCallbackQueue() override;

  class vtkSharedFutureBase : public vtkObjectBase
  {
  public:
    vtkBaseTypeMacro(vtkSharedFutureBase, vtkObjectBase);
 
    vtkSharedFutureBase()
      : NumberOfPriorSharedFuturesRemaining(0)
      , Status(CONSTRUCTING)
    {
    }

    virtual void Wait() const
    {
      if (this->Status == READY)
      {
        return;
      }
      std::unique_lock<std::mutex> lock(this->Mutex);
      if (this->Status != READY)
      {
        this->ConditionVariable.wait(lock, [this] { return this->Status == READY; });
      }
    }
 
    friend class vtkThreadedCallbackQueue;
 
  private:
    virtual void operator()() = 0;

    std::atomic_int NumberOfPriorSharedFuturesRemaining;

    std::atomic_int Status;

    vtkIdType InvokerIndex;

    bool IsHighPriority = false;

    std::vector<vtkSmartPointer<vtkSharedFutureBase>> Dependents;
 
    mutable std::mutex Mutex;
    mutable std::condition_variable ConditionVariable;
 
    vtkSharedFutureBase(const vtkSharedFutureBase& other) = delete;
    void operator=(const vtkSharedFutureBase& other) = delete;
  };

  template <class ReturnT>
  class vtkSharedFuture : public vtkSharedFutureBase
  {
  public:
    vtkAbstractTypeMacro(vtkSharedFuture<ReturnT>, vtkSharedFutureBase);
 
    using ReturnLValueRef = typename ReturnValueWrapper<ReturnT>::ReturnLValueRef;
    using ReturnConstLValueRef = typename ReturnValueWrapper<ReturnT>::ReturnConstLValueRef;
 
    vtkSharedFuture() = default;

    ReturnLValueRef Get();

    ReturnConstLValueRef Get() const;
 
    friend class vtkThreadedCallbackQueue;
 
  private:
    ReturnValueWrapper<ReturnT> ReturnValue;
 
    vtkSharedFuture(const vtkSharedFuture<ReturnT>& other) = delete;
    void operator=(const vtkSharedFuture<ReturnT>& other) = delete;
  };
 
  using SharedFutureBasePointer = vtkSmartPointer<vtkSharedFutureBase>;
  template <class ReturnT>
  using SharedFuturePointer = vtkSmartPointer<vtkSharedFuture<ReturnT>>;

  template <class FT, class... ArgsT>
  SharedFuturePointer<InvokeResult<FT>> Push(FT&& f, ArgsT&&... args);

  template <class SharedFutureContainerT, class FT, class... ArgsT>
  SharedFuturePointer<InvokeResult<FT>> PushDependent(
    SharedFutureContainerT&& priorSharedFutures, FT&& f, ArgsT&&... args);

  template <class SharedFutureContainerT>
  void Wait(SharedFutureContainerT&& priorSharedFuture);


  template <class ReturnT>
  typename vtkSharedFuture<ReturnT>::ReturnLValueRef Get(SharedFuturePointer<ReturnT>& future);
  template <class ReturnT>
  typename vtkSharedFuture<ReturnT>::ReturnConstLValueRef Get(
    const SharedFuturePointer<ReturnT>& future);

  void SetNumberOfThreads(int numberOfThreads);

  int GetNumberOfThreads() const { return this->NumberOfThreads; }
 
private:

  template <class FT, class... ArgsT>
  class vtkInvoker;
 
  struct InvokerImpl;
 
  template <class FT, class... ArgsT>
  using InvokerPointer = vtkSmartPointer<vtkInvoker<FT, ArgsT...>>;
 
  class ThreadWorker;
 
  friend class ThreadWorker;

  enum Status
  {
    CONSTRUCTING = 0x00,

    ON_HOLD = 0x01,

    ENQUEUED = 0x02,

    RUNNING = 0x04,

    READY = 0x08
  };

  void Sync(int startId = 0);

  void PopFrontNullptr();

  void SignalDependentSharedFutures(vtkSharedFutureBase* invoker);

  template <class SharedFutureContainerT, class InvokerT>
  void HandleDependentInvoker(SharedFutureContainerT&& priorSharedFutures, InvokerT&& invoker);

  void Invoke(vtkSharedFutureBase* invoker);

  bool TryInvoke(vtkSharedFutureBase* invoker);

  template <class FT, class... ArgsT>
  void PushControl(FT&& f, ArgsT&&... args);

  template <class SharedFutureContainerT>
  static bool MustWait(SharedFutureContainerT&& priorSharedFutures);

  std::deque<SharedFutureBasePointer> InvokerQueue;

  std::mutex Mutex;

  std::mutex ControlMutex;

  std::mutex DestroyMutex;

  std::mutex ThreadIdToIndexMutex;
 
  std::condition_variable ConditionVariable;

  std::atomic_bool Destroying{ false };

  std::atomic_int NumberOfThreads;
 
  std::vector<std::thread> Threads;

  std::unordered_map<std::thread::id, std::shared_ptr<std::atomic_int>> ThreadIdToIndex;

  std::unordered_set<SharedFutureBasePointer> ControlFutures;
 
  vtkThreadedCallbackQueue(const vtkThreadedCallbackQueue&) = delete;
  void operator=(const vtkThreadedCallbackQueue&) = delete;
};
 
VTK_ABI_NAMESPACE_END
 
#include "vtkThreadedCallbackQueue.txx"
 
#endif
#endif
// VTK-HeaderTest-Exclude: vtkThreadedCallbackQueue.h