[core] Async APIs for the New GcsClient.

[rynewang]

Implements async binding of C++ GcsClient in Python as NewGcsAioClient.

Previously we only have sync GcsClient bindings, ones that blocks on completion. To facilitate GcsAioClient we use python thread pool executor - one dedicated thread blocked for the sync call, whose underlying API is async. This is a big waste and we can do better.

The trick is to play the callback-to-async games wisely. Invoke a C++ async API with a python callback function; the callback serializes the reply data or exception, then switch to the python asyncio thread and complete a future. The future, in turn, is awaited by a converter function that does any python-side treatment (e.g. python protobuf deserialization, or converting to dict), then pass on to user code. The end result is an async method just like Python-native ones.

This PR adds the NewGcsAioClient and uses it as implementation of GcsAioClient by default. Can switch back to the OldGcsAioClient by RAY_USE_OLD_GCS_CLIENT=1.

[jjyao]

Ideally we can share the same underlying GCSClient?

[rynewang]

yes maybe in a next PR

[jjyao]

LG!

[jjyao]

What do you want to say here? I feel these comments add confusion rather than clarity here.

[rynewang]

reworded.

[jjyao]

This result PyObject will be freed by Python, like there is no leak here?

[rynewang]

yes the result is managed by CPython ref counts, we don't need to do anything here. A rule of thumb is: if you don't Py_INCREF you don't need to Py_DECREF.

[jjyao]

Suggested change

	cdef void assign_and_decrement(result, void* fut_ptr):
	cdef void assign_and_decrement_fut(result, void* fut_ptr):

[jjyao]

Please also update PR title and description.

[jjyao]

In theory, this is never optional right?

[rynewang]

If status is not OK, the optional[int] would be nullopt. Otherwise, right it should never be optional.

[rynewang]

added assert

[jjyao]

I feel we should handle optional here. It's error-prone that this method is convert_optional_bool but it assumes it's never optional?

[jjyao]

Status InternalKVAccessor::Put(const std::string &ns,
                               const std::string &key,
                               const std::string &value,
                               bool overwrite,
                               const int64_t timeout_ms,
                               bool &added) {
  std::promise<Status> ret_promise;
  RAY_CHECK_OK(AsyncInternalKVPut(
      ns,
      key,
      value,
      overwrite,
      timeout_ms,
      [&ret_promise, &added](Status status, std::optional<int> added_num) {
        added = static_cast<bool>(added_num.value_or(0));
        ret_promise.set_value(status);
      }));
  return ret_promise.get_future().get();
}

For example, this handles the optional

[rynewang]

ok I will assert for each optional

[rynewang]

that optional check would never be used because its caller always populates:

  client_impl_->GetGcsRpcClient().InternalKVPut(
      req,
      [callback](const Status &status, const rpc::InternalKVPutReply &reply) {
        callback(status, reply.added_num());
      },
      timeout_ms);

but nonetheless I added assert for all optional converters