Use cached version of is_coroutine_function in stream handling to (#4481

)

mitigate re-checking the same function many times.

Decorate function definition, cache all function definitions.

Oops, cache is python 3.9-only

Alias duplicate function.

Use distributed's version of this util.

bad rebase.

distributed/actor.py

@@ -1,12 +1,11 @@
 import asyncio
 import functools
-from inspect import iscoroutinefunction
 import threading
 from queue import Queue
 
 from .client import Future, default_client
 from .protocol import to_serialize
-from .utils import thread_state, sync
+from .utils import iscoroutinefunction, thread_state, sync
 from .utils_comm import WrappedKey
 from .worker import get_worker
 

distributed/utils.py

@@ -1246,6 +1246,7 @@ def color_of(x, palette=palette):
     return palette[n % len(palette)]
 
 
+@functools.lru_cache(None)
 def iscoroutinefunction(f):
     return inspect.iscoroutinefunction(f) or gen.is_coroutine_function(f)
 
@@ -1343,8 +1344,7 @@ def parse_ports(port):
     return ports
 
 
-def is_coroutine_function(f):
-    return asyncio.iscoroutinefunction(f) or gen.is_coroutine_function(f)
+is_coroutine_function = iscoroutinefunction
 
 
 class Log(str):