=================================== FAILURES ===================================
______________________________ test_lots_of_tasks ______________________________

c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:52907', workers: 0, cores: 0, tasks: 0>
a = <Worker 'tcp://127.0.0.1:52908', name: 0, status: closed, stored: 0, running: 0/1, ready: 0, comm: 0, waiting: 0>
b = <Worker 'tcp://127.0.0.1:52910', name: 1, status: closed, stored: 0, running: 0/2, ready: 0, comm: 0, waiting: 0>

    @gen_cluster(
        client=True,
        scheduler_kwargs={"dashboard": True},
        config={
            "distributed.scheduler.dashboard.tasks.task-stream-length": 10,
            "distributed.scheduler.dashboard.status.task-stream-length": 10,
        },
    )
    async def test_lots_of_tasks(c, s, a, b):
        import tlz as toolz
    
        ts = TaskStream(s)
        ts.update()
        futures = c.map(toolz.identity, range(100))
        await wait(futures)
    
        tsp = s.plugins[TaskStreamPlugin.name]
        assert len(tsp.buffer) == 10
        ts.update()
>       assert len(ts.source.data["start"]) == 10
E       assert 11 == 10
E        +  where 11 = len(array([ 1.54638672,  0.        ,  0.99829102,  7.20776367,  5.12182617,\n        6.47998047, 83.5144043 , 49.33178711, 79.44384766, 82.61474609,\n       89.00756836]))

distributed/dashboard/tests/test_scheduler_bokeh.py:873: AssertionError
----------------------------- Captured stdout call -----------------------------
Dumped cluster state to test_cluster_dump/test_lots_of_tasks.yaml
----------------------------- Captured stderr call -----------------------------
2022-04-26 16:02:35,377 - distributed.worker - WARNING - Heartbeat to scheduler failed
Traceback (most recent call last):
  File "/Users/runner/work/distributed/distributed/distributed/comm/tcp.py", line 226, in read
    frames_nbytes = await stream.read_bytes(fmt_size)
tornado.iostream.StreamClosedError: Stream is closed

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/Users/runner/work/distributed/distributed/distributed/worker.py", line 1150, in heartbeat
    response = await retry_operation(
  File "/Users/runner/work/distributed/distributed/distributed/utils_comm.py", line 381, in retry_operation
    return await retry(
  File "/Users/runner/work/distributed/distributed/distributed/utils_comm.py", line 366, in retry
    return await coro()
  File "/Users/runner/work/distributed/distributed/distributed/core.py", line 945, in send_recv_from_rpc
    return await send_recv(comm=comm, op=key, **kwargs)
  File "/Users/runner/work/distributed/distributed/distributed/core.py", line 709, in send_recv
    response = await comm.read(deserializers=deserializers)
  File "/Users/runner/work/distributed/distributed/distributed/comm/tcp.py", line 242, in read
    convert_stream_closed_error(self, e)
  File "/Users/runner/work/distributed/distributed/distributed/comm/tcp.py", line 150, in convert_stream_closed_error
    raise CommClosedError(f"in {obj}: {exc}") from exc
distributed.comm.core.CommClosedError: in <TCP (closed) ConnectionPool.heartbeat_worker local=tcp://127.0.0.1:52917 remote=tcp://127.0.0.1:52907>: Stream is closed
___________________ test_last_in_first_out[queue on worker] ____________________

c = <Client: No scheduler connected>
s = <Scheduler 'tcp://127.0.0.1:59925', workers: 0, cores: 0, tasks: 0>
a = <Worker 'tcp://127.0.0.1:59926', name: 0, status: closed, stored: 0, running: 1/1, ready: 2, comm: 0, waiting: 0>
pause = False

    @gen_blockable_cluster
    async def test_last_in_first_out(c, s, a, pause):
        async with block_worker(c, s, a, pause, 15, 5):
            xs = [c.submit(slowinc, i, delay=0.05, key=f"x{i}") for i in range(5)]
            ys = [c.submit(slowinc, xs[i], delay=0.05, key=f"y{i}") for i in range(5)]
            zs = [c.submit(slowinc, ys[i], delay=0.05, key=f"z{i}") for i in range(5)]
    
        while not any(s.tasks[z.key].state == "memory" for z in zs):
            await asyncio.sleep(0.01)
>       assert not all(s.tasks[x.key].state == "memory" for x in xs)
E       assert not True
E        +  where True = all(<generator object test_last_in_first_out.<locals>.<genexpr> at 0x13887e500>)

distributed/tests/test_priorities.py:233: AssertionError