Repository: ray-project/ray
File: src/ray/gcs/actor/tests/gcs_actor_manager_test.cc (C++)

Code before:
   * 2. Create the actor
   * 3. Actor connecting to the gcs
   * Also verifies that the actor is discoverable.
   *
   * @param job_id The job ID.
   * @param actor_name The name of the actor.
   * @param ray_namespace The ray namespace.
   * @return The actor.
   */
  std::shared_ptr<gcs::GcsActor> RegisterAndCreateNamedActor(
      const JobID &job_id,
      const std::string &actor_name,
      const std::string &ray_namespace,
      const rpc::SendReplyCallback &send_reply_callback) {
    // Spinning up the actor
    auto register_request1 = GenRegisterActorRequest(job_id,
                                                     /*max_restarts=*/-1,
                                                     /*detached=*/false,
                                                     /*actor_name=*/actor_name,
                                                     /*ray_namespace=*/ray_namespace);

    rpc::RegisterActorReply register_reply1;
    gcs_actor_manager_->HandleRegisterActor(
        register_request1, &register_reply1, send_reply_callback);
    while (io_service_.poll_one()) {
    }

    rpc::CreateActorRequest create_request1;
    create_request1.mutable_task_spec()->CopyFrom(register_request1.task_spec());
    rpc::CreateActorReply create_reply1;
    gcs_actor_manager_->HandleCreateActor(
        create_request1, &create_reply1, send_reply_callback);

    std::shared_ptr<gcs::GcsActor> actor = mock_actor_scheduler_->actors.back();
    mock_actor_scheduler_->actors.pop_back();

    // Mock actor connecting to the gcs
    rpc::Address address = RandomAddress();
    actor->UpdateAddress(address);
    gcs_actor_manager_->OnActorCreationSuccess(actor, rpc::PushTaskReply());
    while (io_service_.poll_one()) {
    }
    RAY_CHECK_EQ(actor->GetState(), rpc::ActorTableData::ALIVE);

    // Verify the actor is discoverable
    rpc::GetNamedActorInfoRequest get_request1;
    get_request1.set_name(actor_name);
    get_request1.set_ray_namespace(ray_namespace);

    rpc::GetNamedActorInfoReply get_reply1;
    gcs_actor_manager_->HandleGetNamedActorInfo(

Review comment (suggestion):
I'm sketched out by this. Unless your intention is to wait for the entire io_service_ inside the GCS to drain (which is a hell of an intention), I would use a promise.
```C++
    std::promise<void> handle_actor_promise;
    gcs_actor_manager_->HandleRegisterActor(
        register_request1, &register_reply1, [&handle_actor_promise](){ handle_actor_promise.set_value(); });
    handle_actor_promise.get_future().wait();
```

Added benefit of not burning CPU in a busy loop. 

There's probably a follow-up for _someone_ to create an io_context class for tests that provide more readable methods like `wait_until_drained`. 

Code after:
   * Also verifies that the actor is discoverable.
   *
   * @param job_id The job ID.
   * @param actor_name The name of the actor.
   * @param ray_namespace The ray namespace.
   * @param actor[out] Set to the created actor on success.
   */
  void RegisterAndCreateNamedActor(const JobID &job_id,
                                   const std::string &actor_name,
                                   const std::string &ray_namespace,
                                   const rpc::SendReplyCallback &send_reply_callback,
                                   std::shared_ptr<gcs::GcsActor> &actor) {
    // Spinning up the actor
    auto register_request1 = GenRegisterActorRequest(job_id,
                                                     /*max_restarts=*/-1,
                                                     /*detached=*/false,
                                                     /*actor_name=*/actor_name,
                                                     /*ray_namespace=*/ray_namespace);

    rpc::RegisterActorReply register_reply1;
    gcs_actor_manager_->HandleRegisterActor(
        register_request1, &register_reply1, send_reply_callback);
    drain_io_context();

    rpc::CreateActorRequest create_request1;
    create_request1.mutable_task_spec()->CopyFrom(register_request1.task_spec());
    rpc::CreateActorReply create_reply1;
    gcs_actor_manager_->HandleCreateActor(
        create_request1, &create_reply1, send_reply_callback);

    actor = mock_actor_scheduler_->actors.back();
    mock_actor_scheduler_->actors.pop_back();

    // Mock actor connecting to the gcs
    rpc::Address address = RandomAddress();
    actor->UpdateAddress(address);
    gcs_actor_manager_->OnActorCreationSuccess(actor, rpc::PushTaskReply());
    drain_io_context();
    ASSERT_EQ(actor->GetState(), rpc::ActorTableData::ALIVE);

    // Verify the actor is discoverable
    rpc::GetNamedActorInfoRequest get_request1;
    get_request1.set_name(actor_name);
    get_request1.set_ray_namespace(ray_namespace);

    rpc::GetNamedActorInfoReply get_reply1;
    gcs_actor_manager_->HandleGetNamedActorInfo(
        get_request1, &get_reply1, send_reply_callback);
    ASSERT_EQ(get_reply1.status().code(), static_cast<int>(StatusCode::OK))
        << absl::StrFormat(
               "Failed to get named actor info for actor %s in namespace %s. "