Avoid is_coroutine_function calls in Server

[mrocklin]

We currently support handlers that are either synchronous or asynchronous

distributed/distributed/core.py

Lines 567 to 572 in 1297b18

	handler = self.stream_handlers[op]
	if is_coroutine_function(handler):
	self.loop.add_callback(handler, **merge(extra, msg))
	await gen.sleep(0)
	else:
	handler(**merge(extra, msg))

In #4443 we learned that these checks take up some time. In a recent run it takes 100ms out of about a 2.5s run, so around 4%

There are a couple of ways to address this:

1. We can try to find some other cheaper check
2. We can make all of our handlers async def functions and asyncify the entire codebase (we should also check that this doesn't incur other costs if so)

cc @dask/scheduler-performance

[fjetter]

We intend to revert the caching of this check in #5985

It's hard to follow and understand where the 4% / 100ms measurement comes from. Microbenchmarking does not reveal something like this

Python 3.8.12 | packaged by conda-forge | (default, Jan 30 2022, 23:13:24)
Type 'copyright', 'credits' or 'license' for more information
IPython 8.0.1 -- An enhanced Interactive Python. Type '?' for help.

In [1]: from tornado import gen

In [2]: from distributed.utils import iscoroutinefunction

In [3]: class A:
   ...:     async def method(self):
   ...:         await asyncio.sleep(0)
   ...:         return
   ...:
   ...:     @gen.coroutine
   ...:     def gen_coro(self):
   ...:         yield gen.moment()

In [4]: %timeit iscoroutinefunction(A.method)
434 ns ± 1.74 ns per loop (mean ± std. dev. of 7 runs, 1,000,000 loops each)

In [5]: %timeit iscoroutinefunction(A.gen_coro)
513 ns ± 2.64 ns per loop (mean ± std. dev. of 7 runs, 1,000,000 loops each)

In [6]: import inspect

In [7]: %timeit inspect.iscoroutinefunction(A.method)
395 ns ± 1.76 ns per loop (mean ± std. dev. of 7 runs, 1,000,000 loops each)

If we perceive this to be still an issue, I suggest to
A) Define all handlers as async
B) Specifically mark handlers as async or not (e.g. handlers dict would have a tuple (True, self.coroutinefunction); Ugly but effective, I suppose)

Before we do anything here, I would love to confirm this as an actual source of slowdown
cc @dask/scheduler-performance

[fjetter]

A third alternative to avoid the binding of the instance is to define handlers based on the class methods but not the instance methods, e.g. handlers = {"op": Worker.handler} instead of handlers = {"op": self.handler}. We'd need to modify handle_stream, etc. accordingly, of course

[mrocklin]

Yup, that could make sense. I recommend that we just go ahead with this for now though and see if any future work is actually necessary. (I suspect that this is also your plan)

[ian-r-rose]

There are some year-old tests on the average duration is_coroutine_function this in this comment: #4474 (comment). I don't know whether this would be reproducible on today's code base, but back then we see a few percent for a scheduler-bound computation.

At the time, I preferred moving to fully async stream handlers (that's what was done in #4474), but we went for caching to have a smaller API change.