tensorflow/compiler/xla/pjrt/distributed/client_server_test.cc
// We should eventually connect, even if some clients are late to show up.
We should eventually connect, even if some clients are late to show up.
We should eventually connect even if some clients are late to show up .
TEST_P(ClientServerTest, LateClientsAreOk) {
  int num_nodes = 3;
  DistributedRuntimeServiceImpl::Options service_options;
  service_options.num_nodes = num_nodes;
  StartService(service_options, GetParam().use_coordination_service);
  absl::Barrier barrier(num_nodes);
  auto thread_fn = [&](int node_id) -> xla::Status {
    DistributedRuntimeClient::Options client_options;
    client_options.node_id = node_id;
    client_options.init_timeout = absl::Milliseconds(20000);
    client_options.rpc_timeout = absl::Milliseconds(200);
    auto client = GetDistributedRuntimeClient(
        server_->InProcessChannel(::grpc::ChannelArguments()), client_options);
    barrier.Block();
    absl::SleepFor(absl::Milliseconds(200) * node_id);
    TF_RETURN_IF_ERROR(client->Connect());
    TF_RETURN_IF_ERROR(client->Shutdown());
    return xla::Status::OK();
  };
  std::vector<xla::Status> statuses(num_nodes);
  {
    tensorflow::thread::ThreadPool thread_pool(tensorflow::Env::Default(),
                                               "test_threads", num_nodes);
    for (int i = 0; i < num_nodes; ++i) {
      thread_pool.Schedule([&, i]() { statuses[i] = thread_fn(i); });
    }
  }
  for (int i = 0; i < num_nodes; ++i) {
    TF_EXPECT_OK(statuses[i]);
  }
}
TEST_P ( ClientServerTest , LateClientsAreOk ) { int num_nodes = 3 ; DistributedRuntimeServiceImpl :: Options service_options ; service_options . num_nodes = num_nodes ; StartService ( service_options , GetParam ( ) . use_coordination_service ) ; absl :: Barrier barrier ( num_nodes ) ; auto thread_fn = [ & ] ( int node_id ) -> xla :: Status { DistributedRuntimeClient :: Options client_options ; client_options . node_id = node_id ; client_options . init_timeout = absl :: Milliseconds ( 20000 ) ; client_options . rpc_timeout = absl :: Milliseconds ( 200 ) ; auto client = GetDistributedRuntimeClient ( server_ -> InProcessChannel ( :: grpc :: ChannelArguments ( ) ) , client_options ) ; barrier . Block ( ) ; absl :: SleepFor ( absl :: Milliseconds ( 200 ) * node_id ) ; TF_RETURN_IF_ERROR ( client -> Connect ( ) ) ; TF_RETURN_IF_ERROR ( client -> Shutdown ( ) ) ; return xla :: Status :: OK ( ) ; } ; std :: vector < xla :: Status > statuses ( num_nodes ) ; { tensorflow :: thread :: ThreadPool thread_pool ( tensorflow :: Env :: Default ( ) , " " , num_nodes ) ; for ( int i = 0 ; i < num_nodes ; ++ i ) { thread_pool . Schedule ( [ & , i ] ( ) { statuses [ i ] = thread_fn ( i ) ; } ) ; } } for ( int i = 0 ; i < num_nodes ; ++ i ) { TF_EXPECT_OK ( statuses [ i ] ) ; } }
We should eventually connect, even if some clients are late to show up.
We should eventually connect even if some clients are late to show up .