psycogreen, gevent, eventlet, psycopg2 and django async orm

[allen-munsch]

Trying to determine what all is happening where, I'm just writing a summary of my research here.

What exactly is happening when django is using monkey patched code, and other parts of the code is using ASGI, and async.

WSGI mounted in ASGI
ASGI mounted in WSGI

And more specifically, what about using eventlet / gevent and a monkey patched psycopg2?

• https://github.com/psycopg/psycogreen/blob/master/psycogreen/gevent.py

psycopg2 reaches out to psycogreen.
psycogreen does a conn.poll(), and depending on the state waits on a read/write, or fails out

simply:

def patch_psycopg():
    """Configure Psycopg to be used with gevent in non-blocking way."""
    if not hasattr(extensions, 'set_wait_callback'):
        raise ImportError(
            "support for coroutines not available in this Psycopg version (%s)"
            % psycopg2.__version__
        )

    extensions.set_wait_callback(gevent_wait_callback)


def gevent_wait_callback(conn, timeout=None):
    """A wait callback useful to allow gevent to work with Psycopg."""
    while 1:
        state = conn.poll()
        if state == extensions.POLL_OK:
            break
        elif state == extensions.POLL_READ:
            wait_read(conn.fileno(), timeout=timeout)
        elif state == extensions.POLL_WRITE:
            wait_write(conn.fileno(), timeout=timeout)
        else:
            raise psycopg2.OperationalError("Bad result from poll: %r" % state)

What does psycopg2.extensions.set_wait_callback do?

• https://www.psycopg.org/docs/extensions.html#psycopg2.extensions.set_wait_callback

Register a callback function to block waiting for data.

The callback should have signature fun(conn) and is called to wait for data available whenever a blocking
function from the libpq is called. Use set_wait_callback(None) to revert to the original behaviour (i.e. using
blocking libpq functions).

The function is an hook to allow coroutine-based libraries (such as Eventlet or gevent) to switch when
Psycopg is blocked, allowing other coroutines to run concurrently.

See wait_select() for an example of a wait callback implementation.

So what does wait_read, and wait_write do?

def wait_read(fileno, timeout=None, timeout_exc=_NONE):
    """
    wait_read(fileno, timeout=None, [timeout_exc=None]) -> None
    Block the current greenlet until *fileno* is ready to read.
    For the meaning of the other parameters and possible exceptions,
    see :func:`wait`.
    .. seealso:: :func:`cancel_wait`
    """
    hub = get_hub()
    io = hub.loop.io(fileno, 1)
    try:
        return wait_on_watcher(io, timeout, timeout_exc, hub)
    finally:
        io.close()


def wait_write(fileno, timeout=None, timeout_exc=_NONE, event=_NONE):
    """
    wait_write(fileno, timeout=None, [timeout_exc=None]) -> None
    Block the current greenlet until *fileno* is ready to write.
    For the meaning of the other parameters and possible exceptions,
    see :func:`wait`.
    .. deprecated:: 1.1
       The keyword argument *event* is ignored. Applications should not pass this parameter.
       In the future, doing so will become an error.
    .. seealso:: :func:`cancel_wait`
    """
    # pylint:disable=unused-argument
    hub = get_hub()
    io = hub.loop.io(fileno, 2)
    try:
        return wait_on_watcher(io, timeout, timeout_exc, hub)
    finally:
        io.close()

Okay, so what about using sync_to_async ? How does that factor in here?

• https://github.com/django/asgiref/blob/main/asgiref/sync.py#L514
• https://github.com/django/asgiref/blob/d1ee83f29dd25831e78ddd20f2bd0779e22b896e/asgiref/sync.py#L304

sync_to_async takes up to 3 args

@overload
def sync_to_async(
    func: None = None,
    thread_sensitive: bool = True,
    executor: Optional["ThreadPoolExecutor"] = None,
) -> Callable[[Callable[..., Any]], SyncToAsync]:
    ...
    return SyncToAsync(
        func,
        thread_sensitive=thread_sensitive,
        executor=executor,
    )

With some thread sensitivity, and what not eventually it gets to:

        # Run the code in the right thread
        future = loop.run_in_executor(
            executor,
            functools.partial(
                self.thread_handler,
                loop,
                self.get_current_task(),
                sys.exc_info(),
                func,
                *args,
                **kwargs,
            ),
        )

• https://github.com/django/asgiref/blob/d1ee83f29dd25831e78ddd20f2bd0779e22b896e/asgiref/sync.py#L377-L433

According to https://docs.python.org/3/library/asyncio-eventloop.html#asyncio.loop.run_in_executor

The executor argument should be an concurrent.futures.Executor instance. The default executor is used if executor is None.

sync_to_async uses a :

from concurrent.futures import Future, ThreadPoolExecutor

How does the concurrent.futures.ThreadPoolExecutor work, and or differ from the gevent.threadpool.ThreadPoolExecutor?

• http://www.gevent.org/api/gevent.threadpool.html
• http://www.gevent.org/api/gevent.threadpool.html#gevent.threadpool.ThreadPoolExecutor

Bases: concurrent.futures.thread.ThreadPoolExecutor

A version of concurrent.futures.ThreadPoolExecutor that always uses native threads, even when threading is monkey-patched.

The Future objects returned from this object can be used with gevent waiting primitives like gevent.wait().

Curious ... does that mean wait_read, and wait_write as well?

Caution

If threading is not monkey-patched, then the Future objects returned by this object are not guaranteed to work with as_completed()
and wait(). The individual blocking methods like result() and exception() will always work.

New in version 1.2a1: This is a provisional API.

Takes the same arguments as concurrent.futures.ThreadPoolExecuter, which vary between Python versions.>
The first argument is always max_workers, the maximum number of threads to use. Most other arguments, while accepted, are ignored.

• https://github.com/gevent/gevent/blob/83c9e2ae5b0834b8f84233760aabe82c3ba065b4/src/gevent/threadpool.py#L748
• https://github.com/gevent/gevent/blob/83c9e2ae5b0834b8f84233760aabe82c3ba065b4/src/gevent/threadpool.py#L230
• https://github.com/gevent/gevent/blob/83c9e2ae5b0834b8f84233760aabe82c3ba065b4/src/gevent/threadpool.py#L675

So given all this information what is the summary?

• It seems like it should be possible to:

sync_to_async(executor=gevent.threadpool.ThreadPoolExecutor)

Feedback?

an untested branch : #7

[rednaks]

Oh I can see we pretty much addressing the same question as I did in #7 (comment)

so let me just look deeply into this and I'll give you my feedback.

Thank you for sharing your research !

[allen-munsch]

Aye, from Andrew Godwin django/asgiref#264

Looks like there's some hang ups on getting this to work. I don't really have the bandwidth or in-depth knowledge to get it hacked together.

A branch where I tried a drop-in replacement of the default sync_to_async executor #7