From 9c0ef11ac7a2d4959fd52c4f4a0bc40a314b4008 Mon Sep 17 00:00:00 2001 From: Tim <43156029+AtlantaPepsi@users.noreply.github.com> Date: Wed, 17 Jan 2024 11:33:01 -0500 Subject: [PATCH] Adding timeout functionality/EnvVar to TestBed (#1044) * Adding timeout functionality/EnvVar to TestBed * updating timeout unit to microseconds Signed-off-by: Tim Hu --- test/ReduceTests.cpp | 2 +- test/common/EnvVars.cpp | 4 ++- test/common/EnvVars.hpp | 1 + test/common/ErrCode.hpp | 3 ++- test/common/TestBed.cpp | 1 + test/common/TestBedChild.cpp | 47 ++++++++++++++++++++++++++++++++++-- 6 files changed, 53 insertions(+), 5 deletions(-) diff --git a/test/ReduceTests.cpp b/test/ReduceTests.cpp index 73280b0952..5cc8b5f755 100644 --- a/test/ReduceTests.cpp +++ b/test/ReduceTests.cpp @@ -7,7 +7,7 @@ namespace RcclUnitTesting { - TEST(Reduce, OutOfPlace) + TEST(Reduce, OutOfPlace) { TestBed testBed; diff --git a/test/common/EnvVars.cpp b/test/common/EnvVars.cpp index 95b593fb9e..42251948b4 100644 --- a/test/common/EnvVars.cpp +++ b/test/common/EnvVars.cpp @@ -63,6 +63,7 @@ namespace RcclUnitTesting maxRanksPerGpu = GetEnvVar("UT_MAX_RANKS_PER_GPU", 1); showTiming = GetEnvVar("UT_SHOW_TIMING", 1); useInteractive = GetEnvVar("UT_INTERACTIVE", 0); + timeoutUs = GetEnvVar("UT_TIMEOUT_US" , 200000); // Limit number of supported reduction operators to just ncclSum if only allReduce is built #ifdef BUILD_ALLREDUCE_ONLY @@ -195,7 +196,8 @@ namespace RcclUnitTesting std::make_tuple("UT_MAX_RANKS_PER_GPU", maxRanksPerGpu, "Maximum number of ranks using the same GPU"), std::make_tuple("UT_PRINT_VALUES" , printValues , "Print array values (-1 for all)"), std::make_tuple("UT_SHOW_TIMING" , showTiming , "Show timing table"), - std::make_tuple("UT_INTERACTIVE" , useInteractive, "Run in interactive mode") + std::make_tuple("UT_INTERACTIVE" , useInteractive, "Run in interactive mode"), + std::make_tuple("UT_TIMEOUT_US" , timeoutUs , "Timeout limit for collective calls in us") }; printf("================================================================================\n"); diff --git a/test/common/EnvVars.hpp b/test/common/EnvVars.hpp index 7614225c52..1d21ab210c 100644 --- a/test/common/EnvVars.hpp +++ b/test/common/EnvVars.hpp @@ -28,6 +28,7 @@ namespace RcclUnitTesting int maxRanksPerGpu; // Number of ranks using the same GPU [UT_MAX_RANKS_PER_GPU] bool showTiming; // Show timing per case at end [UT_SHOW_TIMING] bool useInteractive; // Run in interactive mode [UT_INTERACTIVE] + int timeoutUs; // Set timeout for child in microseconds [UT_TIMEOUT_US] // Constructor that parses and collects environment variables EnvVars(); diff --git a/test/common/ErrCode.hpp b/test/common/ErrCode.hpp index 89f5e4f17e..f8ab991b41 100644 --- a/test/common/ErrCode.hpp +++ b/test/common/ErrCode.hpp @@ -11,7 +11,8 @@ namespace RcclUnitTesting typedef enum { TEST_SUCCESS = 0, - TEST_FAIL = 1 + TEST_FAIL = 1, + TEST_TIMEOUT = 2 } ErrCode; #define ERROR(...) printf("\033[0;31m" "[ ERROR ] " "\033[0m" __VA_ARGS__) diff --git a/test/common/TestBed.cpp b/test/common/TestBed.cpp index 20f7f55188..d61b6b832c 100644 --- a/test/common/TestBed.cpp +++ b/test/common/TestBed.cpp @@ -296,6 +296,7 @@ namespace RcclUnitTesting { InteractiveWait("Starting ExecuteCollectives for child " + std::to_string(childId)); PIPE_WRITE(childId, cmd); + PIPE_WRITE(childId, ev.timeoutUs); PIPE_WRITE(childId, useHipGraph); int tempCurrentRanks = currentRanks.size(); PIPE_WRITE(childId, tempCurrentRanks); diff --git a/test/common/TestBedChild.cpp b/test/common/TestBedChild.cpp index 54356d0f0f..38dd0dc0ac 100644 --- a/test/common/TestBedChild.cpp +++ b/test/common/TestBedChild.cpp @@ -393,6 +393,9 @@ namespace RcclUnitTesting ErrCode TestBedChild::ExecuteCollectives() { + int timeoutUs = 0; + PIPE_READ(timeoutUs); + bool useHipGraph = false; PIPE_READ(useHipGraph); @@ -432,7 +435,6 @@ namespace RcclUnitTesting { for (int localRank : localRanksToExecute) { - CHECK_HIP(hipSetDevice(this->deviceIds[localRank])); if (this->verbose) INFO("Capturing stream for rank %d\n", localRank); CHECK_HIP(hipSetDevice(this->deviceIds[localRank])); for (int i = 0; i < this->numStreamsPerGroup; i++) @@ -659,10 +661,47 @@ namespace RcclUnitTesting } // Synchronize + std::vector streamsToComplete; + for (int localRank : localRanksToExecute) + { + for (int i = 0; i < this->numStreamsPerGroup; i++) + streamsToComplete.push_back(this->streams[localRank][i]); + } + int usElapsed = 0; + using namespace std::chrono; + using Clock = std::chrono::high_resolution_clock; + if (this->verbose) INFO("Starting sychronization and timing\n"); + const auto start = Clock::now(); + while (!streamsToComplete.empty() && usElapsed < timeoutUs) + { + for (int i = 0; i < streamsToComplete.size(); i++) + { + if (hipStreamQuery(streamsToComplete[i]) == hipSuccess) + { + streamsToComplete.erase(streamsToComplete.begin() + i); + i--; + } + } + usElapsed = duration_cast(Clock::now() - start).count(); + } + + // timed out + if (!streamsToComplete.empty()) + { + if (this->verbose) INFO("Collective timed out, aborting\n"); + for (int localRank : localRanksToExecute) + { + ncclCommAbort(this->comms[localRank]); + timeoutUs = -1; + } + } + + // extra sync to flush GPU cache for validation later + // TODO: remove this after figuring out & fixing the exact behavior + // of fencing between kernels and at hipStreamQuery for (int localRank : localRanksToExecute) { if (this->verbose) INFO("Starting synchronization for rank %d\n", localRank); - CHECK_HIP(hipSetDevice(this->deviceIds[localRank])); for (int i = 0; i < this->numStreamsPerGroup; i++) CHECK_HIP(hipStreamSynchronize(this->streams[localRank][i])); } @@ -699,6 +738,10 @@ namespace RcclUnitTesting collArg.expected.ToString(collArg.dataType, numOutputElementsToPrint).c_str()); } } + + if (timeoutUs == -1) + return TEST_TIMEOUT; + if (this->verbose) INFO("Child %d finishes ExecuteCollectives()\n", this->childId); return TEST_SUCCESS; }