[core] Introduces Postable for InternalKVInterface.

[rynewang]

InternalKVInterface is used by various GCS managers to talk to the internal KV Redis, and we did not pay too much attention in the thread safety issues. It was fine because everyone was on a same main thread & io context. But now we want to harden GCS by making managers thread safe.

To achieve this, introduce a new type Postable<R(Args...)>. It's just a std::function + an ioctx.

Methods:

• Post: post the function to the held ioctx.
• OnInvocation: adds a callback, invoked on Post. does not take any args and returns void.
• TransformArg: apply a mapper to the input argument, into a new Postable.
• Rebind: apply a function converter to allow more generic transformations, with the same ioctx.

The new stack:

• Managers: in their own io_ctx
• InternalKVInterface (implemented by StoreClientInternalKV)
  • used to take a std::function<R(T)>
  • (New!) take a Postable<R(T)> which is a function + an ioctx
• StoreClient (RedisStoreClient, InMemoryStoreClient, ObservableStoreClient)
  • used to take a std::function<R(T)>
  • (New!) take a Postable<R(T)> which is a function + an ioctx
• In the end of the call stack, invoke Postable by Post().

So a timeline of a call to internal kv may look like this:

1. From a Gcs Manager, calls into RedisContext
2. RedisContext dispatches a request_context->Run(); into internal kv ioctx
3. hiredis works in internal kv ioctx
4. hiredis calls back, dispatches to the Gcs Manager's ioctx, provided in Postable.

If it's not redis but InMemoryStoreClient, it's like this

• Managers: in their own io_ctx
• InternalKVInterface (implemented by StoreClientInternalKV)
  • used to take a std::function<R(T)>
  • (New!) take a Dispatchable<R(T)> which is a function + an ioctx
  • calls the underlying StoreClient with dispatched callbacks
• StoreClient (InMemoryStoreClient): use mutex, then posts to the callback caller's desired ioctx.

[dayshah]

what's the plan on testing / figuring out race conditions that could've been missed?

[dayshah]

return size_t to avoid cast

[rynewang]

we need to allow -1 to indicate "use default thread". do you think std::optional<size_t> is the way to go?

[dayshah]

if func_ has a capture, you'd be copying everything in the capture here

Moving the func also means you can't dispatch twice. Should set func_ to nullptr after dispatch to assure it isn't called again.

Also should it be a RAY_CHECK(_func) here instead as well, since if dispatch gets called without a valid function we won't get any fatal logs or anything, it'll just fail silently

[jjyao]

I think it's safe to say that a Dispatchable can only be dispatched once?

[dentiny]

It would be nice if we could document it clearly whether repeated invocation is expected / encouraged / acceptable.

[jjyao]

I think it's safe to say that a Dispatchable can only be dispatched once?

[rynewang]

Hi @jjyao @dentiny @dayshah, I updated this PR with the new Postable design. Sketch (how to review):

1. a new function_traits.h to handle std::function, lambdas and all other callables
2. postable.h defines a Postable. It can Rebind or TransformArg, or OnInvocation those are transformers to the func_ within the same ioctx.
3. All StoreClient and InternalKVInterface APIs are changed (std::function -> Postable). The middle layers transform the Postables, while the bottom layer (Redis, InMemory) invokes Post().
4. All callers of InternalKVInterface, that is all GCS*Managers, now pass in io_context_. Many of them used to not have one - they receive one from ctor
5. GcsServer updated to provide ioctx to those Managers' ctors.

It looks like a lot of code but many of changes in the Managers are extra indentation from a wrapping block: $lambda -> {$lambda, io_context_}.

[dayshah]

insane type gymnastics, generally lgtm