Method FuseRecvTensorAsync

tensorflow/contrib/star/star_worker_service.cc:501–617

}

void StarWorker::FuseRecvTensorAsync(CallOptions* opts,
                                     const FuseRecvTensorRequest* request,
                                     StarFuseTensorResponse* response,
                                     StatusCallback done) {
    const int64 step_id = request->step_id();
    int fuse_count = request->rendezvous_key_size();
    std::vector<Rendezvous::ParsedKey> parsed_keys(fuse_count);
    std::vector<Device*>* src_devs = new std::vector<Device*>(fuse_count, nullptr);

    for (int idx = 0; idx < fuse_count; ++idx) {
      const string& key = request->rendezvous_key(idx);
      Status s = Rendezvous::ParseKey(key, &parsed_keys[idx]);
      // LOG(INFO) << "parsed_keys at index " << idx << " is " << parsed_keys[idx].FullKey()
      //          << " incarnation is " << parsed_keys[idx].src_incarnation;
      if (s.ok()) {
        s = PrepareRecvTensor(parsed_keys[idx], &(*src_devs)[idx]);
      }

      if (!s.ok()) {
        LOG(ERROR) << "PrepareRecvTensor failed, tensor:" << key;
        delete src_devs;
        done(s);
        return;
      }
    }

    // make call opts useful.
    // opts->SetCancelCallback([this, step_id]() { AbortStep(step_id); });
    env_->rendezvous_mgr->FuseRecvLocalAsync(
      step_id, parsed_keys,
      [opts, request, response, done, fuse_count, src_devs](
          const Status& status,
          const std::vector<Rendezvous::Args>& send_argses,
          const Rendezvous::Args& recv_args,
          const std::vector<Tensor>& vals,
          const std::vector<bool>& is_deads) {
        // opts->ClearCancelCallback();

        if (!status.ok()) {
          LOG(ERROR) << "env_->rendezvous_mgr->FuseRecvLocalAsync failed, error msg is: "
                     << status.error_message();
        }

        if (status.ok()) {
          response->Init(fuse_count);
          int *fuse_counter = new int(fuse_count);

          for (int idx = 0; idx < fuse_count; ++idx) {
            response->SetIsDeadByIndex(idx, is_deads[idx]);
            bool can_memcpy = DataTypeCanUseMemcpy(vals[idx].dtype());

            if ((*src_devs)[idx]->tensorflow_gpu_device_info() &&
                (!send_argses[idx].alloc_attrs.on_host())) {
#if GOOGLE_CUDA
              CHECK(send_argses[idx].device_context)
                << "send dev name: " << (*src_devs)[idx]->name()
                << " gpu_info: " << (*src_devs)[idx]->tensorflow_gpu_device_info();