[SYCL] Defer buffer release when no host memory to be updated

sycl/CMakeLists.txt

@@ -28,7 +28,7 @@ include(SYCLUtils)
 # The change in SYCL_MAJOR_VERSION must be accompanied with the same update in
 # llvm/clang/lib/Driver/CMakeLists.txt.
 set(SYCL_MAJOR_VERSION 6)
-set(SYCL_MINOR_VERSION 0)
+set(SYCL_MINOR_VERSION 1)
 set(SYCL_PATCH_VERSION 0)
 set(SYCL_DEV_ABI_VERSION 0)
 if (SYCL_ADD_DEV_VERSION_POSTFIX)

sycl/include/sycl/buffer.hpp

@@ -117,6 +117,8 @@ class __SYCL_EXPORT buffer_plain {
 
   size_t getSize() const;
 
+  void handleRelease(bool DefaultAllocator) const;
+
   std::shared_ptr<detail::buffer_impl> impl;
 };
 
@@ -457,7 +459,11 @@ class buffer : public detail::buffer_plain {
 
   buffer &operator=(buffer &&rhs) = default;
 
-  ~buffer() = default;
+  ~buffer() {
+    buffer_plain::handleRelease(
+        std::is_same<AllocatorT,
+                     detail::sycl_memory_object_allocator<T>>::value);
+  }
 
   bool operator==(const buffer &rhs) const { return impl == rhs.impl; }
 

sycl/include/sycl/detail/sycl_mem_obj_allocator.hpp

@@ -60,7 +60,6 @@ class SYCLMemObjAllocatorHolder : public SYCLMemObjAllocator {
   }
 
   virtual std::size_t getValueSize() const override { return MValueSize; }
-
 protected:
   virtual void *getAllocatorImpl() override { return &MAllocator; }
 

sycl/source/buffer.cpp

@@ -121,6 +121,11 @@ void buffer_plain::addOrReplaceAccessorProperties(
 
 size_t buffer_plain::getSize() const { return impl->getSizeInBytes(); }
 
+void buffer_plain::handleRelease(bool DefaultAllocator) const {
+  if (impl.use_count() == 1)
+    impl->detachMemoryObject(impl, DefaultAllocator);
+}
+
 } // namespace detail
 } // __SYCL_INLINE_VER_NAMESPACE(_V1)
 } // namespace sycl

sycl/source/detail/event_impl.cpp

@@ -443,6 +443,11 @@ void event_impl::cleanDepEventsThroughOneLevel() {
   }
 }
 
+bool event_impl::isCompleted() {
+  return get_info<info::event::command_execution_status>() ==
+         info::event_command_status::complete;
+}
+
 } // namespace detail
 } // __SYCL_INLINE_VER_NAMESPACE(_V1)
 } // namespace sycl

sycl/source/detail/event_impl.hpp

@@ -234,6 +234,8 @@ class event_impl {
   /// state.
   bool isInitialized() const noexcept { return MIsInitialized; }
 
+  bool isCompleted();
+
 private:
   // When instrumentation is enabled emits trace event for event wait begin and
   // returns the telemetry event generated for the wait

sycl/source/detail/global_handler.cpp

@@ -47,6 +47,15 @@ T &GlobalHandler::getOrCreate(InstWithLock<T> &IWL, Types... Args) {
   return *IWL.Inst;
 }
 
+void GlobalHandler::attachScheduler(Scheduler *scheduler) {
+  // Test method, do not protect with lock since releaseResources will cause dead lock due to host queue release
+  // const LockGuard Lock{MScheduler.Lock};
+  if (MScheduler.Inst)
+    MScheduler.Inst->releaseResources();
+
+  MScheduler.Inst.reset(scheduler);
+}
+
 Scheduler &GlobalHandler::getScheduler() { return getOrCreate(MScheduler); }
 
 ProgramManager &GlobalHandler::getProgramManager() {
@@ -151,6 +160,8 @@ void shutdown() {
 
   // First, release resources, that may access plugins.
   GlobalHandler::instance().MPlatformCache.Inst.reset(nullptr);
+  if (GlobalHandler::instance().MScheduler.Inst)
+    GlobalHandler::instance().MScheduler.Inst->releaseResources();
   GlobalHandler::instance().MScheduler.Inst.reset(nullptr);
   GlobalHandler::instance().MProgramManager.Inst.reset(nullptr);
 

sycl/source/detail/global_handler.hpp

@@ -73,6 +73,9 @@ class GlobalHandler {
 
   void unloadPlugins();
 
+  // For testing purposes only
+  void attachScheduler(Scheduler *scheduler);
+
 private:
   friend void releaseDefaultContexts();
   friend void shutdown();

sycl/source/detail/scheduler/scheduler.cpp

@@ -26,51 +26,73 @@ namespace sycl {
 __SYCL_INLINE_VER_NAMESPACE(_V1) {
 namespace detail {
 
-void Scheduler::waitForRecordToFinish(MemObjRecord *Record,
-                                      ReadLockT &GraphReadLock) {
-#ifdef XPTI_ENABLE_INSTRUMENTATION
-  // Will contain the list of dependencies for the Release Command
-  std::set<Command *> DepCommands;
-#endif
+bool Scheduler::waitForRecordToFinish(MemObjRecord *Record,
+                                      ReadLockT &GraphReadLock,
+                                      bool ForceWait) {
+  assert(Record);
   std::vector<Command *> ToCleanUp;
   for (Command *Cmd : Record->MReadLeaves) {
+    if (Cmd->getEvent()->isCompleted())
+      continue;
+
     EnqueueResultT Res;
     bool Enqueued = GraphProcessor::enqueueCommand(Cmd, Res, ToCleanUp);
     if (!Enqueued && EnqueueResultT::SyclEnqueueFailed == Res.MResult)
       throw runtime_error("Enqueue process failed.",
                           PI_ERROR_INVALID_OPERATION);
-#ifdef XPTI_ENABLE_INSTRUMENTATION
-    // Capture the dependencies
-    DepCommands.insert(Cmd);
-#endif
-    GraphProcessor::waitForEvent(Cmd->getEvent(), GraphReadLock, ToCleanUp);
+    if (ForceWait) {
+      GraphProcessor::waitForEvent(Cmd->getEvent(), GraphReadLock, ToCleanUp);
+    } else
+      return false;
   }
   for (Command *Cmd : Record->MWriteLeaves) {
+    if (Cmd->getEvent()->isCompleted())
+      continue;
+
     EnqueueResultT Res;
     bool Enqueued = GraphProcessor::enqueueCommand(Cmd, Res, ToCleanUp);
     if (!Enqueued && EnqueueResultT::SyclEnqueueFailed == Res.MResult)
       throw runtime_error("Enqueue process failed.",
                           PI_ERROR_INVALID_OPERATION);
-#ifdef XPTI_ENABLE_INSTRUMENTATION
-    DepCommands.insert(Cmd);
-#endif
-    GraphProcessor::waitForEvent(Cmd->getEvent(), GraphReadLock, ToCleanUp);
+    if (ForceWait) {
+      GraphProcessor::waitForEvent(Cmd->getEvent(), GraphReadLock, ToCleanUp);
+    } else
+      return false;
   }
+  // all dependencies is completed and we can enqueue all ReleaseCmds in advance
   for (AllocaCommandBase *AllocaCmd : Record->MAllocaCommands) {
     Command *ReleaseCmd = AllocaCmd->getReleaseCmd();
-    EnqueueResultT Res;
-    bool Enqueued = GraphProcessor::enqueueCommand(ReleaseCmd, Res, ToCleanUp);
-    if (!Enqueued && EnqueueResultT::SyclEnqueueFailed == Res.MResult)
-      throw runtime_error("Enqueue process failed.",
-                          PI_ERROR_INVALID_OPERATION);
+    if (ReleaseCmd->isSuccessfullyEnqueued())
+      continue;
 #ifdef XPTI_ENABLE_INSTRUMENTATION
+    // Will contain the list of dependencies for the Release Command
+    std::set<Command *> DepCommands;
+    // Capture the read dependencies
+    for (Command *Cmd : Record->MWriteLeaves)
+      DepCommands.insert(Cmd);
+    for (Command *Cmd : Record->MReadLeaves)
+      DepCommands.insert(Cmd);
     // Report these dependencies to the Command so these dependencies can be
     // reported as edges
     ReleaseCmd->resolveReleaseDependencies(DepCommands);
 #endif
-    GraphProcessor::waitForEvent(ReleaseCmd->getEvent(), GraphReadLock,
-                                 ToCleanUp);
+    EnqueueResultT Res;
+    bool Enqueued = GraphProcessor::enqueueCommand(ReleaseCmd, Res, ToCleanUp);
+    if (!Enqueued && EnqueueResultT::SyclEnqueueFailed == Res.MResult)
+      throw runtime_error("Enqueue process failed.",
+                          PI_ERROR_INVALID_OPERATION);
   }
+  // enqueue is fully done and we can check if ReleaseCmd is completed
+  for (AllocaCommandBase *AllocaCmd : Record->MAllocaCommands) {
+    Command *Cmd = AllocaCmd->getReleaseCmd();
+    if (Cmd->getEvent()->isCompleted())
+      continue;
+    if (ForceWait)
+      GraphProcessor::waitForEvent(Cmd->getEvent(), GraphReadLock, ToCleanUp);
+    else
+      return false;
+  }
+  return true;
 }
 
 EventImplPtr Scheduler::addCG(std::unique_ptr<detail::CG> CommandGroup,
@@ -258,6 +280,18 @@ void Scheduler::cleanupFinishedCommands(EventImplPtr FinishedEvent) {
   deallocateStreams(StreamsToDeallocate);
 }
 
+inline void Scheduler::releaseMemObjRecord(
+    detail::SYCLMemObjI *MemObj,
+    std::vector<std::shared_ptr<stream_impl>> &StreamsToDeallocate,
+    std::vector<std::shared_ptr<const void>> &AuxResourcesToDeallocate) {
+  MemObjRecord *Record = MGraphBuilder.getMemObjRecord(MemObj);
+  assert(Record);
+  MGraphBuilder.decrementLeafCountersForRecord(Record);
+  MGraphBuilder.cleanupCommandsForRecord(Record, StreamsToDeallocate,
+                                         AuxResourcesToDeallocate);
+  MGraphBuilder.removeRecordForMemObj(MemObj);
+}
+
 void Scheduler::removeMemoryObject(detail::SYCLMemObjI *MemObj) {
   // We are going to traverse a graph of finished commands. Gather stream
   // objects from these commands if any and deallocate buffers for these stream
@@ -283,16 +317,14 @@ void Scheduler::removeMemoryObject(detail::SYCLMemObjI *MemObj) {
         // No operations were performed on the mem object
         return;
 
-      waitForRecordToFinish(Record, Lock);
+      waitForRecordToFinish(Record, Lock, true);
     }
 
     {
       WriteLockT Lock(MGraphLock, std::defer_lock);
       acquireWriteLock(Lock);
-      MGraphBuilder.decrementLeafCountersForRecord(Record);
-      MGraphBuilder.cleanupCommandsForRecord(Record, StreamsToDeallocate,
-                                             AuxResourcesToDeallocate);
-      MGraphBuilder.removeRecordForMemObj(MemObj);
+      releaseMemObjRecord(MemObj, StreamsToDeallocate,
+                          AuxResourcesToDeallocate);
     }
   }
   deallocateStreams(StreamsToDeallocate);
@@ -395,6 +427,16 @@ Scheduler::Scheduler() {
 }
 
 Scheduler::~Scheduler() {
+  // Please be aware that releaseResources should be called before deletion of
+  // Scheduler. Otherwise there can be the case when objects Scheduler keeps as
+  // fields may need Scheduler for their release and they work with Scheduler
+  // via GlobalHandler::getScheduler that will create new Scheduler object.
+  // Still keep it here but it should no almost nothing if releaseResources
+  // called before.
+  releaseResources();
+}
+
+void Scheduler::releaseResources() {
   // By specification there are several possible sync points: buffer
   // destruction, wait() method of a queue or event. Stream doesn't introduce
   // any synchronization point. It is guaranteed that stream is flushed and
@@ -414,6 +456,17 @@ Scheduler::~Scheduler() {
   // haven't been freed because of the graph mutex being locked at the time,
   // clean them up now.
   cleanupCommands({});
+  DefaultHostQueue.reset();
+
+  // We need loop since sometimes we may need new objects to be added to
+  // deferred mem objects storage during cleanup. Known example is: we cleanup
+  // existing deferred mem objects under write lock, during this process we
+  // cleanup commands related to this record, command may have last reference to
+  // queue_impl, ~queue_impl is called and buffer for assert (which is created
+  // with size only so all confitions for deferred release are satisfied) is
+  // added to deferred mem obj storage. So we may end up with leak.
+  while (!isNoDeferredMemObjects())
+    cleanupDeferredMemObjects(true);
 }
 
 void Scheduler::acquireWriteLock(WriteLockT &Lock) {
@@ -442,8 +495,8 @@ MemObjRecord *Scheduler::getMemObjRecord(const Requirement *const Req) {
 }
 
 void Scheduler::cleanupCommands(const std::vector<Command *> &Cmds) {
-  if (Cmds.empty())
-  {
+  cleanupDeferredMemObjects(false);
+  if (Cmds.empty()) {
     std::lock_guard<std::mutex> Lock{MDeferredCleanupMutex};
     if (MDeferredCleanupCommands.empty())
       return;
@@ -472,6 +525,75 @@ void Scheduler::cleanupCommands(const std::vector<Command *> &Cmds) {
   }
 }
 
+void Scheduler::deferMemObjRelease(const std::shared_ptr<SYCLMemObjI> &MemObj) {
+  {
+    std::lock_guard<std::mutex> Lock{MDeferredMemReleaseMutex};
+    MDeferredMemObjRelease.push_back(MemObj);
+  }
+  cleanupDeferredMemObjects(false);
+}
+
+inline bool Scheduler::isNoDeferredMemObjects() {
+  std::lock_guard<std::mutex> Lock{MDeferredMemReleaseMutex};
+  return MDeferredMemObjRelease.empty();
+}
+
+void Scheduler::cleanupDeferredMemObjects(bool ForceWait) {
+  if (isNoDeferredMemObjects())
+    return;
+
+  // Need to aggregate ready to release object to acquire write lock once.
+  std::list<std::shared_ptr<SYCLMemObjI>> ObjsReadyToRelease;
+  {
+    ReadLockT Lock(MGraphLock, std::try_to_lock);
+    // if we need blocking mode - force lock waiting
+    if (!Lock.owns_lock() && ForceWait)
+      Lock.lock();
+    if (Lock.owns_lock()) {
+      // Not expected that ForceWait == true with be used in parallel with
+      // adding MemObj to storage, no such scenario.
+      std::lock_guard<std::mutex> LockDef{MDeferredMemReleaseMutex};
+      auto MemObjIt = MDeferredMemObjRelease.begin();
+      while (MemObjIt != MDeferredMemObjRelease.end()) {
+        MemObjRecord *Record = MGraphBuilder.getMemObjRecord((*MemObjIt).get());
+        if (!Record) {
+          // Just trigger delete since no operations on object was perfromed and
+          // no commands and other to wait for
+          MemObjIt = MDeferredMemObjRelease.erase(MemObjIt);
+          continue;
+        }
+        if (!waitForRecordToFinish(Record, Lock, ForceWait)) {
+          MemObjIt++;
+          continue;
+        }
+        ObjsReadyToRelease.push_back(*MemObjIt);
+        MemObjIt = MDeferredMemObjRelease.erase(MemObjIt);
+      }
+    }
+  }
+  if (ObjsReadyToRelease.empty())
+    return;
+
+  std::vector<std::shared_ptr<stream_impl>> StreamsToDeallocate;
+  std::vector<std::shared_ptr<const void>> AuxResourcesToDeallocate;
+  {
+    WriteLockT Lock(MGraphLock, std::try_to_lock);
+    // if we need blocking mode - force lock waiting
+    if (!Lock.owns_lock() && ForceWait)
+      acquireWriteLock(Lock);
+    if (Lock.owns_lock()) {
+      for (auto &MemObj : ObjsReadyToRelease)
+        releaseMemObjRecord(MemObj.get(), StreamsToDeallocate,
+                            AuxResourcesToDeallocate);
+    } else {
+      std::lock_guard<std::mutex> LockDef{MDeferredMemReleaseMutex};
+      MDeferredMemObjRelease.splice(MDeferredMemObjRelease.end(),
+                                    ObjsReadyToRelease);
+    }
+  }
+  deallocateStreams(StreamsToDeallocate);
+  // ObjsReadyToRelease leaving scope and being deleted
+}
 } // namespace detail
 } // __SYCL_INLINE_VER_NAMESPACE(_V1)
 } // namespace sycl