We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Dynamic parallel objects StopIteration error in Python 3.7
### Bug report

**Bug summary**

Dynamic parallel objects would result in [`StopIteration` error stemming from a Python 3.7 new behavior](https://stackoverflow.com/a/51701040/10892982).

**Code for reproduction**

Given `test.py`:
```python
import yt
yt.enable_parallelism()

for x in yt.parallel_objects(range(10), dynamic=True):
    print(x)
```

Run:
```console
$ mpirun python test.py
```

**Actual outcome**

```console
...
  File "test.py", line 4, in <module>
    for x in yt.parallel_objects(range(10), dynamic=True):
  File "/Users/qobilidop/work/rse/yt-project/yt-dev/yt/utilities/parallel_tools/parallel_analysis_interface.py", line 481, in parallel_objects
    storage=storage):
  File "/Users/qobilidop/work/rse/yt-project/yt-dev/yt/utilities/parallel_tools/task_queue.py", line 174, in dynamic_parallel_objects
    for task in my_q:
P001 yt : [ERROR    ] 2019-05-27 12:40:34,817 RuntimeError: generator raised StopIteration
...
```

**Expected outcome**

No error.

**Version Information**

  * Python Version: 3.7
  * yt version: master branch (3.6.dev0)


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `yt/utilities/parallel_tools/task_queue.py`
Content:
```
1 """
2 Task queue in yt
3 
4 
5 
6 """
7 
8 #-----------------------------------------------------------------------------
9 # Copyright (c) 2013, yt Development Team.
10 #
11 # Distributed under the terms of the Modified BSD License.
12 #
13 # The full license is in the file COPYING.txt, distributed with this software.
14 #-----------------------------------------------------------------------------
15 
16 import numpy as np
17 
18 from yt.funcs import mylog
19 from .parallel_analysis_interface import \
20     communication_system, \
21     _get_comm, \
22     parallel_capable, \
23     ResultsStorage
24 
25 messages = dict(
26     task = dict(msg = 'next'),
27     result = dict(msg = 'result'),
28     task_req = dict(msg = 'task_req'),
29     end = dict(msg = 'no_more_tasks'),
30 )
31 
32 class TaskQueueNonRoot(object):
33     def __init__(self, tasks, comm, subcomm):
34         self.tasks = tasks
35         self.results = {}
36         self.comm = comm
37         self.subcomm = subcomm
38 
39     def send_result(self, result):
40         new_msg = messages['result'].copy()
41         new_msg['value'] = result
42         if self.subcomm.rank == 0:
43             self.comm.comm.send(new_msg, dest = 0, tag=1)
44         self.subcomm.barrier()
45 
46     def get_next(self):
47         msg = messages['task_req'].copy()
48         if self.subcomm.rank == 0:
49             self.comm.comm.send(msg, dest = 0, tag=1)
50             msg = self.comm.comm.recv(source = 0, tag=2)
51         msg = self.subcomm.bcast(msg, root=0)
52         if msg['msg'] == messages['end']['msg']:
53             mylog.debug("Notified to end")
54             raise StopIteration
55         return msg['value']
56 
57     def __iter__(self):
58         while 1:
59             yield self.get_next()
60 
61     def run(self, callable):
62         for task in self:
63             self.send_result(callable(task))
64         return self.finalize()
65 
66     def finalize(self, vals = None):
67         return self.comm.comm.bcast(vals, root = 0)
68 
69 class TaskQueueRoot(TaskQueueNonRoot):
70     def __init__(self, tasks, comm, njobs):
71         self.njobs = njobs
72         self.tasks = tasks
73         self.results = {}
74         self.assignments = {}
75         self._notified = 0
76         self._current = 0
77         self._remaining = len(self.tasks)
78         self.comm = comm
79         # Set up threading here
80         # self.dist = threading.Thread(target=self.handle_assignments)
81         # self.dist.daemon = True
82         # self.dist.start()
83 
84     def run(self, func = None):
85         self.comm.probe_loop(1, self.handle_assignment)
86         return self.finalize(self.results)
87 
88     def insert_result(self, source_id, result):
89         task_id = self.assignments[source_id]
90         self.results[task_id] = result
91 
92     def assign_task(self, source_id):
93         if self._remaining == 0:
94             mylog.debug("Notifying %s to end", source_id)
95             msg = messages['end'].copy()
96             self._notified += 1
97         else:
98             msg = messages['task'].copy()
99             task_id = self._current
100             task = self.tasks[task_id]
101             self.assignments[source_id] = task_id
102             self._current += 1
103             self._remaining -= 1
104             msg['value'] = task
105         self.comm.comm.send(msg, dest = source_id, tag = 2)
106 
107     def handle_assignment(self, status):
108         msg = self.comm.comm.recv(source = status.source, tag = 1)
109         if msg['msg'] == messages['result']['msg']:
110             self.insert_result(status.source, msg['value'])
111         elif msg['msg'] == messages['task_req']['msg']:
112             self.assign_task(status.source)
113         else:
114             mylog.error("GOT AN UNKNOWN MESSAGE: %s", msg)
115             raise RuntimeError
116         if self._notified >= self.njobs:
117             raise StopIteration
118 
119 def task_queue(func, tasks, njobs=0):
120     comm = _get_comm(())
121     if not parallel_capable:
122         mylog.error("Cannot create task queue for serial process.")
123         raise RuntimeError
124     my_size = comm.comm.size
125     if njobs <= 0:
126         njobs = my_size - 1
127     if njobs >= my_size:
128         mylog.error("You have asked for %s jobs, but only %s processors are available.",
129                     njobs, (my_size - 1))
130         raise RuntimeError
131     my_rank = comm.rank
132     all_new_comms = np.array_split(np.arange(1, my_size), njobs)
133     all_new_comms.insert(0, np.array([0]))
134     for i,comm_set in enumerate(all_new_comms):
135         if my_rank in comm_set:
136             my_new_id = i
137             break
138     subcomm = communication_system.push_with_ids(all_new_comms[my_new_id].tolist())
139     
140     if comm.comm.rank == 0:
141         my_q = TaskQueueRoot(tasks, comm, njobs)
142     else:
143         my_q = TaskQueueNonRoot(None, comm, subcomm)
144     communication_system.pop()
145     return my_q.run(func)
146 
147 def dynamic_parallel_objects(tasks, njobs=0, storage=None, broadcast=True):
148     comm = _get_comm(())
149     if not parallel_capable:
150         mylog.error("Cannot create task queue for serial process.")
151         raise RuntimeError
152     my_size = comm.comm.size
153     if njobs <= 0:
154         njobs = my_size - 1
155     if njobs >= my_size:
156         mylog.error("You have asked for %s jobs, but only %s processors are available.",
157                     njobs, (my_size - 1))
158         raise RuntimeError
159     my_rank = comm.rank
160     all_new_comms = np.array_split(np.arange(1, my_size), njobs)
161     all_new_comms.insert(0, np.array([0]))
162     for i,comm_set in enumerate(all_new_comms):
163         if my_rank in comm_set:
164             my_new_id = i
165             break
166     subcomm = communication_system.push_with_ids(all_new_comms[my_new_id].tolist())
167     
168     if comm.comm.rank == 0:
169         my_q = TaskQueueRoot(tasks, comm, njobs)
170         my_q.comm.probe_loop(1, my_q.handle_assignment)
171     else:
172         my_q = TaskQueueNonRoot(None, comm, subcomm)
173         if storage is None:
174             for task in my_q:
175                 yield task
176         else:
177             for task in my_q:
178                 rstore = ResultsStorage()
179                 yield rstore, task
180                 my_q.send_result(rstore.result)
181 
182     if storage is not None:
183         if broadcast:
184             my_results = my_q.comm.comm.bcast(my_q.results, root=0)
185         else:
186             my_results = my_q.results
187         storage.update(my_results)
188 
189     communication_system.pop()
190
```
--- 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/yt/utilities/parallel_tools/task_queue.py b/yt/utilities/parallel_tools/task_queue.py
--- a/yt/utilities/parallel_tools/task_queue.py
+++ b/yt/utilities/parallel_tools/task_queue.py
@@ -43,7 +43,7 @@
             self.comm.comm.send(new_msg, dest = 0, tag=1)
         self.subcomm.barrier()
 
-    def get_next(self):
+    def __next__(self):
         msg = messages['task_req'].copy()
         if self.subcomm.rank == 0:
             self.comm.comm.send(msg, dest = 0, tag=1)
@@ -54,9 +54,11 @@
             raise StopIteration
         return msg['value']
 
+    # For Python 2 compatibility
+    next = __next__
+
     def __iter__(self):
-        while 1:
-            yield self.get_next()
+        return self
 
     def run(self, callable):
         for task in self:

{"golden_diff": "diff --git a/yt/utilities/parallel_tools/task_queue.py b/yt/utilities/parallel_tools/task_queue.py\n--- a/yt/utilities/parallel_tools/task_queue.py\n+++ b/yt/utilities/parallel_tools/task_queue.py\n@@ -43,7 +43,7 @@\n             self.comm.comm.send(new_msg, dest = 0, tag=1)\n         self.subcomm.barrier()\n \n-    def get_next(self):\n+    def __next__(self):\n         msg = messages['task_req'].copy()\n         if self.subcomm.rank == 0:\n             self.comm.comm.send(msg, dest = 0, tag=1)\n@@ -54,9 +54,11 @@\n             raise StopIteration\n         return msg['value']\n \n+    # For Python 2 compatibility\n+    next = __next__\n+\n     def __iter__(self):\n-        while 1:\n-            yield self.get_next()\n+        return self\n \n     def run(self, callable):\n         for task in self:\n", "issue": "Dynamic parallel objects StopIteration error in Python 3.7\n### Bug report\r\n\r\n**Bug summary**\r\n\r\nDynamic parallel objects would result in [`StopIteration` error stemming from a Python 3.7 new behavior](https://stackoverflow.com/a/51701040/10892982).\r\n\r\n**Code for reproduction**\r\n\r\nGiven `test.py`:\r\n```python\r\nimport yt\r\nyt.enable_parallelism()\r\n\r\nfor x in yt.parallel_objects(range(10), dynamic=True):\r\n    print(x)\r\n```\r\n\r\nRun:\r\n```console\r\n$ mpirun python test.py\r\n```\r\n\r\n**Actual outcome**\r\n\r\n```console\r\n...\r\n  File \"test.py\", line 4, in <module>\r\n    for x in yt.parallel_objects(range(10), dynamic=True):\r\n  File \"/Users/qobilidop/work/rse/yt-project/yt-dev/yt/utilities/parallel_tools/parallel_analysis_interface.py\", line 481, in parallel_objects\r\n    storage=storage):\r\n  File \"/Users/qobilidop/work/rse/yt-project/yt-dev/yt/utilities/parallel_tools/task_queue.py\", line 174, in dynamic_parallel_objects\r\n    for task in my_q:\r\nP001 yt : [ERROR    ] 2019-05-27 12:40:34,817 RuntimeError: generator raised StopIteration\r\n...\r\n```\r\n\r\n**Expected outcome**\r\n\r\nNo error.\r\n\r\n**Version Information**\r\n\r\n  * Python Version: 3.7\r\n  * yt version: master branch (3.6.dev0)\r\n\n", "before_files": [{"content": "\"\"\"\nTask queue in yt\n\n\n\n\"\"\"\n\n#-----------------------------------------------------------------------------\n# Copyright (c) 2013, yt Development Team.\n#\n# Distributed under the terms of the Modified BSD License.\n#\n# The full license is in the file COPYING.txt, distributed with this software.\n#-----------------------------------------------------------------------------\n\nimport numpy as np\n\nfrom yt.funcs import mylog\nfrom .parallel_analysis_interface import \\\n    communication_system, \\\n    _get_comm, \\\n    parallel_capable, \\\n    ResultsStorage\n\nmessages = dict(\n    task = dict(msg = 'next'),\n    result = dict(msg = 'result'),\n    task_req = dict(msg = 'task_req'),\n    end = dict(msg = 'no_more_tasks'),\n)\n\nclass TaskQueueNonRoot(object):\n    def __init__(self, tasks, comm, subcomm):\n        self.tasks = tasks\n        self.results = {}\n        self.comm = comm\n        self.subcomm = subcomm\n\n    def send_result(self, result):\n        new_msg = messages['result'].copy()\n        new_msg['value'] = result\n        if self.subcomm.rank == 0:\n            self.comm.comm.send(new_msg, dest = 0, tag=1)\n        self.subcomm.barrier()\n\n    def get_next(self):\n        msg = messages['task_req'].copy()\n        if self.subcomm.rank == 0:\n            self.comm.comm.send(msg, dest = 0, tag=1)\n            msg = self.comm.comm.recv(source = 0, tag=2)\n        msg = self.subcomm.bcast(msg, root=0)\n        if msg['msg'] == messages['end']['msg']:\n            mylog.debug(\"Notified to end\")\n            raise StopIteration\n        return msg['value']\n\n    def __iter__(self):\n        while 1:\n            yield self.get_next()\n\n    def run(self, callable):\n        for task in self:\n            self.send_result(callable(task))\n        return self.finalize()\n\n    def finalize(self, vals = None):\n        return self.comm.comm.bcast(vals, root = 0)\n\nclass TaskQueueRoot(TaskQueueNonRoot):\n    def __init__(self, tasks, comm, njobs):\n        self.njobs = njobs\n        self.tasks = tasks\n        self.results = {}\n        self.assignments = {}\n        self._notified = 0\n        self._current = 0\n        self._remaining = len(self.tasks)\n        self.comm = comm\n        # Set up threading here\n        # self.dist = threading.Thread(target=self.handle_assignments)\n        # self.dist.daemon = True\n        # self.dist.start()\n\n    def run(self, func = None):\n        self.comm.probe_loop(1, self.handle_assignment)\n        return self.finalize(self.results)\n\n    def insert_result(self, source_id, result):\n        task_id = self.assignments[source_id]\n        self.results[task_id] = result\n\n    def assign_task(self, source_id):\n        if self._remaining == 0:\n            mylog.debug(\"Notifying %s to end\", source_id)\n            msg = messages['end'].copy()\n            self._notified += 1\n        else:\n            msg = messages['task'].copy()\n            task_id = self._current\n            task = self.tasks[task_id]\n            self.assignments[source_id] = task_id\n            self._current += 1\n            self._remaining -= 1\n            msg['value'] = task\n        self.comm.comm.send(msg, dest = source_id, tag = 2)\n\n    def handle_assignment(self, status):\n        msg = self.comm.comm.recv(source = status.source, tag = 1)\n        if msg['msg'] == messages['result']['msg']:\n            self.insert_result(status.source, msg['value'])\n        elif msg['msg'] == messages['task_req']['msg']:\n            self.assign_task(status.source)\n        else:\n            mylog.error(\"GOT AN UNKNOWN MESSAGE: %s\", msg)\n            raise RuntimeError\n        if self._notified >= self.njobs:\n            raise StopIteration\n\ndef task_queue(func, tasks, njobs=0):\n    comm = _get_comm(())\n    if not parallel_capable:\n        mylog.error(\"Cannot create task queue for serial process.\")\n        raise RuntimeError\n    my_size = comm.comm.size\n    if njobs <= 0:\n        njobs = my_size - 1\n    if njobs >= my_size:\n        mylog.error(\"You have asked for %s jobs, but only %s processors are available.\",\n                    njobs, (my_size - 1))\n        raise RuntimeError\n    my_rank = comm.rank\n    all_new_comms = np.array_split(np.arange(1, my_size), njobs)\n    all_new_comms.insert(0, np.array([0]))\n    for i,comm_set in enumerate(all_new_comms):\n        if my_rank in comm_set:\n            my_new_id = i\n            break\n    subcomm = communication_system.push_with_ids(all_new_comms[my_new_id].tolist())\n    \n    if comm.comm.rank == 0:\n        my_q = TaskQueueRoot(tasks, comm, njobs)\n    else:\n        my_q = TaskQueueNonRoot(None, comm, subcomm)\n    communication_system.pop()\n    return my_q.run(func)\n\ndef dynamic_parallel_objects(tasks, njobs=0, storage=None, broadcast=True):\n    comm = _get_comm(())\n    if not parallel_capable:\n        mylog.error(\"Cannot create task queue for serial process.\")\n        raise RuntimeError\n    my_size = comm.comm.size\n    if njobs <= 0:\n        njobs = my_size - 1\n    if njobs >= my_size:\n        mylog.error(\"You have asked for %s jobs, but only %s processors are available.\",\n                    njobs, (my_size - 1))\n        raise RuntimeError\n    my_rank = comm.rank\n    all_new_comms = np.array_split(np.arange(1, my_size), njobs)\n    all_new_comms.insert(0, np.array([0]))\n    for i,comm_set in enumerate(all_new_comms):\n        if my_rank in comm_set:\n            my_new_id = i\n            break\n    subcomm = communication_system.push_with_ids(all_new_comms[my_new_id].tolist())\n    \n    if comm.comm.rank == 0:\n        my_q = TaskQueueRoot(tasks, comm, njobs)\n        my_q.comm.probe_loop(1, my_q.handle_assignment)\n    else:\n        my_q = TaskQueueNonRoot(None, comm, subcomm)\n        if storage is None:\n            for task in my_q:\n                yield task\n        else:\n            for task in my_q:\n                rstore = ResultsStorage()\n                yield rstore, task\n                my_q.send_result(rstore.result)\n\n    if storage is not None:\n        if broadcast:\n            my_results = my_q.comm.comm.bcast(my_q.results, root=0)\n        else:\n            my_results = my_q.results\n        storage.update(my_results)\n\n    communication_system.pop()\n", "path": "yt/utilities/parallel_tools/task_queue.py"}], "after_files": [{"content": "\"\"\"\nTask queue in yt\n\n\n\n\"\"\"\n\n#-----------------------------------------------------------------------------\n# Copyright (c) 2013, yt Development Team.\n#\n# Distributed under the terms of the Modified BSD License.\n#\n# The full license is in the file COPYING.txt, distributed with this software.\n#-----------------------------------------------------------------------------\n\nimport numpy as np\n\nfrom yt.funcs import mylog\nfrom .parallel_analysis_interface import \\\n    communication_system, \\\n    _get_comm, \\\n    parallel_capable, \\\n    ResultsStorage\n\nmessages = dict(\n    task = dict(msg = 'next'),\n    result = dict(msg = 'result'),\n    task_req = dict(msg = 'task_req'),\n    end = dict(msg = 'no_more_tasks'),\n)\n\nclass TaskQueueNonRoot(object):\n    def __init__(self, tasks, comm, subcomm):\n        self.tasks = tasks\n        self.results = {}\n        self.comm = comm\n        self.subcomm = subcomm\n\n    def send_result(self, result):\n        new_msg = messages['result'].copy()\n        new_msg['value'] = result\n        if self.subcomm.rank == 0:\n            self.comm.comm.send(new_msg, dest = 0, tag=1)\n        self.subcomm.barrier()\n\n    def __next__(self):\n        msg = messages['task_req'].copy()\n        if self.subcomm.rank == 0:\n            self.comm.comm.send(msg, dest = 0, tag=1)\n            msg = self.comm.comm.recv(source = 0, tag=2)\n        msg = self.subcomm.bcast(msg, root=0)\n        if msg['msg'] == messages['end']['msg']:\n            mylog.debug(\"Notified to end\")\n            raise StopIteration\n        return msg['value']\n\n    # For Python 2 compatibility\n    next = __next__\n\n    def __iter__(self):\n        return self\n\n    def run(self, callable):\n        for task in self:\n            self.send_result(callable(task))\n        return self.finalize()\n\n    def finalize(self, vals = None):\n        return self.comm.comm.bcast(vals, root = 0)\n\nclass TaskQueueRoot(TaskQueueNonRoot):\n    def __init__(self, tasks, comm, njobs):\n        self.njobs = njobs\n        self.tasks = tasks\n        self.results = {}\n        self.assignments = {}\n        self._notified = 0\n        self._current = 0\n        self._remaining = len(self.tasks)\n        self.comm = comm\n        # Set up threading here\n        # self.dist = threading.Thread(target=self.handle_assignments)\n        # self.dist.daemon = True\n        # self.dist.start()\n\n    def run(self, func = None):\n        self.comm.probe_loop(1, self.handle_assignment)\n        return self.finalize(self.results)\n\n    def insert_result(self, source_id, result):\n        task_id = self.assignments[source_id]\n        self.results[task_id] = result\n\n    def assign_task(self, source_id):\n        if self._remaining == 0:\n            mylog.debug(\"Notifying %s to end\", source_id)\n            msg = messages['end'].copy()\n            self._notified += 1\n        else:\n            msg = messages['task'].copy()\n            task_id = self._current\n            task = self.tasks[task_id]\n            self.assignments[source_id] = task_id\n            self._current += 1\n            self._remaining -= 1\n            msg['value'] = task\n        self.comm.comm.send(msg, dest = source_id, tag = 2)\n\n    def handle_assignment(self, status):\n        msg = self.comm.comm.recv(source = status.source, tag = 1)\n        if msg['msg'] == messages['result']['msg']:\n            self.insert_result(status.source, msg['value'])\n        elif msg['msg'] == messages['task_req']['msg']:\n            self.assign_task(status.source)\n        else:\n            mylog.error(\"GOT AN UNKNOWN MESSAGE: %s\", msg)\n            raise RuntimeError\n        if self._notified >= self.njobs:\n            raise StopIteration\n\ndef task_queue(func, tasks, njobs=0):\n    comm = _get_comm(())\n    if not parallel_capable:\n        mylog.error(\"Cannot create task queue for serial process.\")\n        raise RuntimeError\n    my_size = comm.comm.size\n    if njobs <= 0:\n        njobs = my_size - 1\n    if njobs >= my_size:\n        mylog.error(\"You have asked for %s jobs, but only %s processors are available.\",\n                    njobs, (my_size - 1))\n        raise RuntimeError\n    my_rank = comm.rank\n    all_new_comms = np.array_split(np.arange(1, my_size), njobs)\n    all_new_comms.insert(0, np.array([0]))\n    for i,comm_set in enumerate(all_new_comms):\n        if my_rank in comm_set:\n            my_new_id = i\n            break\n    subcomm = communication_system.push_with_ids(all_new_comms[my_new_id].tolist())\n    \n    if comm.comm.rank == 0:\n        my_q = TaskQueueRoot(tasks, comm, njobs)\n    else:\n        my_q = TaskQueueNonRoot(None, comm, subcomm)\n    communication_system.pop()\n    return my_q.run(func)\n\ndef dynamic_parallel_objects(tasks, njobs=0, storage=None, broadcast=True):\n    comm = _get_comm(())\n    if not parallel_capable:\n        mylog.error(\"Cannot create task queue for serial process.\")\n        raise RuntimeError\n    my_size = comm.comm.size\n    if njobs <= 0:\n        njobs = my_size - 1\n    if njobs >= my_size:\n        mylog.error(\"You have asked for %s jobs, but only %s processors are available.\",\n                    njobs, (my_size - 1))\n        raise RuntimeError\n    my_rank = comm.rank\n    all_new_comms = np.array_split(np.arange(1, my_size), njobs)\n    all_new_comms.insert(0, np.array([0]))\n    for i,comm_set in enumerate(all_new_comms):\n        if my_rank in comm_set:\n            my_new_id = i\n            break\n    subcomm = communication_system.push_with_ids(all_new_comms[my_new_id].tolist())\n    \n    if comm.comm.rank == 0:\n        my_q = TaskQueueRoot(tasks, comm, njobs)\n        my_q.comm.probe_loop(1, my_q.handle_assignment)\n    else:\n        my_q = TaskQueueNonRoot(None, comm, subcomm)\n        if storage is None:\n            for task in my_q:\n                yield task\n        else:\n            for task in my_q:\n                rstore = ResultsStorage()\n                yield rstore, task\n                my_q.send_result(rstore.result)\n\n    if storage is not None:\n        if broadcast:\n            my_results = my_q.comm.comm.bcast(my_q.results, root=0)\n        else:\n            my_results = my_q.results\n        storage.update(my_results)\n\n    communication_system.pop()\n", "path": "yt/utilities/parallel_tools/task_queue.py"}]}