[core] Raylet adds job to GCS after the driver port announced.

[rynewang]

GCS Client API GetAllJobInfo makes 1 RPC to each driver process for some up-to-date info. To make the call, GCS uses the driver core worker's IP and port address from the Raylet calling ray::gcs::JobInfoAccessor::AsyncAdd method. However Raylet may give a port = 0 in the call, causing GCS not able to access the driver's core worker.

Here is a time order in core worker init:

1. core worker starts
2. core worker -> raylet RegisterClientRequest
3. raylet -> GCS AddJob
4. core worker -> raylet AnnounceWorkerPort.

Note step 3 should actually happen after step 4, because at step 3, raylet does not yet know the core worker's real port. This can happen when the raylet has NodeManagerConfig::min_worker_port set to 0, allowing the core worker to pick a port on its own.

This PR moves the step 3 (raylet -> GCS AddJob) to the raylet function that handles incoming AnnounceWorkerPort message, and instead of the assigned_port, now we give GCS the real port.

One catch is that, previously AnnounceWorkerPort is one-way, core worker does not wait for the reply and continues to the user code immediately. However we do want it to wait until the GCS received the newly added job. We don't want to always add an RTT to all worker inits either, so we only add the reply message when the worker is a driver. Luckily, both sides knows the worker type.

Fixes #44459.

[rynewang]

Note: I expect this PR to not have any performance regressions. The driver needs to wait for the raylet -> GCS RPC to finish anyways in both before & after this PR.

[jjyao]

lg

[jjyao]

Many test failures.

[rynewang]

@jjyao ready to merge

[jjyao]

I still don't understand why we need to do this here?

[rynewang]

So I found after this PR, we got 1 more connection created in the test add node, and is not closed in test remove node. Then I found creating a node triggers gcs to connect with driver core workers, and that connection is persisted in core_worker_connection_pool in an LRU policy. Previously we did not have that because gcs had no way to connect to the driver (port is 0) which is exactly what we are fixing in this PR. So I pre-trigger the gcs->driver core worker connection to make it count in the initial list, so after the add node and remove node, the conn count is stable.

[jjyao]

We can move ray.init(cluster.address) down and add a corresponding ray.shutdown()?

[rynewang]

todo: just shutdown the driver job

[jjyao]

any reason why we need 4 tasks here instead of 1?

[rynewang]

just random. Switched back to only 1 task.

[jjyao]

I think we need some synchronization

[rynewang]

TODO: make it a signal actor enabling task

[jjyao]

In the original code, if this fails, RegisterClientReply will have

// Whether the registration succeeded.
  success: bool;
  // The reason of registration failure.
  failure_reason: string;

set. We should do the same thing for AnnounceWorkerPortReply?

[rynewang]

ok, added reply status. But at the terminal we check-fail it, since the cython binding notify_raylet returns void. Do we want to also change that to a raise exception?

[jjyao]

Why don't we call WriteMessageAsync here following

auto reply =
        ray::protocol::CreateRegisterClientReply(fbb,
                                                 status.ok(),
                                                 fbb.CreateString(status.ToString()),
                                                 to_flatbuf(fbb, self_node_id_),
                                                 assigned_port);
    fbb.Finish(reply);
    client->WriteMessageAsync(
        static_cast<int64_t>(protocol::MessageType::RegisterClientReply),
        fbb.GetSize(),
        fbb.GetBufferPointer(),
        [this, client](const ray::Status &status) {
          if (!status.ok()) {
            DisconnectClient(client,
                             rpc::WorkerExitType::SYSTEM_ERROR,
                             "Worker is failed because the raylet couldn't reply the "
                             "registration request.");
          }
        });

[rynewang]

makes sense. updated

[rynewang]

todo: add a msg to check

[jjyao]

do you know why the connection stays alive even after the driver exits?

[rynewang]

The prestarted worker is not specifically for job 01000000; it's for job = nil (ffffffff): code. So it's not killed after the job finished. Frankly I don't know what it does - the worker can't be used by any jobs right?

[jjyao]

Yea, initially the prestarted worker is not tied to a job but I thought later on when we start the actor, it will late bind to job 0100000

[jjyao]

Oh, those prestarted workers cannot be used for actors

void WorkerPool::PrestartDefaultCpuWorkers(ray::Language language, int64_t num_needed) {
  // default workers uses 1 cpu and doesn't support actor.
  static const WorkerCacheKey kDefaultCpuWorkerCacheKey{/*serialized_runtime_env*/ "",
                                                        {{"CPU", 1}},
                                                        /*is_actor*/ false,
                                                        /*is_gpu*/ false};