We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
AllProgress plugin not removed after progress_stream: `TypeError: remove_plugin() takes 0 positional arguments but 2 were given`
**What happened**:
```
distributed/diagnostics/tests/test_progress_stream.py::test_progress_quads_too_many PASSED
distributed.worker - WARNING - Compute Failed
Function:  div
args:      (1, 0)
kwargs:    {}
Exception: "ZeroDivisionError('division by zero')"
distributed.utils - ERROR - remove_plugin() takes 0 positional arguments but 2 were given
Traceback (most recent call last):
  File "/Users/runner/work/distributed/distributed/distributed/utils.py", line 680, in log_errors
    yield
  File "/Users/runner/work/distributed/distributed/distributed/scheduler.py", line 7057, in feed
    teardown(self, state)
TypeError: remove_plugin() takes 0 positional arguments but 2 were given
distributed.core - ERROR - Exception while handling op feed
Traceback (most recent call last):
  File "/Users/runner/work/distributed/distributed/distributed/core.py", line 520, in handle_comm
    result = await result
  File "/Users/runner/work/distributed/distributed/distributed/scheduler.py", line 7057, in feed
    teardown(self, state)
TypeError: remove_plugin() takes 0 positional arguments but 2 were given
tornado.application - ERROR - Exception in callback functools.partial(<function TCPServer._handle_connection.<locals>.<lambda> at 0x13f9a6f70>, <Task finished name='Task-49498' coro=<BaseTCPListener._handle_stream() done, defined at /Users/runner/work/distributed/distributed/distributed/comm/tcp.py:513> exception=TypeError('remove_plugin() takes 0 positional arguments but 2 were given')>)
Traceback (most recent call last):
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.9/site-packages/tornado/ioloop.py", line 741, in _run_callback
    ret = callback()
  File "/Users/runner/miniconda3/envs/dask-distributed/lib/python3.9/site-packages/tornado/tcpserver.py", line 331, in <lambda>
    gen.convert_yielded(future), lambda f: f.result()
  File "/Users/runner/work/distributed/distributed/distributed/comm/tcp.py", line 530, in _handle_stream
    await self.comm_handler(comm)
  File "/Users/runner/work/distributed/distributed/distributed/core.py", line 520, in handle_comm
    result = await result
  File "/Users/runner/work/distributed/distributed/distributed/scheduler.py", line 7057, in feed
    teardown(self, state)
TypeError: remove_plugin() takes 0 positional arguments but 2 were given
```
**What you expected to happen**:
The "all-progress" plugin should be removed when "progress_stream" is torn down.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `distributed/diagnostics/progress_stream.py`
Content:
```
1 import logging
2 from functools import partial
3 
4 from tlz import merge, valmap
5 
6 from ..core import coerce_to_address, connect
7 from ..scheduler import Scheduler
8 from ..utils import color_of, key_split
9 from ..worker import dumps_function
10 from .progress import AllProgress
11 
12 logger = logging.getLogger(__name__)
13 
14 
15 def counts(scheduler, allprogress):
16     return merge(
17         {"all": valmap(len, allprogress.all), "nbytes": allprogress.nbytes},
18         {
19             state: valmap(len, allprogress.state[state])
20             for state in ["memory", "erred", "released", "processing"]
21         },
22     )
23 
24 
25 def remove_plugin(**kwargs):
26     # Wrapper function around `Scheduler.remove_plugin` to avoid raising a
27     # `PicklingError` when using a cythonized scheduler
28     return Scheduler.remove_plugin(**kwargs)
29 
30 
31 async def progress_stream(address, interval):
32     """Open a TCP connection to scheduler, receive progress messages
33 
34     The messages coming back are dicts containing counts of key groups::
35 
36         {'inc': {'all': 5, 'memory': 2, 'erred': 0, 'released': 1},
37          'dec': {'all': 1, 'memory': 0, 'erred': 0, 'released': 0}}
38 
39     Parameters
40     ----------
41     address: address of scheduler
42     interval: time between batches, in seconds
43 
44     Examples
45     --------
46     >>> stream = await eventstream('127.0.0.1:8786', 0.100)  # doctest: +SKIP
47     >>> print(await read(stream))  # doctest: +SKIP
48     """
49     address = coerce_to_address(address)
50     comm = await connect(address)
51     await comm.write(
52         {
53             "op": "feed",
54             "setup": dumps_function(AllProgress),
55             "function": dumps_function(counts),
56             "interval": interval,
57             "teardown": dumps_function(partial(remove_plugin, name=AllProgress.name)),
58         }
59     )
60     return comm
61 
62 
63 def progress_quads(msg, nrows=8, ncols=3):
64     """
65     >>> msg = {'all': {'inc': 5, 'dec': 1, 'add': 4},
66     ...        'memory': {'inc': 2, 'dec': 0, 'add': 1},
67     ...        'erred': {'inc': 0, 'dec': 1, 'add': 0},
68     ...        'released': {'inc': 1, 'dec': 0, 'add': 1},
69     ...        'processing': {'inc': 1, 'dec': 0, 'add': 2}}
70 
71     >>> progress_quads(msg, nrows=2)  # doctest: +SKIP
72     {'name': ['inc', 'add', 'dec'],
73      'left': [0, 0, 1],
74      'right': [0.9, 0.9, 1.9],
75      'top': [0, -1, 0],
76      'bottom': [-.8, -1.8, -.8],
77      'released': [1, 1, 0],
78      'memory': [2, 1, 0],
79      'erred': [0, 0, 1],
80      'processing': [1, 0, 2],
81      'done': ['3 / 5', '2 / 4', '1 / 1'],
82      'released-loc': [.2/.9, .25 / 0.9, 1],
83      'memory-loc': [3 / 5 / .9, .5 / 0.9, 1],
84      'erred-loc': [3 / 5 / .9, .5 / 0.9, 1.9],
85      'processing-loc': [4 / 5, 1 / 1, 1]}}
86     """
87     width = 0.9
88     names = sorted(msg["all"], key=msg["all"].get, reverse=True)
89     names = names[: nrows * ncols]
90     n = len(names)
91     d = {k: [v.get(name, 0) for name in names] for k, v in msg.items()}
92 
93     d["name"] = names
94     d["show-name"] = [name if len(name) <= 15 else name[:12] + "..." for name in names]
95     d["left"] = [i // nrows for i in range(n)]
96     d["right"] = [i // nrows + width for i in range(n)]
97     d["top"] = [-(i % nrows) for i in range(n)]
98     d["bottom"] = [-(i % nrows) - 0.8 for i in range(n)]
99     d["color"] = [color_of(name) for name in names]
100 
101     d["released-loc"] = []
102     d["memory-loc"] = []
103     d["erred-loc"] = []
104     d["processing-loc"] = []
105     d["done"] = []
106     for r, m, e, p, a, l in zip(
107         d["released"], d["memory"], d["erred"], d["processing"], d["all"], d["left"]
108     ):
109         rl = width * r / a + l
110         ml = width * (r + m) / a + l
111         el = width * (r + m + e) / a + l
112         pl = width * (p + r + m + e) / a + l
113         done = "%d / %d" % (r + m + e, a)
114         d["released-loc"].append(rl)
115         d["memory-loc"].append(ml)
116         d["erred-loc"].append(el)
117         d["processing-loc"].append(pl)
118         d["done"].append(done)
119 
120     return d
121 
122 
123 def color_of_message(msg):
124     if msg["status"] == "OK":
125         split = key_split(msg["key"])
126         return color_of(split)
127     else:
128         return "black"
129 
130 
131 colors = {
132     "transfer": "red",
133     "disk-write": "orange",
134     "disk-read": "orange",
135     "deserialize": "gray",
136     "compute": color_of_message,
137 }
138 
139 
140 alphas = {
141     "transfer": 0.4,
142     "compute": 1,
143     "deserialize": 0.4,
144     "disk-write": 0.4,
145     "disk-read": 0.4,
146 }
147 
148 
149 prefix = {
150     "transfer": "transfer-",
151     "disk-write": "disk-write-",
152     "disk-read": "disk-read-",
153     "deserialize": "deserialize-",
154     "compute": "",
155 }
156 
157 
158 def task_stream_append(lists, msg, workers):
159     key = msg["key"]
160     name = key_split(key)
161     startstops = msg.get("startstops", [])
162 
163     for startstop in startstops:
164         color = colors[startstop["action"]]
165         if type(color) is not str:
166             color = color(msg)
167 
168         lists["start"].append((startstop["start"] + startstop["stop"]) / 2 * 1000)
169         lists["duration"].append(1000 * (startstop["stop"] - startstop["start"]))
170         lists["key"].append(key)
171         lists["name"].append(prefix[startstop["action"]] + name)
172         lists["color"].append(color)
173         lists["alpha"].append(alphas[startstop["action"]])
174         lists["worker"].append(msg["worker"])
175 
176         worker_thread = "%s-%d" % (msg["worker"], msg["thread"])
177         lists["worker_thread"].append(worker_thread)
178         if worker_thread not in workers:
179             workers[worker_thread] = len(workers) / 2
180         lists["y"].append(workers[worker_thread])
181 
182     return len(startstops)
183
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/distributed/diagnostics/progress_stream.py b/distributed/diagnostics/progress_stream.py
--- a/distributed/diagnostics/progress_stream.py
+++ b/distributed/diagnostics/progress_stream.py
@@ -1,10 +1,8 @@
 import logging
-from functools import partial
 
 from tlz import merge, valmap
 
 from ..core import coerce_to_address, connect
-from ..scheduler import Scheduler
 from ..utils import color_of, key_split
 from ..worker import dumps_function
 from .progress import AllProgress
@@ -22,10 +20,10 @@
     )
 
 
-def remove_plugin(**kwargs):
+def _remove_all_progress_plugin(self, *args, **kwargs):
     # Wrapper function around `Scheduler.remove_plugin` to avoid raising a
     # `PicklingError` when using a cythonized scheduler
-    return Scheduler.remove_plugin(**kwargs)
+    self.remove_plugin(name=AllProgress.name)
 
 
 async def progress_stream(address, interval):
@@ -54,7 +52,7 @@
             "setup": dumps_function(AllProgress),
             "function": dumps_function(counts),
             "interval": interval,
-            "teardown": dumps_function(partial(remove_plugin, name=AllProgress.name)),
+            "teardown": dumps_function(_remove_all_progress_plugin),
         }
     )
     return comm

{"golden_diff": "diff --git a/distributed/diagnostics/progress_stream.py b/distributed/diagnostics/progress_stream.py\n--- a/distributed/diagnostics/progress_stream.py\n+++ b/distributed/diagnostics/progress_stream.py\n@@ -1,10 +1,8 @@\n import logging\n-from functools import partial\n \n from tlz import merge, valmap\n \n from ..core import coerce_to_address, connect\n-from ..scheduler import Scheduler\n from ..utils import color_of, key_split\n from ..worker import dumps_function\n from .progress import AllProgress\n@@ -22,10 +20,10 @@\n     )\n \n \n-def remove_plugin(**kwargs):\n+def _remove_all_progress_plugin(self, *args, **kwargs):\n     # Wrapper function around `Scheduler.remove_plugin` to avoid raising a\n     # `PicklingError` when using a cythonized scheduler\n-    return Scheduler.remove_plugin(**kwargs)\n+    self.remove_plugin(name=AllProgress.name)\n \n \n async def progress_stream(address, interval):\n@@ -54,7 +52,7 @@\n             \"setup\": dumps_function(AllProgress),\n             \"function\": dumps_function(counts),\n             \"interval\": interval,\n-            \"teardown\": dumps_function(partial(remove_plugin, name=AllProgress.name)),\n+            \"teardown\": dumps_function(_remove_all_progress_plugin),\n         }\n     )\n     return comm\n", "issue": "AllProgress plugin not removed after progress_stream: `TypeError: remove_plugin() takes 0 positional arguments but 2 were given`\n**What happened**:\r\n```\r\ndistributed/diagnostics/tests/test_progress_stream.py::test_progress_quads_too_many PASSED\r\ndistributed.worker - WARNING - Compute Failed\r\nFunction:  div\r\nargs:      (1, 0)\r\nkwargs:    {}\r\nException: \"ZeroDivisionError('division by zero')\"\r\ndistributed.utils - ERROR - remove_plugin() takes 0 positional arguments but 2 were given\r\nTraceback (most recent call last):\r\n  File \"/Users/runner/work/distributed/distributed/distributed/utils.py\", line 680, in log_errors\r\n    yield\r\n  File \"/Users/runner/work/distributed/distributed/distributed/scheduler.py\", line 7057, in feed\r\n    teardown(self, state)\r\nTypeError: remove_plugin() takes 0 positional arguments but 2 were given\r\ndistributed.core - ERROR - Exception while handling op feed\r\nTraceback (most recent call last):\r\n  File \"/Users/runner/work/distributed/distributed/distributed/core.py\", line 520, in handle_comm\r\n    result = await result\r\n  File \"/Users/runner/work/distributed/distributed/distributed/scheduler.py\", line 7057, in feed\r\n    teardown(self, state)\r\nTypeError: remove_plugin() takes 0 positional arguments but 2 were given\r\ntornado.application - ERROR - Exception in callback functools.partial(<function TCPServer._handle_connection.<locals>.<lambda> at 0x13f9a6f70>, <Task finished name='Task-49498' coro=<BaseTCPListener._handle_stream() done, defined at /Users/runner/work/distributed/distributed/distributed/comm/tcp.py:513> exception=TypeError('remove_plugin() takes 0 positional arguments but 2 were given')>)\r\nTraceback (most recent call last):\r\n  File \"/Users/runner/miniconda3/envs/dask-distributed/lib/python3.9/site-packages/tornado/ioloop.py\", line 741, in _run_callback\r\n    ret = callback()\r\n  File \"/Users/runner/miniconda3/envs/dask-distributed/lib/python3.9/site-packages/tornado/tcpserver.py\", line 331, in <lambda>\r\n    gen.convert_yielded(future), lambda f: f.result()\r\n  File \"/Users/runner/work/distributed/distributed/distributed/comm/tcp.py\", line 530, in _handle_stream\r\n    await self.comm_handler(comm)\r\n  File \"/Users/runner/work/distributed/distributed/distributed/core.py\", line 520, in handle_comm\r\n    result = await result\r\n  File \"/Users/runner/work/distributed/distributed/distributed/scheduler.py\", line 7057, in feed\r\n    teardown(self, state)\r\nTypeError: remove_plugin() takes 0 positional arguments but 2 were given\r\n```\r\n**What you expected to happen**:\r\nThe \"all-progress\" plugin should be removed when \"progress_stream\" is torn down.\n", "before_files": [{"content": "import logging\nfrom functools import partial\n\nfrom tlz import merge, valmap\n\nfrom ..core import coerce_to_address, connect\nfrom ..scheduler import Scheduler\nfrom ..utils import color_of, key_split\nfrom ..worker import dumps_function\nfrom .progress import AllProgress\n\nlogger = logging.getLogger(__name__)\n\n\ndef counts(scheduler, allprogress):\n    return merge(\n        {\"all\": valmap(len, allprogress.all), \"nbytes\": allprogress.nbytes},\n        {\n            state: valmap(len, allprogress.state[state])\n            for state in [\"memory\", \"erred\", \"released\", \"processing\"]\n        },\n    )\n\n\ndef remove_plugin(**kwargs):\n    # Wrapper function around `Scheduler.remove_plugin` to avoid raising a\n    # `PicklingError` when using a cythonized scheduler\n    return Scheduler.remove_plugin(**kwargs)\n\n\nasync def progress_stream(address, interval):\n    \"\"\"Open a TCP connection to scheduler, receive progress messages\n\n    The messages coming back are dicts containing counts of key groups::\n\n        {'inc': {'all': 5, 'memory': 2, 'erred': 0, 'released': 1},\n         'dec': {'all': 1, 'memory': 0, 'erred': 0, 'released': 0}}\n\n    Parameters\n    ----------\n    address: address of scheduler\n    interval: time between batches, in seconds\n\n    Examples\n    --------\n    >>> stream = await eventstream('127.0.0.1:8786', 0.100)  # doctest: +SKIP\n    >>> print(await read(stream))  # doctest: +SKIP\n    \"\"\"\n    address = coerce_to_address(address)\n    comm = await connect(address)\n    await comm.write(\n        {\n            \"op\": \"feed\",\n            \"setup\": dumps_function(AllProgress),\n            \"function\": dumps_function(counts),\n            \"interval\": interval,\n            \"teardown\": dumps_function(partial(remove_plugin, name=AllProgress.name)),\n        }\n    )\n    return comm\n\n\ndef progress_quads(msg, nrows=8, ncols=3):\n    \"\"\"\n    >>> msg = {'all': {'inc': 5, 'dec': 1, 'add': 4},\n    ...        'memory': {'inc': 2, 'dec': 0, 'add': 1},\n    ...        'erred': {'inc': 0, 'dec': 1, 'add': 0},\n    ...        'released': {'inc': 1, 'dec': 0, 'add': 1},\n    ...        'processing': {'inc': 1, 'dec': 0, 'add': 2}}\n\n    >>> progress_quads(msg, nrows=2)  # doctest: +SKIP\n    {'name': ['inc', 'add', 'dec'],\n     'left': [0, 0, 1],\n     'right': [0.9, 0.9, 1.9],\n     'top': [0, -1, 0],\n     'bottom': [-.8, -1.8, -.8],\n     'released': [1, 1, 0],\n     'memory': [2, 1, 0],\n     'erred': [0, 0, 1],\n     'processing': [1, 0, 2],\n     'done': ['3 / 5', '2 / 4', '1 / 1'],\n     'released-loc': [.2/.9, .25 / 0.9, 1],\n     'memory-loc': [3 / 5 / .9, .5 / 0.9, 1],\n     'erred-loc': [3 / 5 / .9, .5 / 0.9, 1.9],\n     'processing-loc': [4 / 5, 1 / 1, 1]}}\n    \"\"\"\n    width = 0.9\n    names = sorted(msg[\"all\"], key=msg[\"all\"].get, reverse=True)\n    names = names[: nrows * ncols]\n    n = len(names)\n    d = {k: [v.get(name, 0) for name in names] for k, v in msg.items()}\n\n    d[\"name\"] = names\n    d[\"show-name\"] = [name if len(name) <= 15 else name[:12] + \"...\" for name in names]\n    d[\"left\"] = [i // nrows for i in range(n)]\n    d[\"right\"] = [i // nrows + width for i in range(n)]\n    d[\"top\"] = [-(i % nrows) for i in range(n)]\n    d[\"bottom\"] = [-(i % nrows) - 0.8 for i in range(n)]\n    d[\"color\"] = [color_of(name) for name in names]\n\n    d[\"released-loc\"] = []\n    d[\"memory-loc\"] = []\n    d[\"erred-loc\"] = []\n    d[\"processing-loc\"] = []\n    d[\"done\"] = []\n    for r, m, e, p, a, l in zip(\n        d[\"released\"], d[\"memory\"], d[\"erred\"], d[\"processing\"], d[\"all\"], d[\"left\"]\n    ):\n        rl = width * r / a + l\n        ml = width * (r + m) / a + l\n        el = width * (r + m + e) / a + l\n        pl = width * (p + r + m + e) / a + l\n        done = \"%d / %d\" % (r + m + e, a)\n        d[\"released-loc\"].append(rl)\n        d[\"memory-loc\"].append(ml)\n        d[\"erred-loc\"].append(el)\n        d[\"processing-loc\"].append(pl)\n        d[\"done\"].append(done)\n\n    return d\n\n\ndef color_of_message(msg):\n    if msg[\"status\"] == \"OK\":\n        split = key_split(msg[\"key\"])\n        return color_of(split)\n    else:\n        return \"black\"\n\n\ncolors = {\n    \"transfer\": \"red\",\n    \"disk-write\": \"orange\",\n    \"disk-read\": \"orange\",\n    \"deserialize\": \"gray\",\n    \"compute\": color_of_message,\n}\n\n\nalphas = {\n    \"transfer\": 0.4,\n    \"compute\": 1,\n    \"deserialize\": 0.4,\n    \"disk-write\": 0.4,\n    \"disk-read\": 0.4,\n}\n\n\nprefix = {\n    \"transfer\": \"transfer-\",\n    \"disk-write\": \"disk-write-\",\n    \"disk-read\": \"disk-read-\",\n    \"deserialize\": \"deserialize-\",\n    \"compute\": \"\",\n}\n\n\ndef task_stream_append(lists, msg, workers):\n    key = msg[\"key\"]\n    name = key_split(key)\n    startstops = msg.get(\"startstops\", [])\n\n    for startstop in startstops:\n        color = colors[startstop[\"action\"]]\n        if type(color) is not str:\n            color = color(msg)\n\n        lists[\"start\"].append((startstop[\"start\"] + startstop[\"stop\"]) / 2 * 1000)\n        lists[\"duration\"].append(1000 * (startstop[\"stop\"] - startstop[\"start\"]))\n        lists[\"key\"].append(key)\n        lists[\"name\"].append(prefix[startstop[\"action\"]] + name)\n        lists[\"color\"].append(color)\n        lists[\"alpha\"].append(alphas[startstop[\"action\"]])\n        lists[\"worker\"].append(msg[\"worker\"])\n\n        worker_thread = \"%s-%d\" % (msg[\"worker\"], msg[\"thread\"])\n        lists[\"worker_thread\"].append(worker_thread)\n        if worker_thread not in workers:\n            workers[worker_thread] = len(workers) / 2\n        lists[\"y\"].append(workers[worker_thread])\n\n    return len(startstops)\n", "path": "distributed/diagnostics/progress_stream.py"}], "after_files": [{"content": "import logging\n\nfrom tlz import merge, valmap\n\nfrom ..core import coerce_to_address, connect\nfrom ..utils import color_of, key_split\nfrom ..worker import dumps_function\nfrom .progress import AllProgress\n\nlogger = logging.getLogger(__name__)\n\n\ndef counts(scheduler, allprogress):\n    return merge(\n        {\"all\": valmap(len, allprogress.all), \"nbytes\": allprogress.nbytes},\n        {\n            state: valmap(len, allprogress.state[state])\n            for state in [\"memory\", \"erred\", \"released\", \"processing\"]\n        },\n    )\n\n\ndef _remove_all_progress_plugin(self, *args, **kwargs):\n    # Wrapper function around `Scheduler.remove_plugin` to avoid raising a\n    # `PicklingError` when using a cythonized scheduler\n    self.remove_plugin(name=AllProgress.name)\n\n\nasync def progress_stream(address, interval):\n    \"\"\"Open a TCP connection to scheduler, receive progress messages\n\n    The messages coming back are dicts containing counts of key groups::\n\n        {'inc': {'all': 5, 'memory': 2, 'erred': 0, 'released': 1},\n         'dec': {'all': 1, 'memory': 0, 'erred': 0, 'released': 0}}\n\n    Parameters\n    ----------\n    address: address of scheduler\n    interval: time between batches, in seconds\n\n    Examples\n    --------\n    >>> stream = await eventstream('127.0.0.1:8786', 0.100)  # doctest: +SKIP\n    >>> print(await read(stream))  # doctest: +SKIP\n    \"\"\"\n    address = coerce_to_address(address)\n    comm = await connect(address)\n    await comm.write(\n        {\n            \"op\": \"feed\",\n            \"setup\": dumps_function(AllProgress),\n            \"function\": dumps_function(counts),\n            \"interval\": interval,\n            \"teardown\": dumps_function(_remove_all_progress_plugin),\n        }\n    )\n    return comm\n\n\ndef progress_quads(msg, nrows=8, ncols=3):\n    \"\"\"\n    >>> msg = {'all': {'inc': 5, 'dec': 1, 'add': 4},\n    ...        'memory': {'inc': 2, 'dec': 0, 'add': 1},\n    ...        'erred': {'inc': 0, 'dec': 1, 'add': 0},\n    ...        'released': {'inc': 1, 'dec': 0, 'add': 1},\n    ...        'processing': {'inc': 1, 'dec': 0, 'add': 2}}\n\n    >>> progress_quads(msg, nrows=2)  # doctest: +SKIP\n    {'name': ['inc', 'add', 'dec'],\n     'left': [0, 0, 1],\n     'right': [0.9, 0.9, 1.9],\n     'top': [0, -1, 0],\n     'bottom': [-.8, -1.8, -.8],\n     'released': [1, 1, 0],\n     'memory': [2, 1, 0],\n     'erred': [0, 0, 1],\n     'processing': [1, 0, 2],\n     'done': ['3 / 5', '2 / 4', '1 / 1'],\n     'released-loc': [.2/.9, .25 / 0.9, 1],\n     'memory-loc': [3 / 5 / .9, .5 / 0.9, 1],\n     'erred-loc': [3 / 5 / .9, .5 / 0.9, 1.9],\n     'processing-loc': [4 / 5, 1 / 1, 1]}}\n    \"\"\"\n    width = 0.9\n    names = sorted(msg[\"all\"], key=msg[\"all\"].get, reverse=True)\n    names = names[: nrows * ncols]\n    n = len(names)\n    d = {k: [v.get(name, 0) for name in names] for k, v in msg.items()}\n\n    d[\"name\"] = names\n    d[\"show-name\"] = [name if len(name) <= 15 else name[:12] + \"...\" for name in names]\n    d[\"left\"] = [i // nrows for i in range(n)]\n    d[\"right\"] = [i // nrows + width for i in range(n)]\n    d[\"top\"] = [-(i % nrows) for i in range(n)]\n    d[\"bottom\"] = [-(i % nrows) - 0.8 for i in range(n)]\n    d[\"color\"] = [color_of(name) for name in names]\n\n    d[\"released-loc\"] = []\n    d[\"memory-loc\"] = []\n    d[\"erred-loc\"] = []\n    d[\"processing-loc\"] = []\n    d[\"done\"] = []\n    for r, m, e, p, a, l in zip(\n        d[\"released\"], d[\"memory\"], d[\"erred\"], d[\"processing\"], d[\"all\"], d[\"left\"]\n    ):\n        rl = width * r / a + l\n        ml = width * (r + m) / a + l\n        el = width * (r + m + e) / a + l\n        pl = width * (p + r + m + e) / a + l\n        done = \"%d / %d\" % (r + m + e, a)\n        d[\"released-loc\"].append(rl)\n        d[\"memory-loc\"].append(ml)\n        d[\"erred-loc\"].append(el)\n        d[\"processing-loc\"].append(pl)\n        d[\"done\"].append(done)\n\n    return d\n\n\ndef color_of_message(msg):\n    if msg[\"status\"] == \"OK\":\n        split = key_split(msg[\"key\"])\n        return color_of(split)\n    else:\n        return \"black\"\n\n\ncolors = {\n    \"transfer\": \"red\",\n    \"disk-write\": \"orange\",\n    \"disk-read\": \"orange\",\n    \"deserialize\": \"gray\",\n    \"compute\": color_of_message,\n}\n\n\nalphas = {\n    \"transfer\": 0.4,\n    \"compute\": 1,\n    \"deserialize\": 0.4,\n    \"disk-write\": 0.4,\n    \"disk-read\": 0.4,\n}\n\n\nprefix = {\n    \"transfer\": \"transfer-\",\n    \"disk-write\": \"disk-write-\",\n    \"disk-read\": \"disk-read-\",\n    \"deserialize\": \"deserialize-\",\n    \"compute\": \"\",\n}\n\n\ndef task_stream_append(lists, msg, workers):\n    key = msg[\"key\"]\n    name = key_split(key)\n    startstops = msg.get(\"startstops\", [])\n\n    for startstop in startstops:\n        color = colors[startstop[\"action\"]]\n        if type(color) is not str:\n            color = color(msg)\n\n        lists[\"start\"].append((startstop[\"start\"] + startstop[\"stop\"]) / 2 * 1000)\n        lists[\"duration\"].append(1000 * (startstop[\"stop\"] - startstop[\"start\"]))\n        lists[\"key\"].append(key)\n        lists[\"name\"].append(prefix[startstop[\"action\"]] + name)\n        lists[\"color\"].append(color)\n        lists[\"alpha\"].append(alphas[startstop[\"action\"]])\n        lists[\"worker\"].append(msg[\"worker\"])\n\n        worker_thread = \"%s-%d\" % (msg[\"worker\"], msg[\"thread\"])\n        lists[\"worker_thread\"].append(worker_thread)\n        if worker_thread not in workers:\n            workers[worker_thread] = len(workers) / 2\n        lists[\"y\"].append(workers[worker_thread])\n\n    return len(startstops)\n", "path": "distributed/diagnostics/progress_stream.py"}]}