Improve robustness of async reactivity

[jcheng5]

Before this PR:

• Each WebSocket connection ran its own while/receive()/manageInputs/flush loop in its own independent asyncio.Task, so reactive invalidation/execution from two different Tasks could be interleaved. This level of concurrency makes the reactive loop extremely difficult to reason about, unless each Task's graph of reactive objects is guaranteed to be totally separate from any others (and maybe still problematic then--I haven't looked too closely at global state or ReactiveEnvironment state).
• In the reactive flush code, the default mode was "concurrent" execution of async observers, with an asyncio.gather() call at the end. I think this is also too aggressive of a default (although I could maybe be convinced otherwise).

After this PR:

• Each WebSocket still runs its own while/receive()/manageInputs/flush loop, but, only after acquiring an asyncio.Lock on the ReactiveEnvironment. (Currently, we don't prevent anyone from doing reactive reads/writes/flushes without holding the lock, so if you spawn a Task that could be a possibility.) The lock is acquired after an incoming message is deserialized, and released after all sessions' pending output is flushed.
• The introduction of an asyncio.Lock made it trivial to add invalidate_later support, so this PR does that too.
• The reactive flush code now only has a sequential mode.

This leaves us with an async execution model that is simple and robust--but not concurrent. It doesn't help you if you actually want to execute logic that doesn't block the (now serial) reactive loop. We're punting on that for right now as the high priority was robustness, and we can add opt-in concurrency later. Some notes regarding how we might want to approach that:

• Using asyncio.create_task() in an observer and then not awaiting it, seems like an easy way to get execution that's totally separate from the reactive loop, but it's actually quite subtle to get this right.
  • To be safe, an async non-blocking observer probably needs to read any reactive sources it needs while still under the global lock, do its work on a different task, then grab the global lock again before writing the results to anywhere reactive (thus triggering reactive invalidation). And it also probably then needs to trigger a reactive flush and session output flush, all under the lock.
  • An error during the execution of the task should be reflected in the owning session somehow.
  • Might (or might not) want to hold off on flushing of the owning session's output and/or processing of the owning session's incoming messages, until all sub-Tasks for that session are done.
  • This all points to some higher-level wrapper probably being needed to do this kind of thing.
• The common case could be that each session has totally independent reactive graphs, it would be neat to have a way for app authors to opt into one-loop-per-session (and throw warnings/errors if you "cross the streams" by having reactive objects shared across sessions).

[wch]

Following up an earlier discussion: I thought more about the possibility of deadlocks. I'm writing this out in part to clarify my thinking about it. (I think the code in this PR is safe.)

When there is just one lock, it is possible for there to be a deadlock, but only if a locked section of code calls (and awaits) other code which tries to acquire the lock.

import asyncio

lock = None

async def bar():
    async with lock:
        print("bar")

async def foo():
    global lock
    lock = asyncio.Lock()
    async with lock:
        print("foo")
        await bar() 

asyncio.run(foo()
#> foo
  [Python hangs]

If we create a task and await it inside the locked section, that also doesn't help:

async def foo2():
    global lock
    lock = asyncio.Lock()
    async with lock:
        print("foo2")
        await asyncio.create_task(bar())

asyncio.run(foo2())
#> foo2
  [Python hangs]

But if it's awaited outside of the locked section, then it's OK:

async def foo3():
    global lock
    lock = asyncio.Lock()
    async with lock:
        print("foo3")
        task = asyncio.create_task(bar())
    await task

asyncio.run(foo3())
#> foo3
#> bar

I think the code in the PR is safe, but this is something to keep in mind in the future.

[jcheng5]

Those are good points. The first can be dealt with using a “reentrant mutex” that allows a Task to reacquire a lock it already has (I don’t think asyncio.Lock is reentrant but it’d be trivial to write a wrapper).

It does feel super weird to run this much user code under a mutex, I have to admit.

[jcheng5]

TODO: Change OrderedDict usage in Callbacks/AsyncCallbacks to dict. Winston pointed out traversal order is guaranteed since Python 3.7 Done

[wch]

I think we want this to be:

Suggested change

	except BaseException:
	except Exception:

According to the exceptions docs:

Exception: All built-in, non-system-exiting exceptions are derived from this class. All user-defined exceptions should also be derived from this class.

Also see: https://stackoverflow.com/a/63169967/412655

[jcheng5]

Hmmm, even if I'm just printing and re-raising?

[wch]

Oh hm, good point.

[wch]

In addition to the code comment I made, it would be good to have tests for invalidate_later.

Other than those things, looks good!