[SYCL][Graph] Throw an exception if multiple concurrent submissions (#…

…321)

* [SYCL][Graph] Throw an exception if the graph is resubmitted while the previous submission has not completed.

Adds a check for previous submission completion before submitting a graph to backend.
Throws an invalid exception if the previous submission has not completed.
Adds a test to verify the exception throwing.
Updates e2e tests to conform to this behaviour (required by the specification).

* [SYCL][Graph] Add missing test update and change function name

sycl/source/detail/graph_impl.cpp

@@ -623,6 +623,11 @@ exec_graph_impl::enqueue(const std::shared_ptr<sycl::detail::queue_impl> &Queue,
  sycl::detail::EventImplPtr NewEvent;
 
  if (CommandBuffer) {
+ if (!previousSubmissionCompleted()) {
+ throw sycl::exception(make_error_code(errc::invalid),
+ "This Graph cannot be submitted at the moment "
+ "because the previous run has not yet completed.");
+ }
  NewEvent = CreateNewEvent();
  sycl::detail::pi::PiEvent *OutEvent = &NewEvent->getHandleRef();
  // Merge requirements from the nodes into requirements (if any) from the

sycl/source/detail/graph_impl.hpp

@@ -14,6 +14,7 @@
 #include <sycl/handler.hpp>
 
 #include <detail/accessor_impl.hpp>
+#include <detail/event_impl.hpp>
 #include <detail/kernel_impl.hpp>
 
 #include <cstring>
@@ -869,6 +870,20 @@ class exec_graph_impl {
  return false;
  }
 
+ /// Checks if the previous submissions of this graph have been completed
+ /// This function checks the status of events associated to the previous graph
+ /// submissions.
+ /// @return true if all previous submissions have been completed, false
+ /// otherwise.
+ bool previousSubmissionCompleted() const {
+ for (auto Event : MExecutionEvents) {
+ if (!Event->isCompleted()) {
+ return false;
+ }
+ }
+ return true;
+ }
+
 private:
  /// Create a command-group for the node and add it to command-buffer by
  /// going through the scheduler.

sycl/test-e2e/Graph/Explicit/enqueue_ordering.cpp

@@ -70,6 +70,8 @@ int main() {
  });
  });
 
+ E4.wait();
+
  // Buffer elements set to 22
  Queue.submit([&](handler &CGH) {
  CGH.depends_on(E5);

sycl/test-e2e/Graph/Inputs/add_nodes_after_finalize.cpp

@@ -63,13 +63,15 @@ int main() {
  CGH.depends_on(Event);
  CGH.ext_oneapi_graph(GraphExec);
  });
+ Event.wait();
  }
 
  for (unsigned n = 0; n < Iterations; n++) {
  Event = Queue.submit([&](handler &CGH) {
  CGH.depends_on(Event);
  CGH.ext_oneapi_graph(GraphExecAdditional);
  });
+ Event.wait();
  }
 
  Queue.wait_and_throw();

sycl/test-e2e/Graph/Inputs/basic_buffer.cpp

@@ -41,6 +41,7 @@ int main() {
  CGH.depends_on(Event);
  CGH.ext_oneapi_graph(GraphExec);
  });
+ Event.wait();
  }
  Queue.wait_and_throw();
  }

sycl/test-e2e/Graph/Inputs/basic_usm.cpp

@@ -49,6 +49,7 @@ int main() {
  CGH.depends_on(Event);
  CGH.ext_oneapi_graph(GraphExec);
  });
+ Event.wait();
  }
 
  Queue.wait_and_throw();

sycl/test-e2e/Graph/Inputs/basic_usm_host.cpp

@@ -44,6 +44,7 @@ int main() {
  CGH.depends_on(Event);
  CGH.ext_oneapi_graph(GraphExec);
  });
+ Event.wait();
  }
 
  Queue.wait_and_throw();

sycl/test-e2e/Graph/Inputs/basic_usm_mixed.cpp

@@ -47,6 +47,7 @@ int main() {
  CGH.depends_on(Event);
  CGH.ext_oneapi_graph(GraphExec);
  });
+ Event.wait();
  }
 
  Queue.wait_and_throw();

sycl/test-e2e/Graph/Inputs/basic_usm_shared.cpp

@@ -44,6 +44,7 @@ int main() {
  CGH.depends_on(Event);
  CGH.ext_oneapi_graph(GraphExec);
  });
+ Event.wait();
  }
 
  Queue.wait_and_throw();

sycl/test-e2e/Graph/Inputs/buffer_copy.cpp

@@ -110,6 +110,7 @@ int main() {
  CGH.depends_on(Event);
  CGH.ext_oneapi_graph(GraphExec);
  });
+ Event.wait();
  }
  Queue.wait_and_throw();
  }

sycl/test-e2e/Graph/Inputs/buffer_copy_2d.cpp

@@ -105,6 +105,7 @@ int main() {
  CGH.depends_on(Event);
  CGH.ext_oneapi_graph(GraphExec);
  });
+ Event.wait();
  }
  Queue.wait_and_throw();
  }

sycl/test-e2e/Graph/Inputs/buffer_ordering.cpp

@@ -72,7 +72,8 @@ int main() {
  });
 
  // Buffer elements set to 10
- Queue.submit([&](handler &CGH) { CGH.ext_oneapi_graph(ExecGraph); });
+ auto Event =
+ Queue.submit([&](handler &CGH) { CGH.ext_oneapi_graph(ExecGraph); });
 
  // Buffer elements set to 20
  Queue.submit([&](handler &CGH) {
@@ -83,6 +84,7 @@ int main() {
  });
  });
 
+ Event.wait();
  // Buffer elements set to 22
  Queue.submit([&](handler &CGH) { CGH.ext_oneapi_graph(ExecGraph); });
 

sycl/test-e2e/Graph/Inputs/queue_shortcuts.cpp

@@ -38,12 +38,14 @@ int main() {
 
  // Execute several iterations of the graph using the different shortcuts
  event Event = Queue.ext_oneapi_graph(GraphExec);
+ Event.wait();
 
  assert(Iterations > 2);
  const size_t LoopIterations = Iterations - 2;
  std::vector<event> Events(LoopIterations);
  for (unsigned n = 0; n < LoopIterations; n++) {
  Events[n] = Queue.ext_oneapi_graph(GraphExec, Event);
+ Events[n].wait();
  }
 
  Queue.ext_oneapi_graph(GraphExec, Events).wait();

sycl/test-e2e/Graph/Inputs/sub_graph.cpp

@@ -103,6 +103,7 @@ int main() {
  CGH.depends_on(Event);
  CGH.ext_oneapi_graph(MainGraphExec);
  });
+ Event.wait();
  }
  Queue.wait_and_throw();
 

sycl/test-e2e/Graph/Inputs/temp_buffer_reinterpret.cpp

@@ -48,6 +48,7 @@ int main() {
  CGH.depends_on(Event);
  CGH.ext_oneapi_graph(GraphExec);
  });
+ Event.wait();
  }
 
  Queue.copy(BufferA.get_access(), DataA.data());

sycl/test-e2e/Graph/Inputs/usm_copy.cpp

@@ -103,6 +103,7 @@ int main() {
  CGH.depends_on(Event);
  CGH.ext_oneapi_graph(GraphExec);
  });
+ Event.wait();
  }
 
  Queue.copy(PtrA, DataA.data(), Size, Event);

sycl/test-e2e/Graph/RecordReplay/after_use.cpp

@@ -54,6 +54,7 @@ int main() {
  CGH.depends_on(Event);
  CGH.ext_oneapi_graph(GraphExec);
  });
+ Event.wait();
  }
  Queue.wait_and_throw();
 

sycl/test-e2e/Graph/RecordReplay/barrier_with_work.cpp

@@ -100,6 +100,7 @@ int main() {
  CGH.depends_on(Event);
  CGH.ext_oneapi_graph(GraphExec);
  });
+ Event.wait();
  }
  Queue.wait_and_throw();
 

sycl/test-e2e/Graph/submission_while_executing.cpp

@@ -0,0 +1,96 @@
+// REQUIRES: cuda || level_zero, gpu
+// RUN: %{build} -o %t.out
+// RUN: %{run} %t.out
+// RUN: %if ext_oneapi_level_zero %{env ZE_DEBUG=4 %{run} %t.out 2>&1 | FileCheck %s %}
+//
+// CHECK-NOT: LEAK
+
+// Test calling queue::submit(graph) while the previous submission of graph has
+// not been completed. The second run is to check that there are no leaks
+// reported with the embedded ZE_DEBUG=4 testing capability.
+
+#include "graph_common.hpp"
+
+int main() {
+ queue Queue{{sycl::ext::intel::property::queue::no_immediate_command_list{}}};
+
+ using T = int;
+
+ size_t LargeSize =
+ 1000000; // we use large Size to increase the kernel execution time
+ size_t NumIterations = 10;
+ size_t SuccessfulSubmissions = 0;
+
+ std::vector<T> DataA(LargeSize), DataB(LargeSize), DataC(LargeSize);
+
+ std::iota(DataA.begin(), DataA.end(), 1);
+ std::iota(DataB.begin(), DataB.end(), 10);
+ std::iota(DataC.begin(), DataC.end(), 1000);
+
+ std::vector<T> ReferenceA(DataA), ReferenceB(DataB), ReferenceC(DataC);
+
+ exp_ext::command_graph Graph{Queue.get_context(), Queue.get_device()};
+
+ T *PtrA = malloc_device<T>(LargeSize, Queue);
+ T *PtrB = malloc_device<T>(LargeSize, Queue);
+ T *PtrC = malloc_device<T>(LargeSize, Queue);
+
+ Queue.copy(DataA.data(), PtrA, LargeSize);
+ Queue.copy(DataB.data(), PtrB, LargeSize);
+ Queue.copy(DataC.data(), PtrC, LargeSize);
+ Queue.wait_and_throw();
+
+ Graph.begin_recording(Queue);
+ run_kernels_usm(Queue, LargeSize, PtrA, PtrB, PtrC);
+ Graph.end_recording();
+
+ auto GraphExec = Graph.finalize();
+
+ // Serial Submissions
+ for (unsigned i = 0; i < NumIterations; ++i) {
+ Queue.submit([&](handler &CGH) { CGH.ext_oneapi_graph(GraphExec); });
+ Queue.wait_and_throw();
+ }
+
+ // Concurrent Submissions
+ sycl::event Event;
+ sycl::info::event_command_status PreEventInfo =
+ sycl::info::event_command_status::ext_oneapi_unknown;
+ std::error_code ErrorCode = make_error_code(sycl::errc::success);
+ for (unsigned i = 0; i < NumIterations; ++i) {
+ try {
+ Event =
+ Queue.submit([&](handler &CGH) { CGH.ext_oneapi_graph(GraphExec); });
+ } catch (const sycl::exception &e) {
+ ErrorCode = e.code();
+ }
+ if ((PreEventInfo == sycl::info::event_command_status::submitted) ||
+ (PreEventInfo == sycl::info::event_command_status::running)) {
+ assert(ErrorCode == sycl::errc::invalid);
+ } else {
+ // Submission has succeeded
+ SuccessfulSubmissions++;
+ }
+ PreEventInfo =
+ Event.get_info<sycl::info::event::command_execution_status>();
+ }
+ Queue.wait_and_throw();
+
+ Queue.copy(PtrA, DataA.data(), LargeSize);
+ Queue.copy(PtrB, DataB.data(), LargeSize);
+ Queue.copy(PtrC, DataC.data(), LargeSize);
+ Queue.wait_and_throw();
+
+ free(PtrA, Queue);
+ free(PtrB, Queue);
+ free(PtrC, Queue);
+
+ // Compute the reference based on the total number of successful executions
+ calculate_reference_data(NumIterations + SuccessfulSubmissions, LargeSize,
+ ReferenceA, ReferenceB, ReferenceC);
+ assert(ReferenceA == DataA);
+ assert(ReferenceB == DataB);
+ assert(ReferenceC == DataC);
+
+ return 0;
+}