scheduler.hpp

Go to the documentation of this file.

//==-------------- scheduler.hpp - SYCL standard header file ---------------==//
//
// Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
// See https://llvm.org/LICENSE.txt for license information.
// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
//
//===----------------------------------------------------------------------===//
 
#pragma once
 
#include <detail/scheduler/commands.hpp>
#include <detail/scheduler/leaves_collection.hpp>
#include <detail/sycl_mem_obj_i.hpp>
#include <sycl/detail/cg.hpp>
 
#include <cstddef>
#include <memory>
#include <queue>
#include <set>
#include <shared_mutex>
#include <unordered_map>
#include <unordered_set>
#include <vector>
 
 
// For testing purposes
class MockScheduler;
 
namespace sycl {
__SYCL_INLINE_VER_NAMESPACE(_V1) {
namespace detail {
class queue_impl;
class event_impl;
class context_impl;
class DispatchHostTask;
 
using ContextImplPtr = std::shared_ptr<detail::context_impl>;
using EventImplPtr = std::shared_ptr<detail::event_impl>;
using QueueImplPtr = std::shared_ptr<detail::queue_impl>;
using StreamImplPtr = std::shared_ptr<detail::stream_impl>;
 
using QueueIdT = std::hash<std::shared_ptr<detail::queue_impl>>::result_type;
using CommandPtr = std::unique_ptr<Command>;
using FusionList = std::unique_ptr<KernelFusionCommand>;
using FusionMap = std::unordered_map<QueueIdT, FusionList>;
 
struct MemObjRecord {
  MemObjRecord(ContextImplPtr Ctx, std::size_t LeafLimit,
               LeavesCollection::AllocateDependencyF AllocateDependency)
      : MReadLeaves{this, LeafLimit, AllocateDependency},
        MWriteLeaves{this, LeafLimit, AllocateDependency}, MCurContext{Ctx} {}
 
  // Contains all allocation commands for the memory object.
  std::vector<AllocaCommandBase *> MAllocaCommands;
 
  // Contains latest read only commands working with memory object.
  LeavesCollection MReadLeaves;
 
  // Contains latest write commands working with memory object.
  LeavesCollection MWriteLeaves;
 
  // The context which has the latest state of the memory object.
  ContextImplPtr MCurContext;
 
  // The mode this object can be accessed with from the host context.
  // Valid only if the current context is host.
  access::mode MHostAccess = access::mode::read_write;
 
  // The flag indicates that the content of the memory object was/will be
  // modified. Used while deciding if copy back needed.
  bool MMemModified = false;
};
 
//
//    +----------+     +----------+     +----------+
//    |          |     |          |     |          |
//    | Allocate |<----| Execute  |<----| Execute  |
//    |          |     |          |     |          |
//    +----------+     +----------+     +----------+
//
//                             +----------+
//                             |          |
//                             |    D     |
//                             |          |
//                             +----------+
//                            /            \
//                           /              \
//                          v                v
//                      +----------+     +----------+
//                      |          |     |          |
//                      |     B    |     |    C     |
//                      |          |     |          |
//                      +----------+     +----------+
//                             \            /
//                              \          /
//                               v        v
//                              +----------+
//                              |          |
//                              |    A     |
//                              |          |
//                              +----------+
class Scheduler {
public:
  EventImplPtr addCG(std::unique_ptr<detail::CG> CommandGroup,
                     const QueueImplPtr &Queue);
 
  EventImplPtr addCopyBack(Requirement *Req);
 
  void waitForEvent(const EventImplPtr &Event);
 
  bool removeMemoryObject(detail::SYCLMemObjI *MemObj, bool StrictLock = true);
 
  EventImplPtr addHostAccessor(Requirement *Req);
 
  void releaseHostAccessor(Requirement *Req);
 
  static Scheduler &getInstance();
 
  QueueImplPtr getDefaultHostQueue() { return DefaultHostQueue; }
 
  const QueueImplPtr &getDefaultHostQueue() const { return DefaultHostQueue; }
 
  static MemObjRecord *getMemObjRecord(const Requirement *const Req);
 
  void deferMemObjRelease(const std::shared_ptr<detail::SYCLMemObjI> &MemObj);
 
  void startFusion(QueueImplPtr Queue);
 
  void cancelFusion(QueueImplPtr Queue);
 
  EventImplPtr completeFusion(QueueImplPtr Queue, const property_list &);
 
  bool isInFusionMode(QueueIdT Queue);
 
  Scheduler();
  ~Scheduler();
  void releaseResources();
  bool isDeferredMemObjectsEmpty();
 
  void enqueueCommandForCG(EventImplPtr NewEvent,
                           std::vector<Command *> &AuxilaryCmds,
                           BlockingT Blocking = NON_BLOCKING);
 
protected:
  using RWLockT = std::shared_timed_mutex;
  using ReadLockT = std::shared_lock<RWLockT>;
  using WriteLockT = std::unique_lock<RWLockT>;
 
  WriteLockT acquireWriteLock() {
#ifdef _WIN32
    WriteLockT Lock(MGraphLock, std::defer_lock);
    while (!Lock.try_lock_for(std::chrono::milliseconds(10))) {
      // Without yield while loop acts like endless while loop and occupies the
      // whole CPU when multiple command groups are created in multiple host
      // threads
      std::this_thread::yield();
    }
#else
    WriteLockT Lock(MGraphLock);
    // It is a deadlock on UNIX in implementation of lock and lock_shared, if
    // try_lock in the loop above will be executed, so using a single lock here
#endif // _WIN32
    return Lock;
  }
 
  ReadLockT acquireReadLock() { return ReadLockT{MGraphLock}; }
 
  void cleanupCommands(const std::vector<Command *> &Cmds);
 
  void NotifyHostTaskCompletion(Command *Cmd);
 
  static void enqueueLeavesOfReqUnlocked(const Requirement *const Req,
                                         ReadLockT &GraphReadLock,
                                         std::vector<Command *> &ToCleanUp);
 
  static void
  enqueueUnblockedCommands(const std::vector<EventImplPtr> &CmdsToEnqueue,
                           ReadLockT &GraphReadLock,
                           std::vector<Command *> &ToCleanUp);
 
  // May lock graph with read and write modes during execution.
  void cleanupDeferredMemObjects(BlockingT Blocking);
 
  // POD struct to convey some additional information from GraphBuilder::addCG
  // to the Scheduler to support kernel fusion.
  struct GraphBuildResult {
    Command *NewCmd;
    EventImplPtr NewEvent;
    bool ShouldEnqueue;
  };
 
  void registerAuxiliaryResources(
      EventImplPtr &Event, std::vector<std::shared_ptr<const void>> Resources);
  void cleanupAuxiliaryResources(BlockingT Blocking);
 
  class GraphBuilder {
  public:
    GraphBuilder();
 
    GraphBuildResult addCG(std::unique_ptr<detail::CG> CommandGroup,
                           const QueueImplPtr &Queue,
                           std::vector<Command *> &ToEnqueue);
 
    Command *addCGUpdateHost(std::unique_ptr<detail::CG> CommandGroup,
                             const QueueImplPtr &HostQueue,
                             std::vector<Command *> &ToEnqueue);
 
    Command *addCopyBack(Requirement *Req, std::vector<Command *> &ToEnqueue);
 
    Command *addHostAccessor(Requirement *Req,
                             std::vector<Command *> &ToEnqueue);
 
    void optimize();
 
    void optimize(const EventImplPtr &Event);
 
    void cleanupCommand(Command *Cmd, bool AllowUnsubmitted = false);
 
    void rescheduleCommand(Command *Cmd, const QueueImplPtr &Queue);
 
    MemObjRecord *getMemObjRecord(SYCLMemObjI *MemObject);
 
    MemObjRecord *getOrInsertMemObjRecord(const QueueImplPtr &Queue,
                                          const Requirement *Req,
                                          std::vector<Command *> &ToEnqueue);
 
    void decrementLeafCountersForRecord(MemObjRecord *Record);
 
    void cleanupCommandsForRecord(MemObjRecord *Record);
 
    void removeRecordForMemObj(SYCLMemObjI *MemObject);
 
    void addNodeToLeaves(MemObjRecord *Record, Command *Cmd,
                         access::mode AccessMode,
                         std::vector<Command *> &ToEnqueue);
 
    void updateLeaves(const std::set<Command *> &Cmds, MemObjRecord *Record,
                      access::mode AccessMode,
                      std::vector<Command *> &ToCleanUp);
 
    Command *connectDepEvent(Command *const Cmd, const EventImplPtr &DepEvent,
                             const DepDesc &Dep,
                             std::vector<Command *> &ToCleanUp);
 
    void startFusion(QueueImplPtr Queue);
 
    void cancelFusion(QueueImplPtr Queue, std::vector<Command *> &ToEnqueue);
 
    EventImplPtr completeFusion(QueueImplPtr Queue,
                                std::vector<Command *> &ToEnqueue,
                                const property_list &);
 
    bool isInFusionMode(QueueIdT queue);
 
    std::vector<SYCLMemObjI *> MMemObjs;
 
  private:
    Command *insertMemoryMove(MemObjRecord *Record, Requirement *Req,
                              const QueueImplPtr &Queue,
                              std::vector<Command *> &ToEnqueue);
 
    // Inserts commands required to remap the memory object to its current host
    // context so that the required access mode becomes valid.
    Command *remapMemoryObject(MemObjRecord *Record, Requirement *Req,
                               AllocaCommandBase *HostAllocaCmd,
                               std::vector<Command *> &ToEnqueue);
 
    UpdateHostRequirementCommand *
    insertUpdateHostReqCmd(MemObjRecord *Record, Requirement *Req,
                           const QueueImplPtr &Queue,
                           std::vector<Command *> &ToEnqueue);
 
    std::set<Command *> findDepsForReq(MemObjRecord *Record,
                                       const Requirement *Req,
                                       const ContextImplPtr &Context);
 
    EmptyCommand *addEmptyCmd(Command *Cmd,
                              const std::vector<Requirement *> &Req,
                              const QueueImplPtr &Queue,
                              Command::BlockReason Reason,
                              std::vector<Command *> &ToEnqueue,
                              const bool AddDepsToLeaves = true);
 
    void createGraphForCommand(Command *NewCmd, CG &CG, bool isInteropTask,
                               std::vector<Requirement *> &Reqs,
                               const std::vector<detail::EventImplPtr> &Events,
                               QueueImplPtr Queue,
                               std::vector<Command *> &ToEnqueue);
 
  protected:
    DepDesc findDepForRecord(Command *Cmd, MemObjRecord *Record);
 
    AllocaCommandBase *findAllocaForReq(MemObjRecord *Record,
                                        const Requirement *Req,
                                        const ContextImplPtr &Context,
                                        bool AllowConst = true);
 
    friend class Command;
 
  private:
    friend class ::MockScheduler;
 
    AllocaCommandBase *
    getOrCreateAllocaForReq(MemObjRecord *Record, const Requirement *Req,
                            const QueueImplPtr &Queue,
                            std::vector<Command *> &ToEnqueue);
 
    void markModifiedIfWrite(MemObjRecord *Record, Requirement *Req);
 
    FusionMap::iterator findFusionList(QueueIdT Id) {
      return MFusionMap.find(Id);
    }
 
    void removeNodeFromGraph(Command *Node, std::vector<Command *> &ToEnqueue);
 
    std::queue<Command *> MCmdsToVisit;
    std::vector<Command *> MVisitedCmds;
 
    FusionMap MFusionMap;
 
    void printGraphAsDot(const char *ModeName);
    enum PrintOptions {
      BeforeAddCG = 0,
      AfterAddCG,
      BeforeAddCopyBack,
      AfterAddCopyBack,
      BeforeAddHostAcc,
      AfterAddHostAcc,
      AfterFusionComplete,
      AfterFusionCancel,
      Size
    };
    std::array<bool, PrintOptions::Size> MPrintOptionsArray{false};
  };
 
  class GraphProcessor {
  public:
    static void waitForEvent(const EventImplPtr &Event,
                             ReadLockT &GraphReadLock,
                             std::vector<Command *> &ToCleanUp,
                             bool LockTheLock = true);
 
    static bool enqueueCommand(Command *Cmd, ReadLockT &GraphReadLock,
                               EnqueueResultT &EnqueueResult,
                               std::vector<Command *> &ToCleanUp,
                               Command *RootCommand,
                               BlockingT Blocking = NON_BLOCKING);
 
    static bool handleBlockingCmd(Command *Cmd, EnqueueResultT &EnqueueResult,
                                  Command *RootCommand, BlockingT Blocking);
  };
 
  void waitForRecordToFinish(MemObjRecord *Record, ReadLockT &GraphReadLock);
  bool checkLeavesCompletion(MemObjRecord *Record);
 
  GraphBuilder MGraphBuilder;
  RWLockT MGraphLock;
 
  std::vector<Command *> MDeferredCleanupCommands;
  std::mutex MDeferredCleanupMutex;
 
  std::vector<std::shared_ptr<SYCLMemObjI>> MDeferredMemObjRelease;
  std::mutex MDeferredMemReleaseMutex;
 
  std::unordered_map<EventImplPtr, std::vector<std::shared_ptr<const void>>>
      MAuxiliaryResources;
  std::mutex MAuxiliaryResourcesMutex;
 
  QueueImplPtr DefaultHostQueue;
 
  // This thread local flag is a workaround for a problem with managing
  // auxiliary resources. We would like to release internal buffers used for
  // reductions in a deferred manner, but marking them individually isn't an
  // option since all auxiliary resources (buffers, host memory, USM) are passed
  // to the library as type erased shared pointers. This flag makes it so that
  // release of every memory object is deferred while it's set, and it should
  // only be set during release of auxiliary resources.
  // TODO Remove once ABI breaking changes are allowed.
  friend class SYCLMemObjT;
  static thread_local bool ForceDeferredMemObjRelease;
  struct ForceDeferredReleaseWrapper {
    ForceDeferredReleaseWrapper() { ForceDeferredMemObjRelease = true; };
    ~ForceDeferredReleaseWrapper() { ForceDeferredMemObjRelease = false; };
  };
 
  friend class Command;
  friend class DispatchHostTask;
  friend class queue_impl;
  friend class event_impl;
  friend class ::MockScheduler;
 
private:
  static void printFusionWarning(const std::string &Message);
 
  static KernelFusionCommand *isPartOfActiveFusion(Command *Cmd);
};
 
} // namespace detail
} // __SYCL_INLINE_VER_NAMESPACE(_V1)
} // namespace sycl