diff --git a/hipamd/src/hip_graph.cpp b/hipamd/src/hip_graph.cpp index a737700ca8..ac7144121e 100644 --- a/hipamd/src/hip_graph.cpp +++ b/hipamd/src/hip_graph.cpp @@ -1306,6 +1306,9 @@ hipError_t ihipGraphInstantiate(hip::GraphExec** pGraphExec, hip::Graph* graph, std::vector> parallelLists; std::unordered_map> nodeWaitLists; clonedGraph->GetRunList(parallelLists, nodeWaitLists); + if (DEBUG_HIP_FORCE_GRAPH_QUEUES != 0) { + clonedGraph->ScheduleNodes(); + } *pGraphExec = new hip::GraphExec(graphNodes, parallelLists, nodeWaitLists, clonedGraph, clonedNodes, flags); diff --git a/hipamd/src/hip_graph_internal.cpp b/hipamd/src/hip_graph_internal.cpp index e0b5e90a59..f7d0520c8e 100644 --- a/hipamd/src/hip_graph_internal.cpp +++ b/hipamd/src/hip_graph_internal.cpp @@ -240,6 +240,63 @@ void Graph::GetRunList(std::vector>& parallelLists, } } +// ================================================================================================ +void Graph::ScheduleOneNode(Node node, int stream_id) { + if (node->stream_id_ == -1) { + // Assign active stream to the current node + node->stream_id_ = stream_id; + max_streams_ = std::max(max_streams_, stream_id); + + // Update the dependencies if a signal is required + for (auto dep: node->GetDependencies()) { + // Check if the stream ID doesn't match and enable signal + if (dep->stream_id_ != node->stream_id_) { + dep->signal_is_required_ |= true; + } + } + // Process child graph separately, since, there is no connection + if (node->GetType() == hipGraphNodeTypeGraph) { + auto child = reinterpret_cast(node)->childGraph_; + child->ScheduleNodes(); + max_streams_ = std::max(max_streams_, child->max_streams_); + } + for (auto edge: node->GetEdges()) { + ScheduleOneNode(edge, stream_id); + // 1. Each extra edge will get a new stream from the pool + // 2. Streams will be reused if the number of edges > streams + stream_id = (stream_id + 1) % DEBUG_HIP_FORCE_GRAPH_QUEUES; + } + } +} + +// ================================================================================================ +void Graph::ScheduleNodes() { + for (auto node : vertices_) { + node->stream_id_ = -1; + node->signal_is_required_ = false; + } + memset(&roots_[0], 0, sizeof(Node) * roots_.size()); + max_streams_ = 0; + // Start processing all nodes in the graph to find async executions. + int stream_id = 0; + for (auto node : vertices_) { + if (node->stream_id_ == -1) { + ScheduleOneNode(node, stream_id); + // Find the root nodes + if ((node->GetDependencies().size() == 0) && (node->stream_id_ != 0)) { + // Fill in only the first in the sequence + if (roots_[node->stream_id_] == nullptr) { + roots_[node->stream_id_] = node; + } + } + // 1. Each extra root will get a new stream from the pool + // 2. Streams will be recycled if the number of roots > streams + stream_id = (stream_id + 1) % DEBUG_HIP_FORCE_GRAPH_QUEUES; + } + } +} + +// ================================================================================================ bool Graph::TopologicalOrder(std::vector& TopoOrder) { std::queue q; std::unordered_map inDegree; @@ -298,6 +355,8 @@ Graph* Graph::clone(std::unordered_map& clonedNodes) const { userObj->retain(); newGraph->graphUserObj_.insert(userObj); } + // Clone the root nodes to the new graph + memcpy(&newGraph->roots_[0], &roots_[0], sizeof(Node) * roots_.size()); return newGraph; } @@ -334,17 +393,21 @@ hipError_t GraphExec::CreateStreams(uint32_t num_streams) { return hipSuccess; } +// ================================================================================================ hipError_t GraphExec::Init() { hipError_t status = hipSuccess; size_t min_num_streams = 1; - - for (auto& node : topoOrder_) { - status = node->GetNumParallelStreams(min_num_streams); - if (status != hipSuccess) { - return status; + if ((DEBUG_HIP_FORCE_GRAPH_QUEUES == 0) || (parallelLists_.size() == 1)) { + for (auto& node : topoOrder_) { + status = node->GetNumParallelStreams(min_num_streams); + if (status != hipSuccess) { + return status; + } } + status = CreateStreams(parallelLists_.size() - 1 + min_num_streams); + } else { + status = CreateStreams(clonedGraph_->max_streams_); } - status = CreateStreams(parallelLists_.size() - 1 + min_num_streams); if (status != hipSuccess) { return status; } @@ -558,6 +621,172 @@ hipError_t EnqueueGraphWithSingleList(std::vector& topoOrder, hip::St return status; } +// ================================================================================================ +void Graph::UpdateStreams( + hip::Stream* launch_stream, + const std::vector& parallel_streams) { + // Allocate array for parallel streams, based on the graph scheduling + current stream + streams_.resize(parallel_streams.size() + 1); + + // Current stream is the default in the assignment + streams_[0] = launch_stream; + // Assign the streams in the array of all streams + for (uint32_t i = 0; i < parallel_streams.size(); ++i) { + streams_[i + 1] = parallel_streams[i]; + } +} + +// ================================================================================================ +bool Graph::RunOneNode(Node node, bool wait) { + if (node->launch_id_ == -1) { + // Process child graph separately, since, there is no connection + if (node->GetType() == hipGraphNodeTypeGraph) { + auto child = reinterpret_cast(node)->childGraph_; + child->RunNodes(node->stream_id_, &streams_); + } + // Clear the storage of the wait nodes + memset(&wait_order_[0], 0, sizeof(Node) * wait_order_.size()); + amd::Command::EventWaitList waitList; + // Walk through dependencies and find the last launches on each parallel stream + for (auto depNode : node->GetDependencies()) { + // Process only the nodes that have been submitted + if (depNode->launch_id_ != -1) { + // If it's the same stream then skip the signal, since it's in order + if (depNode->stream_id_ != node->stream_id_) { + // If there is no wait node on the stream, then assign one + if ((wait_order_[depNode->stream_id_] == nullptr) || + // If another node executed on the same stream, then use the latest launch only, + // since the same stream has in-order run + (wait_order_[depNode->stream_id_]->launch_id_ < depNode->launch_id_)) { + wait_order_[depNode->stream_id_] = depNode; + } + } + } else { + // It should be a safe return, + // since the last edge to this dependency has to submit the command + return true; + } + } + // Create a wait list from the last launches of all dependencies + for (auto dep : wait_order_) { + if (dep != nullptr) { + // Add all commands in the wait list + for (auto command : dep->GetCommands()) { + waitList.push_back(command); + } + } + } + // Assing a stream to the current node + node->SetStream(streams_); + // Create the execution commands on the assigned stream + auto status = node->CreateCommand(node->GetQueue()); + if (status != hipSuccess) { + LogPrintfError("Command creation for node id(%d) failed!", current_id_ + 1); + return false; + } + // Retain all commands, since potentially the command can finish before a wait signal + for (auto command : node->GetCommands()) { + command->retain(); + } + + // If a wait was requested, then process the list + if (wait && !waitList.empty()) { + node->UpdateEventWaitLists(waitList); + } + // Start the execution + node->EnqueueCommands(node->GetQueue()); + // Assign the launch ID of the submmitted node + node->launch_id_ = current_id_++; + uint32_t i = 0; + // Execute the nodes in the edges list + for (auto edge: node->GetEdges()) { + // Don't wait in the nodes, executed on the same streams and if it has just one dependency + bool wait = ((i < DEBUG_HIP_FORCE_GRAPH_QUEUES) || + (edge->GetDependencies().size() > 1)) ? true : false; + // Execute the edge node + if (!RunOneNode(edge, wait)) { + return false; + } + i++; + } + if ((i == 0) && (node->stream_id_ != 0)) { + // Add a leaf node into the list for a wait. + // Always use the last node, since it's the latest for the particular queue + leafs_[node->stream_id_] = node; + } + } + return true; +} + +// ================================================================================================ +bool Graph::RunNodes( + int32_t base_stream, + const std::vector* parallel_streams) { + if (parallel_streams != nullptr) { + streams_ = *parallel_streams; + } + + amd::Command::EventWaitList wait_list; + current_id_ = 0; + memset(&leafs_[0], 0, sizeof(Node) * leafs_.size()); + + // Add possible waits in parallel streams for the app's default launch stream + constexpr bool kRetainCommand = true; + auto last_command = streams_[base_stream]->getLastQueuedCommand(kRetainCommand); + if (last_command != nullptr) { + // Add the last command into the waiting list + wait_list.push_back(last_command); + // Check if the graph has multiple root nodes + for (uint32_t i = 0; i < DEBUG_HIP_FORCE_GRAPH_QUEUES; ++i) { + if ((base_stream != i) && (roots_[i] != nullptr)) { + // Wait for the app's queue + auto start_marker = new amd::Marker(*streams_[i], true, wait_list); + if (start_marker != nullptr) { + start_marker->enqueue(); + start_marker->release(); + } + } + } + last_command->release(); + } + + // Run all commands in the graph + for (auto node : vertices_) { + if (node->launch_id_ == -1) { + if (!RunOneNode(node, true)) { + return false; + } + } + } + wait_list.clear(); + // Check if the graph has multiple leaf nodes + for (uint32_t i = 0; i < DEBUG_HIP_FORCE_GRAPH_QUEUES; ++i) { + if ((base_stream != i) && (leafs_[i] != nullptr)) { + // Add all commands in the wait list + for (auto command : leafs_[i]->GetCommands()) { + wait_list.push_back(command); + } + } + } + // Wait for leafs in the graph's app stream + if (wait_list.size() > 0) { + auto end_marker = new amd::Marker(*streams_[base_stream], true, wait_list); + if (end_marker != nullptr) { + end_marker->enqueue(); + end_marker->release(); + } + } + // Release commands after execution + for (auto& node : vertices_) { + node->launch_id_ = -1; + for (auto command : node->GetCommands()) { + command->release(); + } + } + return true; +} + +// ================================================================================================ hipError_t GraphExec::Run(hipStream_t graph_launch_stream) { hipError_t status = hipSuccess; @@ -602,24 +831,34 @@ hipError_t GraphExec::Run(hipStream_t graph_launch_stream) { topoOrder_[i]->EnqueueCommands(launch_stream); } } else { - UpdateStream(parallelLists_, launch_stream, this); - amd::Command* rootCommand = nullptr; - amd::Command* endCommand = nullptr; - status = FillCommands(parallelLists_, nodeWaitLists_, topoOrder_, clonedGraph_, rootCommand, - endCommand, launch_stream); - if (status != hipSuccess) { - return status; - } - if (rootCommand != nullptr) { - rootCommand->enqueue(); - rootCommand->release(); - } - for (int i = 0; i < topoOrder_.size(); i++) { - topoOrder_[i]->EnqueueCommands(topoOrder_[i]->GetQueue()); - } - if (endCommand != nullptr) { - endCommand->enqueue(); - endCommand->release(); + if (DEBUG_HIP_FORCE_GRAPH_QUEUES == 0) { + UpdateStream(parallelLists_, launch_stream, this); + amd::Command* rootCommand = nullptr; + amd::Command* endCommand = nullptr; + status = FillCommands(parallelLists_, nodeWaitLists_, topoOrder_, clonedGraph_, rootCommand, + endCommand, launch_stream); + if (status != hipSuccess) { + return status; + } + if (rootCommand != nullptr) { + rootCommand->enqueue(); + rootCommand->release(); + } + for (int i = 0; i < topoOrder_.size(); i++) { + topoOrder_[i]->EnqueueCommands(topoOrder_[i]->GetQueue()); + } + if (endCommand != nullptr) { + endCommand->enqueue(); + endCommand->release(); + } + } else { + // Update streams for the graph execution + clonedGraph_->UpdateStreams(launch_stream, parallel_streams_); + // Execute all nodes in the graph + if (!clonedGraph_->RunNodes()) { + LogError("Failed to launch nodes!"); + return hipErrorOutOfMemory; + } } } amd::ScopedLock lock(GraphExecStatusLock_); diff --git a/hipamd/src/hip_graph_internal.hpp b/hipamd/src/hip_graph_internal.hpp index d706d9a43f..c06a376286 100644 --- a/hipamd/src/hip_graph_internal.hpp +++ b/hipamd/src/hip_graph_internal.hpp @@ -203,6 +203,9 @@ class GraphKernelArgManager : public amd::ReferenceCountedObject, public amd::Gr struct GraphNode : public hipGraphNodeDOTAttribute { protected: + // Declare Graph and GraphExec as friends of node for simpler access to GraphNode fields + friend class Graph; + friend class GraphExec; hip::Stream* stream_ = nullptr; unsigned int id_; hipGraphNodeType type_; @@ -210,15 +213,16 @@ struct GraphNode : public hipGraphNodeDOTAttribute { std::vector edges_; std::vector dependencies_; bool visited_; - // count of in coming edge s - size_t inDegree_; - // count of outgoing edges - size_t outDegree_; + size_t inDegree_; //!< count of in coming edges (@todo: remove, it's dependencies_.size()) + size_t outDegree_; //!< count of outgoing edges (@todo: remove, it's edges_.size()) + int32_t stream_id_ = -1; //! Stream ID on which this node will be executed + int32_t launch_id_ = -1; //! Launch ID of this node in the entire graph execution sequence static int nextID; struct Graph* parentGraph_; static std::unordered_set nodeSet_; static amd::Monitor nodeSetLock_; unsigned int isEnabled_; + bool signal_is_required_ = false; //!< This node requires a signal on the command std::vector gpuPackets_; //!< GPU Packet to enqueue during graph launch std::string capturedKernelName_; size_t alignedKernArgSize_ = 256; //!< Aligned size required for kernel args @@ -298,6 +302,15 @@ struct GraphNode : public hipGraphNodeDOTAttribute { virtual void SetStream(hip::Stream* stream, GraphExec* ptr = nullptr) { stream_ = stream; } + //! Updates the grpah node with the execution stream + void SetStream( + const std::vector& streams //!< A pool of streams to use in graph's execution + ) { + assert(stream_id_ != -1 && "Stream ID wasn't initialized"); + stream_ = streams[stream_id_]; + // Reset the launch ID after the stream assignment + launch_id_ = -1; + } /// Create amd::command for the graph node virtual hipError_t CreateCommand(hip::Stream* stream) { commands_.clear(); @@ -454,6 +467,9 @@ struct GraphNode : public hipGraphNodeDOTAttribute { }; struct Graph { + // Mark GraphExec as friend for faster access to the Graph fields. + // (@todo GrpahExec should be derived from Graph) + friend class GraphExec; std::vector vertices_; const Graph* pOriginalGraph_ = nullptr; static std::unordered_set graphSet_; @@ -461,6 +477,14 @@ struct Graph { std::unordered_set graphUserObj_; unsigned int id_; static int nextID; + int max_streams_ = 0; //!< Maximum number of extra streams used in the graph launch + std::vector roots_; //!< Root nodes, used in parallel launches + std::vector leafs_; //!< The list of leaf nodes on every parallel stream + //!< Used as a temporary storage for the waiting nodes + //!< to reduce the stack pressure in recursion + std::vector wait_order_; + std::vector streams_; //!< The list of streams, used in the execution + int32_t current_id_ = 0; //!< The current node ID in the graph execution sequence hip::Device* device_; //!< HIP device object hip::MemoryPool* mem_pool_; //!< Memory pool, associated with this graph std::unordered_set capturedNodes_; @@ -476,6 +500,9 @@ struct Graph { mem_pool_ = device->GetGraphMemoryPool(); mem_pool_->retain(); graphInstantiated_ = false; + roots_.resize(DEBUG_HIP_FORCE_GRAPH_QUEUES); + leafs_.resize(DEBUG_HIP_FORCE_GRAPH_QUEUES); + wait_order_.resize(DEBUG_HIP_FORCE_GRAPH_QUEUES); } ~Graph() { @@ -544,6 +571,35 @@ struct Graph { std::unordered_map>& dependencies); void GetRunList(std::vector>& parallelLists, std::unordered_map>& dependencies); + + //! Schedules one node on a vitual stream. + //! It will also process the nodes in edges, using recursion + void ScheduleOneNode( + Node node, //!< Node for scheduling on a virtual stream + int stream_id //!< Current active virtual stream to use for scheduling + ); + + //! Schedules all nodes in the graph into different streams + void ScheduleNodes(); + + //! Update streams for the graph execution + void UpdateStreams( + hip::Stream* launch_stream, //!< Launch stream from the application + const std::vector& parallel_stream //!< The list of parallel streams + ); + + //! Runs one node on the assigned stream + bool RunOneNode( + Node node, //!< Node for the execution on GPU + bool wait //!< Wait dependencies + ); + + //! Runs all nodes from the execution graph on the assigned streams + bool RunNodes( + int32_t base_stream = 0, //!< The base stream to run the graph on + const std::vector* streams = nullptr //!< Streams to run the graph + ); + bool TopologicalOrder(std::vector& TopoOrder); Graph* clone(std::unordered_map& clonedNodes) const; @@ -638,10 +694,10 @@ struct GraphExec : public amd::ReferenceCountedObject { static std::unordered_set graphExecSet_; static amd::Monitor graphExecSetLock_; uint64_t flags_ = 0; - bool repeatLaunch_ = false; GraphKernelArgManager* kernArgManager_ = nullptr; //!< Kernel Arg manager for graph. int instantiateDeviceId_ = -1; - bool hasHiddenHeap_ = false; //!< Hidden heap indicator for Kernel node + bool hasHiddenHeap_ = false; //!< Hidden heap indicator for Kernel node + bool repeatLaunch_ = false; public: GraphExec(std::vector& topoOrder, std::vector>& lists, @@ -663,6 +719,7 @@ struct GraphExec : public amd::ReferenceCountedObject { ~GraphExec() { for (auto stream : parallel_streams_) { if (stream != nullptr) { + stream->finish(); hip::Stream::Destroy(stream); } } @@ -683,6 +740,7 @@ struct GraphExec : public amd::ReferenceCountedObject { } return clonedNode; } + //! Check if kernel node has hidden heap bool HasHiddenHeap() const { return hasHiddenHeap_; } //! Graph has nodes that require hidden heap. @@ -1133,13 +1191,28 @@ class GraphKernelNode : public GraphNode { } commands_.reserve(1); amd::Command* command; + uint32_t flags = 0; + if (DEBUG_HIP_FORCE_ASYNC_QUEUE) { + // If there is one dependency, but many edges, then execute this node in any order + if (((dependencies_.size() == 1) && (dependencies_[0]->GetEdges().size() > 1) && + (DEBUG_HIP_FORCE_GRAPH_QUEUES == 1))) { + // Makes sure the first node in the edges will have a barrier always + if (dependencies_[0]->GetEdges()[0] != this) { + flags = hipExtAnyOrderLaunch; + } + } + } status = ihipLaunchKernelCommand( command, func, kernelParams_.gridDim.x * kernelParams_.blockDim.x, kernelParams_.gridDim.y * kernelParams_.blockDim.y, kernelParams_.gridDim.z * kernelParams_.blockDim.z, kernelParams_.blockDim.x, kernelParams_.blockDim.y, kernelParams_.blockDim.z, kernelParams_.sharedMemBytes, stream, kernelParams_.kernelParams, kernelParams_.extra, kernelEvents_.startEvent_, - kernelEvents_.stopEvent_, 0, 0, 0, 0, 0, 0, 0); + kernelEvents_.stopEvent_, flags, 0, 0, 0, 0, 0, 0); + if (signal_is_required_) { + // Optimize the barriers by adding a signal into the dispatch packet directly + command->SetProfiling(); + } commands_.emplace_back(command); return status; } @@ -2159,6 +2232,7 @@ class GraphHostNode : public GraphNode { } }; +// ================================================================================================ class GraphEmptyNode : public GraphNode { public: GraphEmptyNode() : GraphNode(hipGraphNodeTypeEmpty, "solid", "rectangle", "EMPTY") {} @@ -2173,10 +2247,13 @@ class GraphEmptyNode : public GraphNode { if (status != hipSuccess) { return status; } - amd::Command::EventWaitList waitList; - commands_.reserve(1); - amd::Command* command = new amd::Marker(*stream, !kMarkerDisableFlush, waitList); - commands_.emplace_back(command); + // If just one stream was forced for the execution, then the barrier can be skipped + if (DEBUG_HIP_FORCE_GRAPH_QUEUES != 1) { + amd::Command::EventWaitList waitList; + commands_.reserve(1); + amd::Command* command = new amd::Marker(*stream, !kMarkerDisableFlush, waitList); + commands_.emplace_back(command); + } return hipSuccess; } }; diff --git a/rocclr/device/rocm/rocvirtual.cpp b/rocclr/device/rocm/rocvirtual.cpp index 1b896d98f6..ca10e50857 100644 --- a/rocclr/device/rocm/rocvirtual.cpp +++ b/rocclr/device/rocm/rocvirtual.cpp @@ -1384,6 +1384,9 @@ bool VirtualGPU::initPool(size_t kernarg_pool_size) { roc_device_.info().largeBar_) { kernarg_pool_base_ = reinterpret_cast
(roc_device_.deviceLocalAlloc(kernarg_pool_size_)); + // @note Workaround first access penalty. + // KFD may update CPU page tables on the first CPU access + *kernarg_pool_base_ = 0; } else { kernarg_pool_base_ = reinterpret_cast
(roc_device_.hostAlloc(kernarg_pool_size_, 0, Device::MemorySegment::kKernArg)); diff --git a/rocclr/platform/command.hpp b/rocclr/platform/command.hpp index 101d1f67d3..2ed4d212e0 100644 --- a/rocclr/platform/command.hpp +++ b/rocclr/platform/command.hpp @@ -152,6 +152,12 @@ class Event : public RuntimeObject { } public: + //! Use profiling info to force a tracking signal on command + void SetProfiling() { + EnableProfiling(); + profilingInfo_.marker_ts_ = true; + } + //! Return the context for this event. virtual const Context& context() const = 0; diff --git a/rocclr/utils/flags.hpp b/rocclr/utils/flags.hpp index 93cfd236c8..fa7c8ee2cb 100644 --- a/rocclr/utils/flags.hpp +++ b/rocclr/utils/flags.hpp @@ -249,6 +249,10 @@ release(bool, HIP_VMEM_MANAGE_SUPPORT, true, \ "Virtual Memory Management Support") \ release(bool, DEBUG_HIP_GRAPH_DOT_PRINT, false, \ "Enable/Disable graph debug dot print dump") \ +release(bool, DEBUG_HIP_FORCE_ASYNC_QUEUE, false, \ + "Forces grpahs into async queue mode. DEBUG_HIP_FORCE_GRAPH_QUEUES must be 1") \ +release(uint, DEBUG_HIP_FORCE_GRAPH_QUEUES, 4, \ + "Forces the number of streams for the graph parallel execution") \ release(bool, HIP_ALWAYS_USE_NEW_COMGR_UNBUNDLING_ACTION, false, \ "Force to always use new comgr unbundling action") \ release(bool, DEBUG_HIP_KERNARG_COPY_OPT, true, \