Add barrier call to torch module to support easy synchronization for process sets

[Tixxx]

Checklist before submitting

• Did you read the contributor guide?
• Did you update the docs?
• Did you write any tests to validate this change?
• Did you update the CHANGELOG, if this change affects users?

Description

Add barrier call to torch module to support easy synchronization for process sets. Existing methods that use other apis are either not well known or not designed to be a synchronization point. This also achieves parity with other distributed training frameworks.

This also fixes a minor issue with is_initialized(). The horovod_is_initialized returns an atomic bool which will be interpreted as a random INT by ctypes on Mac. Changed it to return INT instead.

Fixes #3121 (issue).
#3121

Review process to land

1. All tests and other checks must succeed.
2. At least one member of the technical steering committee must review and approve.
3. If any member of the technical steering committee requests changes, they must be addressed.

[github-actions]

Unit Test Results

     692 files  ±0       692 suites  ±0   6h 37m 17s ⏱️ ±0s
     701 tests ±0       652 ✔️ ±0       49 💤 ±0  0 ❌ ±0 
14 817 runs  ±0  10 350 ✔️ ±0  4 467 💤 ±0  0 ❌ ±0 

Results for commit d9afdae. ± Comparison against base commit d9afdae.

♻️ This comment has been updated with latest results.

[github-actions]

Unit Test Results (with flaky tests)

     782 files  ±0       782 suites  ±0   7h 1m 52s ⏱️ ±0s
     701 tests ±0       652 ✔️ ±0       48 💤 ±0  1 ❌ ±0 
16 921 runs  ±0  11 675 ✔️ ±0  5 245 💤 ±0  1 ❌ ±0 

For more details on these failures, see this check.

Results for commit d9afdae. ± Comparison against base commit d9afdae.

♻️ This comment has been updated with latest results.

[chongxiaoc]

LGTM

So Lightning has to use this new op to serve as barrier instead of using join()?

[Tixxx]

LGTM

So Lightning has to use this new op to serve as barrier instead of using join()?

@chongxiaoc Yes, otherwise we will keep bumping into the error reported in the issue. I can take a look on the lightning side as well. As a side note, we need to implement join for other collective operations(allgather, bcast, etc) too to fully support join during training.

[chongxiaoc]

@tgaddair @romerojosh Can you take a look?

[maxhgerlach]

Hi @Tixxx,

this looks like a very useful contribution and would indeed provide a missing feature which is regularly requested.

I left a comment wondering about possible multithreading issues with the current implementation of this PR.

[Tixxx]

Hi @Tixxx,

this looks like a very useful contribution and would indeed provide a missing feature which is regularly requested.

I left a comment wondering about possible multithreading issues with the current implementation of this PR.

@maxhgerlach @tgaddair I added a new barrier request to process the op in the background thread. please take a look. Thanks.

[tgaddair]

LGTM! @romerojosh can you also take a quick look?

[romerojosh]

Thanks for making updates @Tixxx. While this looks like it gets the job done, the implementation here looks a bit involved just to implement a barrier operation in the background thread. In torch.distributed, a barrier is implemented by scheduling a 1 element allreduce. If we do something similar, this would greatly simplify the implementation since you can just implement the barrier op directly in Python using existing allreduce operations.

What do you think?

[Tixxx]

Thanks for making updates @Tixxx. While this looks like it gets the job done, the implementation here looks a bit involved just to implement a barrier operation in the background thread. In torch.distributed, a barrier is implemented by scheduling a 1 element allreduce. If we do something similar, this would greatly simplify the implementation since you can just implement the barrier op directly in Python using existing allreduce operations.

What do you think?

@romerojosh You raised a good point. I had thought about it at the beginning, but after going through c10d's code base, it looks like they only go with allreduce approach for the nccl backend which I think it's due to the lack of barrier api in NCCL. For other backends that expose a barrier api, they call the api directly.
This is where the separation of control plane and computation plane in Horovod comes in handy, we don't need to reply on backend computation primitives for simple control ops like barrier. Also, the complexity of barrier is simpler compared to allreduce. For example, in open mpi, the complexity of mpi_barrier is almost constant when called on mpi_world communicator. Launching an allreduce call for barrier seems a bit of an overkill to me. This might look involved in the implementation's perspective, but IMO it's actually simpler during runtime.
let me know if this makes sense. Thanks!

[romerojosh]

@Tixxx Thanks for the comment. Really, the bulk of the wall time overhead from the barrier would be from the control plane processing, not the actual barrier call. The benefits of a faster barrier op will mostly be hidden by that overhead.
However, as you've already implemented the barrier code in the backend, I'm fine with it. Please do take a look at the generated flatbuffer files though to ensure things are consistent.

[Tixxx]

@Tixxx Thanks for the comment. Really, the bulk of the wall time overhead from the barrier would be from the control plane processing, not the actual barrier call. The benefits of a faster barrier op will mostly be hidden by that overhead. However, as you've already implemented the barrier code in the backend, I'm fine with it. Please do take a look at the generated flatbuffer files though to ensure things are consistent.

@romerojosh thanks a lot for the reminder. I did miss one part in the fbs file. updated to reflect the correct schema.