[core][1/N] Make executor to be a long-running Python thread

[kevin85421]

Why are these changes needed?

Currently, tasks are sometimes executed on the main thread and sometimes on other threads which is created by C++. However, C++ threads are only considered Python threads when they execute Python functions. Hence, when a task finishes, the thread-local state will be garbage-collected. See #46336 (comment) for more details.

This PR uses the CPython API to treat C++ threads as Python threads, even if they do not execute Python functions.
https://docs.python.org/3/c-api/init.html#non-python-created-threads

This PR handles synchronous tasks. After this PR is merged, the follow-up tasks are:

• Always create a default executor.
• Move the constructor to the default executor.
• Support asynchronous tasks.

Related issue number

Part of #46336

Checks

• I've signed off every commit(by using the -s flag, i.e., git commit -s) in this PR.
• I've run scripts/format.sh to lint the changes in this PR.
• I've included any doc changes needed for https://docs.ray.io/en/master/.
  • I've added any new APIs to the API Reference. For example, if I added a
    method in Tune, I've added it in doc/source/tune/api/ under the
    corresponding .rst file.
• I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
• Testing Strategy
  • Unit tests
  • Release tests
  • This PR is not tested :(

[aslonnie]

discussed offline:

• setup.py seems to always overwrite PYTHON3_BIN_PATH https://github.com/ray-project/ray/blob/master/python/setup.py#L537 , so it is probably something else that is missing the env var (rather than the wheel building). please investigate a bit more on why.
• the env var, if needed, should be set in scripts like https://github.com/ray-project/ray/blob/master/python/build-wheel-manylinux2014.sh , rather than in the python wrapper. this will make sure that scripts like https://github.com/ray-project/ray/blob/master/python/README-building-wheels.md will keep working.

[kevin85421]

TODO: undefined symbol for Java

[edoakes]

GetThreadReleaser should be defined as another cdef function in this file instead of in core worker cpp code (that is another abstraction leak similar to the original PR)

[kevin85421]

Replied in #50644 (comment). I agree that it is the best case if we can avoid the C++ util function.

Would you mind sharing more details why it's an abstraction leak?

1. GetThreadReleaser is exclusively used by Cython and is not included in the shared code path for any other language frontends.
2. It doesn't add any new dependencies.
3. There is a directory core_worker/lib/java.

[edoakes]

core_worker/lib/java is just autogenerated bindings for the core worker API (it could/should be moved elsewhere)

[edoakes]

This doesn't need to exist -- you can define it in cython and pass it down

[kevin85421]

To the best of my knowledge, this may not be possible in Cython. I haven't seen any documentation that explicitly states it's impossible. I tried looking for examples on GitHub, but I didn't find any way to capture gstate and return a callable.

I tried multiple ways to capture gstate and return a callable, but I failed. In addition, this method is referenced in the unit tests for Cython's <functional>. You can take a look at _get_function_ptr_from_name in cython/cython@954f0ec.

I'm not familiar with Cython, but this is my current guess based on my investigation. It would be helpful if we could avoid defining this C++ function. Would you mind sharing more details if you have related experiences?

[edoakes]

you should be able to use std::bind and std::placeholder1 to bind the gstate as an arg and then pass the lambda into the c++ layer

or alternatively (maybe easier?) you can wrap the gstate as a unique_ptr<void*>, pass it into c++, then cast it back to gstate in the callback. we do something similar currently for the callback passed to c++ for async get:

ray/python/ray/_raylet.pyx

Line 4717 in e0f7bbc

def set_get_async_callback(self, ObjectRef object_ref, user_callback: Callable):

@israbbani has a follow-up to avoid the manual incref/decref on that codepath and use a unique_ptr instead

[kevin85421]

I use std::bind and it works! Initially, I tried using a lambda function to capture gstate. I didn't know that std::bind existed. In addition, I haven't seen any CPython examples where users need to control the reference count of PyGILState_STATE.

[edoakes]

Looks good, thanks for figuring out the tricky cython bits! All style nits now.

Please add a test for concurrency_group_manager that encodes the expected semantics of the initialize_thread_callback (called before thread executes anything, the returned callback is called on shutdown).

Are we certain that the gstate object does not need to be reference counted? Just want to make sure there is no concern of crashing during the shutdown sequence due to a memory corruption issue.

[kevin85421]

Please add a test for concurrency_group_manager that encodes the expected semantics of the initialize_thread_callback (called before thread executes anything, the returned callback is called on shutdown).

Test added. ee93309

[kevin85421]

Are we certain that the gstate object does not need to be reference counted? Just want to make sure there is no concern of crashing during the shutdown sequence due to a memory corruption issue.

I will conduct some experiments to verify it.

[edoakes]

Great work!

[edoakes]

Let me know when ready for merge. DCO is failing btw.

[kevin85421]

I followed the instructions on the DCO page to fix the DCO issue, but somehow it messed up the commits. I will open a new PR instead.

[kevin85421]

Open a new PR: #51005

[kevin85421]

Open a new PR: #51016