Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly
Appearance settings

[c10d] PGNCCL refactor part 1: adds assert size==1 #119099

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 5 commits into from
Closed

[c10d] PGNCCL refactor part 1: adds assert size==1 #119099

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
e65f5b7
Refactor tests written in multi-device style
kwen2501 Feb 2, 2024
cfdb922
Add assert size==1 to PGNCCL
kwen2501 Feb 2, 2024
5a40684
Make internal test happy
kwen2501 Feb 6, 2024
7ccb5b8
Skip test at SetUp if no CUDA
kwen2501 Feb 7, 2024
fd49e29
Add ;
kwen2501 Feb 7, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
51 changes: 9 additions & 42 deletions 51 test/cpp/c10d/ProcessGroupNCCLErrorsTest.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -254,16 +254,19 @@ class ProcessGroupNCCLErrorsTest : public ::testing::Test {
void SetUp() override {
// Enable LOG(INFO) messages.
c10::initLogging();
size_t numDevices = cudaNumDevices();
// Need to have this check for at SetUp to make sure we only run the test --
// including the init -- when there are GPUs available.
if (skipTest()) {
GTEST_SKIP() << "Skipping ProcessGroupNCCLErrorsTest because system "
<< "requirement is not met (no CUDA or GPU).";
}

size_t numDevices = 1; // One device per rank (thread)
TemporaryFile file;
store_ = c10::make_intrusive<::c10d::FileStore>(file.path, 1);

at::cuda::OptionalCUDAGuard deviceGuard;
tensors_.resize(numDevices);
for (const auto i : c10::irange(numDevices)) {
deviceGuard.set_index(i);
tensors_[i] = at::ones({3, 3}, at::kCUDA);
}
tensors_[0] = at::empty({3, 3}, at::kCUDA);
}

void TearDown() override {
Expand All @@ -275,18 +278,13 @@ class ProcessGroupNCCLErrorsTest : public ::testing::Test {
};

TEST_F(ProcessGroupNCCLErrorsTest, testNCCLErrorsBlocking) {
if (skipTest()) {
return;
}

ASSERT_TRUE(setenv(c10d::TORCH_NCCL_BLOCKING_WAIT[0].c_str(), "1", 1) == 0);
auto options = c10d::ProcessGroupNCCL::Options::create();
options->timeout = std::chrono::milliseconds(1000);
ProcessGroupNCCLSimulateErrors pg(store_, 0, 1, options);

auto work = pg.allreduce(tensors_);
work->wait();
EXPECT_TRUE(work->isSuccess());
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is bc single device style is blocking and multi was non blocking ? Or what

Copy link
Contributor Author

@kwen2501 kwen2501 Feb 3, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No, it is just because the is_success API has been deprecated for a long time at the pybind place:
https://github.com/pytorch/pytorch/blob/main/torch/csrc/distributed/c10d/init.cpp#L2556-L2560
so taking this chance to take out the c++ code as well.
Also because its c++ code would complicate my refactorization.

It makes sense (to deprecate it) -- how can we tell a collective is successful? Are we telling user that the result is correct? We obviously don't/cannot do data check.. So at best we can only tell is_complete().

EXPECT_EQ(1, pg.getNCCLCommCacheSize());

// Now run all reduce with errors.
Expand All @@ -296,25 +294,19 @@ TEST_F(ProcessGroupNCCLErrorsTest, testNCCLErrorsBlocking) {

// Verify the work item failed.
EXPECT_TRUE(work->isCompleted());
EXPECT_FALSE(work->isSuccess());
EXPECT_THROW(work->wait(), std::runtime_error);

// Communicators might be aborted here, further operations would fail.
}

TEST_F(ProcessGroupNCCLErrorsTest, testNCCLTimedoutErrorsBlocking) {
if (skipTest()) {
return;
}

ASSERT_TRUE(setenv(c10d::TORCH_NCCL_BLOCKING_WAIT[0].c_str(), "1", 1) == 0);
auto options = c10d::ProcessGroupNCCL::Options::create();
options->timeout = std::chrono::milliseconds(3000);
ProcessGroupNCCLTimedOutErrors pg(store_, 0, 1, options);

auto work = pg.allreduce(tensors_);
work->wait();
EXPECT_TRUE(work->isSuccess());
EXPECT_EQ(1, pg.getNCCLCommCacheSize());

// Now run all reduce with errors.
Expand All @@ -326,17 +318,12 @@ TEST_F(ProcessGroupNCCLErrorsTest, testNCCLTimedoutErrorsBlocking) {
}

TEST_F(ProcessGroupNCCLErrorsTest, testNCCLErrorsNonBlocking) {
if (skipTest()) {
return;
}

auto options = c10d::ProcessGroupNCCL::Options::create();
options->timeout = std::chrono::milliseconds(3000);
ProcessGroupNCCLSimulateErrors pg(store_, 0, 1, options);

auto work = pg.allreduce(tensors_);
pg.barrier()->wait();
EXPECT_TRUE(work->isSuccess());
EXPECT_EQ(1, pg.getNCCLCommCacheSize());

// Now run all reduce with errors.
Expand All @@ -347,10 +334,7 @@ TEST_F(ProcessGroupNCCLErrorsTest, testNCCLErrorsNonBlocking) {
work->wait();
pg.barrier()->wait();

// Verify the work item failed.
EXPECT_TRUE(work->isCompleted());
EXPECT_FALSE(work->isSuccess());

// Communicators might be aborted here, further operations would fail.
}

Expand Down Expand Up @@ -388,10 +372,6 @@ class TestDebugInfoWriter : public c10d::DebugInfoWriter {
};

TEST_F(ProcessGroupNCCLErrorsTest, testNCCLErrorsNoHeartbeat) {
if (skipTest()) {
return;
}

int heartBeatIntervalInSec = 2;
std::string timeInterval = std::to_string(heartBeatIntervalInSec);
ASSERT_TRUE(setenv(c10d::TORCH_NCCL_BLOCKING_WAIT[0].c_str(), "1", 1) == 0);
Expand Down Expand Up @@ -426,7 +406,6 @@ TEST_F(ProcessGroupNCCLErrorsTest, testNCCLErrorsNoHeartbeat) {
// Normal collective case.
auto work = pg.allreduce(tensors_);
work->wait();
EXPECT_TRUE(work->isSuccess());

work = pg.allreduce(tensors_);
{
Expand All @@ -440,7 +419,6 @@ TEST_F(ProcessGroupNCCLErrorsTest, testNCCLErrorsNoHeartbeat) {
EXPECT_TRUE(pg.getErrorCaughtFlag());
}
work->wait();
EXPECT_TRUE(work->isSuccess());
EXPECT_TRUE(traces.size() > 0);
auto filename = c10::str(tempFilename, 0);
auto traceFromStorage = readTraceFromFile(filename, traces.size());
Expand Down Expand Up @@ -492,12 +470,6 @@ class ProcessGroupNCCLWatchdogTimeoutTest : public ProcessGroupNCCLErrorsTest {
};

TEST_F(ProcessGroupNCCLWatchdogTimeoutTest, testNCCLTimedoutDebugInfoFinished) {
// Need to have this check for every test to make sure we only run the test
// when there are GPUs available.
if (skipTest()) {
return;
}

ProcessGroupNCCLNoHeartbeatCaught pg(store_, 0, 1, options_);
// Write debug info will lead to watchdog thread to wait for 30 seconds.
// And this is hard to override, so we just call it before hand. Otherwise,
Expand All @@ -518,11 +490,6 @@ TEST_F(ProcessGroupNCCLWatchdogTimeoutTest, testNCCLTimedoutDebugInfoFinished) {
}

TEST_F(ProcessGroupNCCLWatchdogTimeoutTest, testNCCLTimedoutDebugInfoStuck) {
// Need to have this check for every test to make sure we only run the test
// when there are GPUs available.
if (skipTest()) {
return;
}
ProcessGroupNCCLDebugInfoStuck pg(store_, 0, 1, options_);
// Need to keep main thread sleep longer so that we can let heartbeat monitor
// thread to finish the extra wait and flip the flag.
Expand Down
Loading
Morty Proxy This is a proxified and sanitized view of the page, visit original site.