internal review #6
Conversation
| auto currentTimepoint = std::chrono::steady_clock::now(); | ||
| auto timeElapsed = std::chrono::duration_cast<std::chrono::milliseconds>( | ||
| currentTimepoint - workStartTime_); | ||
| std::chrono::milliseconds opTimeout = std::chrono::milliseconds(60000); |
There was a problem hiding this comment.
where do you use it?
| @@ -67,17 +67,15 @@ ccl::reduction getXcclReduceOp(const ReduceOp& reduceOp, at::Tensor& input) { | |||
| return xcclOps.at(reduceOp); | |||
| } catch (const std::out_of_range&) { | |||
| switch (reduceOp) { | |||
| : Work(rank, opType, "profilingTitle", inputs), | ||
| device_(device), | ||
| workStartTime_(std::chrono::steady_clock::now()) { | ||
| unsigned char enable_timing = 0; |
There was a problem hiding this comment.
If you always set it as 0, then we don't need to keep it, right?
There was a problem hiding this comment.
Yes, Defining this variable serves as a form of annotation, informing reviewers and users that 0 represents the state of enable_timing, which is meaningful.
| "Work ran for ", | ||
| timeElapsed.count(), | ||
| " milliseconds before timing out."); | ||
| TORCH_CHECK(false, exceptionMsg) |
There was a problem hiding this comment.
TORCH_CHECK(false, exceptionMsg);
abort();
| } // namespace | ||
|
|
||
| static std::mutex xcclCommDevIdxMapMutex; | ||
| static std::unordered_map<std::shared_ptr<xcclComm_t>, int> xcclCommDevIdxMap; |
There was a problem hiding this comment.
Those static variables are not used in your code. Please check.
| blockingWait_ = getCvarBool(TORCH_XCCL_BLOCKING_WAIT, false); | ||
| init(); | ||
|
|
||
| // Intel oneCCL requires passing CCL_LOCAL_RANK and CCL_LOCAL_SIZE for non-MPI |
There was a problem hiding this comment.
More comment for why we use LOCAL_RANK and LOCAL_WORLD_SIZE.
| std::shared_ptr<xcclComm_t> ProcessGroupXCCL::getXCCLComm( | ||
| const std::string& deviceKey, | ||
| at::Device& device) { | ||
| if (deviceKey.empty()) { |
There was a problem hiding this comment.
C10_THROW_ERROR_WITH
| devXCCLCommMap_.emplace(deviceKey, XCCLComm); | ||
| } | ||
|
|
||
| xcclStreamsMap_.emplace(deviceKey, std::move(stream)); |
There was a problem hiding this comment.
so xcclEventsMap does not needed?
| PreProcess pre, | ||
| PostProcess post, | ||
| OpType opType) { | ||
| using traits = function_traits<Fn>; |
There was a problem hiding this comment.
which collective need attribute as a must?
There was a problem hiding this comment.
Yes, allgather meet build error
| for (const auto i : c10::irange(inputs.size())) { | ||
| c10::xpu::XPUCachingAllocator::recordStream( | ||
| inputs[i].storage().data_ptr(), stream); | ||
| fn(inputs[i], outputs[i], attr, *comm, stream); |
There was a problem hiding this comment.
add comment for output record stream.
| false, "ProcessGroupXCCL::WorkXCCL::isSuccess not implemented"); | ||
| } | ||
|
|
||
| void abort() override { |
|
|
||
| bool isCompleted() override; | ||
|
|
||
| bool isSuccess() const override { |
| case ReduceOp::BXOR: | ||
| C10_THROW_ERROR(ValueError, "Cannot use ReduceOp.BXOR with NCCL"); | ||
| C10_THROW_ERROR( |
There was a problem hiding this comment.
don't change NCCL now.
|
|
||
| c10::impl::VirtualGuardImpl impl(device.type()); | ||
| c10::Stream stream = impl.getStream(device); | ||
| sycl::queue& q = c10::xpu::XPUStream(stream).queue(); |
There was a problem hiding this comment.
It's a big bug to use current stream as communication stream.
| int rank, | ||
| OpType opType, | ||
| const std::optional<std::vector<at::Tensor>>& inputs) | ||
| : Work(rank, opType, "profilingTitle", inputs), |
| @@ -126,6 +131,13 @@ class TORCH_API ProcessGroup : public torch::CustomClassHolder { | |||
| return backendType_; | |||
| }; | |||
|
|
|||
| inline bool backendSupportsSequenceNumbers(BackendType backendType) { | |||
| if (backendType == BackendType::GLOO || backendType == BackendType::NCCL || | |||
| backendType == BackendType::XCCL || backendType == BackendType::UCC) | |||
There was a problem hiding this comment.
Do you make sure that we need to support this sequence number?
There was a problem hiding this comment.
Sequence number used by RECORD_PARAM_COMMS. so we need it
| @@ -180,7 +181,8 @@ def skip_if_lt_x_gpu(x): | |||
| def decorator(func): | |||
| @wraps(func) | |||
| def wrapper(*args, **kwargs): | |||
| if torch.cuda.is_available() and torch.cuda.device_count() >= x: | |||
| if (torch.cuda.is_available() and torch.cuda.device_count() >= x) or \ | |||
There was a problem hiding this comment.
Don't use if for accelerator related check
Fixes #ISSUE_NUMBER