tensorflow/compiler/xla/pjrt/distributed/client_server_test.cc
// We should eventually time out if a client does not show up.
We should eventually time out if a client does not show up.
We should eventually time out if a client does not show up .
TEST_P(ClientServerTest, ConnectEventuallyTimesOutIfAClientDoesNotShowUp) {
  int num_nodes = 3;
  DistributedRuntimeServiceImpl::Options service_options;
  service_options.num_nodes = num_nodes;
  StartService(service_options, GetParam().use_coordination_service);
  auto thread_fn = [&](int node_id) -> xla::Status {
    DistributedRuntimeClient::Options client_options;
    client_options.node_id = node_id;
    client_options.init_timeout = absl::Milliseconds(500);
    client_options.rpc_timeout = absl::Milliseconds(200);
    auto client = GetDistributedRuntimeClient(
        server_->InProcessChannel(::grpc::ChannelArguments()), client_options);
    TF_RETURN_IF_ERROR(client->Connect());
    TF_RETURN_IF_ERROR(client->Shutdown());
    return xla::Status::OK();
  };
  std::vector<xla::Status> statuses(num_nodes - 1);
  {
    tensorflow::thread::ThreadPool thread_pool(tensorflow::Env::Default(),
                                               "test_threads", num_nodes);
    for (int i = 0; i < num_nodes - 1; ++i) {
      thread_pool.Schedule([&, i]() { statuses[i] = thread_fn(i); });
    }
  }
  for (int i = 0; i < num_nodes - 1; ++i) {
    EXPECT_EQ(statuses[i].code(), tensorflow::error::DEADLINE_EXCEEDED);
  }
}
TEST_P ( ClientServerTest , ConnectEventuallyTimesOutIfAClientDoesNotShowUp ) { int num_nodes = 3 ; DistributedRuntimeServiceImpl :: Options service_options ; service_options . num_nodes = num_nodes ; StartService ( service_options , GetParam ( ) . use_coordination_service ) ; auto thread_fn = [ & ] ( int node_id ) -> xla :: Status { DistributedRuntimeClient :: Options client_options ; client_options . node_id = node_id ; client_options . init_timeout = absl :: Milliseconds ( 500 ) ; client_options . rpc_timeout = absl :: Milliseconds ( 200 ) ; auto client = GetDistributedRuntimeClient ( server_ -> InProcessChannel ( :: grpc :: ChannelArguments ( ) ) , client_options ) ; TF_RETURN_IF_ERROR ( client -> Connect ( ) ) ; TF_RETURN_IF_ERROR ( client -> Shutdown ( ) ) ; return xla :: Status :: OK ( ) ; } ; std :: vector < xla :: Status > statuses ( num_nodes - 1 ) ; { tensorflow :: thread :: ThreadPool thread_pool ( tensorflow :: Env :: Default ( ) , " " , num_nodes ) ; for ( int i = 0 ; i < num_nodes - 1 ; ++ i ) { thread_pool . Schedule ( [ & , i ] ( ) { statuses [ i ] = thread_fn ( i ) ; } ) ; } } for ( int i = 0 ; i < num_nodes - 1 ; ++ i ) { EXPECT_EQ ( statuses [ i ] . code ( ) , tensorflow :: error :: DEADLINE_EXCEEDED ) ; } }
We should eventually time out if a client does not show up.
We should eventually time out if a client does not show up .
// Note: one fewer thread than 'num_nodes'.