use the ProactorEventLoop on windows

continuous_integration/environment-3.8.yaml

@@ -7,7 +7,6 @@ dependencies:
   - packaging
   - pip
   - asyncssh
-  - bokeh
   - click
   - cloudpickle
   - coverage
@@ -46,4 +45,5 @@ dependencies:
   - pip:
       - git+https://github.com/dask/dask
       - git+https://github.com/jcrist/crick  # Only tested here
+      - git+https://github.com/graingert/bokeh@13edfabc315e1df4082c73d4a198b13880795764
       - keras

continuous_integration/environment-3.9.yaml

@@ -8,7 +8,6 @@ dependencies:
   - packaging
   - pip
   - asyncssh
-  - bokeh
   - click
   - cloudpickle
   - coverage
@@ -53,6 +52,7 @@ dependencies:
       - git+https://github.com/dask/dask
       - git+https://github.com/dask/s3fs
       - git+https://github.com/dask/zict
+      - git+https://github.com/graingert/bokeh@13edfabc315e1df4082c73d4a198b13880795764
       # FIXME https://github.com/dask/distributed/issues/5345
       # - git+https://github.com/intake/filesystem_spec
       - git+https://github.com/joblib/joblib

distributed/config.py

@@ -1,4 +1,3 @@
-import asyncio
 import logging
 import logging.config
 import os
@@ -9,7 +8,7 @@
 import dask
 from dask.utils import import_required
 
-from .compatibility import WINDOWS, logging_names
+from .compatibility import logging_names
 
 config = dask.config.config
 
@@ -166,13 +165,8 @@ def initialize_event_loop(config):
             "    pip install uvloop",
         )
         uvloop.install()
-    elif event_loop in {"asyncio", "tornado"}:
-        if WINDOWS:
-            # WindowsProactorEventLoopPolicy is not compatible with tornado 6
-            # fallback to the pre-3.8 default of Selector
-            # https://github.com/tornadoweb/tornado/issues/2608
-            asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
-    else:
+        return
+    if event_loop not in {"asyncio", "tornado"}:
         raise ValueError(
             "Expected distributed.admin.event-loop to be in ('asyncio', 'tornado', 'uvloop'), got %s"
             % dask.config.get("distributed.admin.event-loop")

distributed/diagnostics/tests/test_task_stream.py

@@ -138,7 +138,7 @@ async def test_get_task_stream_plot(c, s, a, b):
     await wait(futures)
 
     data, figure = await c.get_task_stream(plot=True)
-    assert isinstance(figure, bokeh.plotting.Figure)
+    assert isinstance(figure, bokeh.plotting.figure)
 
 
 def test_get_task_stream_save(client, tmpdir):
@@ -153,4 +153,4 @@ def test_get_task_stream_save(client, tmpdir):
     assert "inc" in data
     assert "bokeh" in data
 
-    assert isinstance(ts.figure, bokeh.plotting.Figure)
+    assert isinstance(ts.figure, bokeh.plotting.figure)

requirements.txt

@@ -8,7 +8,7 @@ psutil >= 5.0
 sortedcontainers !=2.0.0, !=2.0.1
 tblib >= 1.6.0
 toolz >= 0.8.2
-tornado >= 6.0.3
+tornado >= 6.1
 zict >= 0.1.3
 pyyaml
 setuptools