| 396 | } |
| 397 | |
| 398 | void RingAlg::DispatchRecv(RingField* rf, const StatusCallback& done) { |
| 399 | DCHECK(rf->do_recv); |
| 400 | string recv_buf_key = |
| 401 | RingAlgBufKey(name_, col_ctx_->exec_key, rf->second_pass, rf->sc_idx, |
| 402 | (rf->rank + (group_size_ - 1)) % group_size_); |
| 403 | VLOG(3) << "DispatchRecv rank=" << col_params_->default_rank << " recv key " |
| 404 | << recv_buf_key << " chunk " << ca_->TBounds(rf->chunk) << " into " |
| 405 | << ((col_params_->merge_op != nullptr) ? "tmp_chunk" : "chunk"); |
| 406 | Tensor* dst_tensor = (!rf->second_pass && (col_params_->merge_op != nullptr)) |
| 407 | ? &rf->tmp_chunk |
| 408 | : &rf->chunk; |
| 409 | col_ctx_->col_exec->RecvFromPeer( |
| 410 | col_params_->instance.device_names[rf->recv_dev_idx], |
| 411 | col_params_->instance.task_names[rf->recv_dev_idx], |
| 412 | col_params_->task.is_local[rf->recv_dev_idx], recv_buf_key, |
| 413 | col_ctx_->device, col_ctx_->op_ctx->op_device_context(), |
| 414 | col_ctx_->op_ctx->output_alloc_attr(0), dst_tensor, |
| 415 | col_ctx_->device_locality, rf->subdiv_idx, done); |
| 416 | } |
| 417 | |
| 418 | string RingAlg::FieldState() { |
| 419 | string s = strings::StrCat( |
nothing calls this directly
no test coverage detected