Skip to content
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.

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

Ucx process bootstrap #598

Open
wants to merge 9 commits into
base: main
Choose a base branch
from
Open

Ucx process bootstrap #598

wants to merge 9 commits into from

Conversation

kaiyingshan
Copy link
Collaborator

@kaiyingshan kaiyingshan marked this pull request as ready for review August 29, 2022 02:54
@nirandaperera
Copy link
Collaborator

@kaiyingshan can you send this commit 19d17b7 as a separate PR to the main branch?

redis->lpush("ucc_helper" + std::to_string(num_comm) + ":" + std::to_string(rank),
"0");

for (int i = 0; i < world_size; i++) {
Copy link
Collaborator

Choose a reason for hiding this comment

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

we might need a barrier like mechanism here to guarantee that all processes have completed pushing data.

void *coll_info, void **req) {
int world_size = ((UCXUCCCommunicator*) coll_info)->GetWorldSize();
int rank = ((UCXUCCCommunicator*) coll_info)->GetRank();
int num_comm = ((UCXUCCCommunicator *)coll_info)->num_oob_allgather;
Copy link
Collaborator

Choose a reason for hiding this comment

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

why do we need this num_comm?

Comment on lines +255 to +256
this->displacements_ = new std::vector<std::vector<int>>(num_buffers);
this->all_recv_counts_ = new std::vector<std::vector<int>>(num_buffers);
Copy link
Collaborator

Choose a reason for hiding this comment

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

I dont think you need to malloc for a vector object. You can simply do this,

this->displacements_ = std::vector<std::vector<int>>(num_buffers);
this->all_recv_counts_ = std::vector<std::vector<int>>(num_buffers);

Copy link
Collaborator

Choose a reason for hiding this comment

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

Oh I think I understand why you did this. This is because of the const in GatherBufferSizes method, right?

@@ -259,17 +262,25 @@ Status UccTableGatherImpl::GatherBufferSizes(const int32_t *send_data, int32_t n
ucc_coll_req_h req;

args.mask = 0;
args.coll_type = UCC_COLL_TYPE_GATHER;
args.coll_type = UCC_COLL_TYPE_ALLGATHER;
Copy link
Collaborator

Choose a reason for hiding this comment

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

put a TODO and refer to the github issue.
Add a comment explaining why we are using allgather

Comment on lines +337 to +341
int sum = 0;
auto& recv_counts_ = (*all_recv_counts_)[buf_idx];
for(auto count: recv_counts_) {
sum += count;
}
Copy link
Collaborator

Choose a reason for hiding this comment

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

use std::accumulate

for (int32_t i = 0; i < num_buffers; ++i) {
(*all_recv_counts_)[i] = cylon::net::receiveCounts(all_buffer_sizes, i,
num_buffers, world_size);
(*displacements_)[i] = std::move(cylon::net::displacementsPerBuffer(all_buffer_sizes, i,
Copy link
Collaborator

Choose a reason for hiding this comment

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

don't think you need the move here.

Comment on lines 275 to 286
if(rank == gather_root) {
args.dst.info.buffer = rcv_data;
args.dst.info.count = num_buffers * world_size;
args.dst.info.count = total_sz;
args.dst.info.datatype = UCC_DT_INT32;
args.dst.info.mem_type = UCC_MEMORY_TYPE_HOST;
} else {
all_buffer_sizes.resize(total_sz);
args.dst.info.buffer = all_buffer_sizes.data();
args.dst.info.count = total_sz;
args.dst.info.datatype = UCC_DT_INT32;
args.dst.info.mem_type = UCC_MEMORY_TYPE_HOST;
}
Copy link
Collaborator

Choose a reason for hiding this comment

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

can we do this?

  if(rank == gather_root) {
    args.dst.info.buffer = rcv_data;
  } else {
    all_buffer_sizes.resize(total_sz);
    args.dst.info.buffer = all_buffer_sizes.data();
  }
args.dst.info.count = total_sz;
args.dst.info.datatype = UCC_DT_INT32;
args.dst.info.mem_type = UCC_MEMORY_TYPE_HOST;

Comment on lines +345 to +348
args.dst.info_v.counts = (ucc_count_t *)(*all_recv_counts_)[buf_idx].data();
args.dst.info_v.displacements = (ucc_aint_t *)(*displacements_)[buf_idx].data();
args.dst.info_v.datatype = UCC_DT_UINT8;
args.dst.info_v.mem_type = UCC_MEMORY_TYPE_HOST;
Copy link
Collaborator

Choose a reason for hiding this comment

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

I actually think we dont need these! we just need to pass recv_data_placeholder buffer and make sure everyone (other than the root) copies data to this dummy buffer. So you might not even need to track all_recv_counts_ and displacements_.

Comment on lines +74 to +75
std::vector<std::vector<int>>* displacements_;
std::vector<std::vector<int>>* all_recv_counts_;
Copy link
Collaborator

Choose a reason for hiding this comment

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

I have a feeling that you might not even need these. I've explained it in a previous comment

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
None yet
Projects
None yet
Development

Successfully merging this pull request may close these issues.

2 participants