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 QueueImplPtr = std::shared_ptr<detail::queue_impl>;
using EventImplPtr = std::shared_ptr<detail::event_impl>;
using ContextImplPtr = std::shared_ptr<detail::context_impl>;
 
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);
 
  void removeMemoryObject(detail::SYCLMemObjI *MemObj);
 
  void cleanupFinishedCommands(const EventImplPtr &FinishedEvent);
 
  EventImplPtr addHostAccessor(Requirement *Req);
 
  void releaseHostAccessor(Requirement *Req);
 
  static Scheduler &getInstance();
 
  void allocateStreamBuffers(stream_impl *, size_t, size_t);
 
  void deallocateStreamBuffers(stream_impl *);
 
  QueueImplPtr getDefaultHostQueue() { return DefaultHostQueue; }
 
  const QueueImplPtr &getDefaultHostQueue() const { return DefaultHostQueue; }
 
  static MemObjRecord *getMemObjRecord(const Requirement *const Req);
 
  Scheduler();
  ~Scheduler();
 
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, Command *BlockingCmd);
 
  static void enqueueLeavesOfReqUnlocked(const Requirement *const Req,
                                         std::vector<Command *> &ToCleanUp);
 
  class GraphBuilder {
  public:
    GraphBuilder();
 
    Command *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);
 
    void cleanupFinishedCommands(
        Command *FinishedCmd,
        std::vector<std::shared_ptr<sycl::detail::stream_impl>> &,
        std::vector<std::shared_ptr<const void>> &);
 
    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,
        std::vector<std::shared_ptr<sycl::detail::stream_impl>> &,
        std::vector<std::shared_ptr<const void>> &);
 
    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);
 
    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);
 
    template <typename T>
    typename std::enable_if_t<
        std::is_same<typename std::remove_cv_t<T>, Requirement>::value,
        EmptyCommand *>
    addEmptyCmd(Command *Cmd, const std::vector<T *> &Req,
                const QueueImplPtr &Queue, Command::BlockReason Reason,
                std::vector<Command *> &ToEnqueue,
                const bool AddDepsToLeaves = true);
 
  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);
 
    std::queue<Command *> MCmdsToVisit;
    std::vector<Command *> MVisitedCmds;
    void printGraphAsDot(const char *ModeName);
    enum PrintOptions {
      BeforeAddCG = 0,
      AfterAddCG,
      BeforeAddCopyBack,
      AfterAddCopyBack,
      BeforeAddHostAcc,
      AfterAddHostAcc,
      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, EnqueueResultT &EnqueueResult,
                               std::vector<Command *> &ToCleanUp,
                               BlockingT Blocking = NON_BLOCKING);
  };
 
  void waitForRecordToFinish(MemObjRecord *Record, ReadLockT &GraphReadLock);
 
  GraphBuilder MGraphBuilder;
  RWLockT MGraphLock;
 
  std::vector<Command *> MDeferredCleanupCommands;
  std::mutex MDeferredCleanupMutex;
 
  QueueImplPtr DefaultHostQueue;
 
  friend class Command;
  friend class DispatchHostTask;
  friend class queue_impl;
  friend class event_impl;
 
  struct StreamBuffers {
    StreamBuffers(size_t StreamBufferSize, size_t FlushBufferSize)
        // Initialize stream buffer with zeros, this is needed for two reasons:
        // 1. We don't need to care about end of line when printing out
        // streamed data.
        // 2. Offset is properly initialized.
        : Data(StreamBufferSize, 0),
          Buf(Data.data(), range<1>(StreamBufferSize),
              {property::buffer::use_host_ptr()}),
          FlushBuf(range<1>(FlushBufferSize)) {
      // Disable copy back on buffer destruction. Copy is scheduled as a host
      // task which fires up as soon as kernel has completed exectuion.
      Buf.set_write_back(false);
      FlushBuf.set_write_back(false);
    }
 
    // Vector on the host side which is used to initialize the stream
    // buffer
    std::vector<char> Data;
 
    // Stream buffer
    buffer<char, 1> Buf;
 
    // Global flush buffer
    buffer<char, 1> FlushBuf;
  };
 
  friend class stream_impl;
  friend void initStream(StreamImplPtr, QueueImplPtr);
 
  // Protects stream buffers pool
  std::recursive_mutex StreamBuffersPoolMutex;
 
  // We need to store a pointer to the structure with stream buffers because we
  // want to avoid a situation when buffers are destructed during destruction of
  // the scheduler. Scheduler is a global object and it can be destructed after
  // all device runtimes are unloaded. Destruction of the buffers at this stage
  // will lead to a faliure. In the correct program there will be sync points
  // for all kernels and all allocated resources will be released by the
  // scheduler. If program is not correct and doesn't have necessary sync point
  // then warning will be issued.
  std::unordered_map<stream_impl *, StreamBuffers *> StreamBuffersPool;
};
 
} // namespace detail
} // __SYCL_INLINE_VER_NAMESPACE(_V1)
} // namespace sycl