We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
worker pod crashes when get task
```
2020-03-12 22:03:32.300671: W tensorflow/core/kernels/data/generator_dataset_op.cc:103] Error occurred when finalizing GeneratorDataset iterator: Cancelled: Operation was cancelled
Traceback (most recent call last):
  File "/usr/lib/python3.6/runpy.py", line 193, in _run_module_as_main
    "__main__", mod_spec)
  File "/usr/lib/python3.6/runpy.py", line 85, in _run_code
    exec(code, run_globals)
  File "/elasticdl/python/worker/main.py", line 44, in <module>
    main()
  File "/elasticdl/python/worker/main.py", line 40, in main
    worker.run()
  File "/elasticdl/python/worker/worker.py", line 1144, in run
    self._train_and_evaluate()
  File "/elasticdl/python/worker/worker.py", line 1074, in _train_and_evaluate
    self._minibatch_size, err_msg
  File "/elasticdl/python/worker/task_data_service.py", line 86, in report_record_done
    task = self._pending_tasks[0]
IndexError: deque index out of range
```

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `elasticdl/python/worker/task_data_service.py`
Content:
```
1 import threading
2 import time
3 from collections import deque
4 
5 import tensorflow as tf
6 
7 from elasticdl.proto import elasticdl_pb2
8 from elasticdl.python.common.constants import TaskExecCounterKey
9 from elasticdl.python.common.log_utils import default_logger as logger
10 from elasticdl.python.data.reader.data_reader_factory import create_data_reader
11 
12 
13 class TaskDataService(object):
14     def __init__(
15         self, worker, training_with_evaluation, data_reader_params=None
16     ):
17         self._worker = worker
18         self._create_data_reader_fn = create_data_reader
19         if self._worker._custom_data_reader is not None:
20             self._create_data_reader_fn = self._worker._custom_data_reader
21         self._training_with_evaluation = training_with_evaluation
22         self._lock = threading.Lock()
23         self._pending_dataset = True
24         self._pending_train_end_callback_task = None
25         if data_reader_params:
26             self.data_reader = self._create_data_reader_fn(
27                 data_origin=None, **data_reader_params
28             )
29         else:
30             self.data_reader = self._create_data_reader_fn(data_origin=None)
31         self._warm_up_task = None
32         self._has_warmed_up = False
33         self._failed_record_count = 0
34         self._reported_record_count = 0
35         self._current_task = None
36         self._pending_tasks = deque()
37 
38     def _reset(self):
39         """
40         Reset pending tasks and record counts
41         """
42         self._reported_record_count = 0
43         self._failed_record_count = 0
44         self._pending_tasks = deque()
45         self._current_task = None
46 
47     def get_current_task(self):
48         return self._current_task
49 
50     def _do_report_task(self, task, err_msg=""):
51         if self._failed_record_count != 0:
52             exec_counters = {
53                 TaskExecCounterKey.FAIL_COUNT: self._failed_record_count
54             }
55         else:
56             exec_counters = None
57         self._worker.report_task_result(
58             task.task_id, err_msg, exec_counters=exec_counters
59         )
60 
61     def _log_fail_records(self, task, err_msg):
62         task_len = task.end - task.start
63         msg = (
64             "records ({f}/{t}) failure, possible "
65             "in task_id: {task_id} "
66             'reason "{err_msg}"'
67         ).format(
68             task_id=task.task_id,
69             err_msg=err_msg,
70             f=self._failed_record_count,
71             t=task_len,
72         )
73         logger.warning(msg)
74 
75     def report_record_done(self, count, err_msg=""):
76         """
77         Report the number of records in the latest processed batch,
78         so TaskDataService knows if some pending tasks are finished
79         and report_task_result to the master.
80         Return True if there are some finished tasks, False otherwise.
81         """
82         self._reported_record_count += count
83         if err_msg:
84             self._failed_record_count += count
85 
86         task = self._pending_tasks[0]
87         total_record_num = task.end - task.start
88         if self._reported_record_count >= total_record_num:
89             if err_msg:
90                 self._log_fail_records(task, err_msg)
91 
92             # Keep popping tasks until the reported record count is less
93             # than the size of the current data since `batch_size` may be
94             # larger than `task.end - task.start`
95             with self._lock:
96                 while self._pending_tasks and self._reported_record_count >= (
97                     self._pending_tasks[0].end - self._pending_tasks[0].start
98                 ):
99                     task = self._pending_tasks[0]
100                     self._reported_record_count -= task.end - task.start
101                     self._pending_tasks.popleft()
102                     self._do_report_task(task, err_msg)
103                     self._failed_record_count = 0
104                 if self._pending_tasks:
105                     self._current_task = self._pending_tasks[0]
106             return True
107         return False
108 
109     def get_dataset_gen(self, task):
110         """
111         If a task exists, this creates a generator, which could be used to
112         creating a `tf.data.Dataset` object in further.
113         """
114         if not task:
115             return None
116         tasks = [task]
117 
118         def gen():
119             for task in tasks:
120                 for data in self.data_reader.read_records(task):
121                     if data:
122                         yield data
123 
124         return gen
125 
126     def get_dataset_by_task(self, task):
127         if task is None:
128             return None
129         gen = self.get_dataset_gen(task)
130         dataset = tf.data.Dataset.from_generator(
131             gen, self.data_reader.records_output_types
132         )
133         return dataset
134 
135     def get_train_end_callback_task(self):
136         return self._pending_train_end_callback_task
137 
138     def clear_train_end_callback_task(self):
139         self._pending_train_end_callback_task = None
140 
141     def get_dataset(self):
142         """
143         If there's more data, this creates a `tf.data.Dataset` object.
144         Otherwise, this returns `None`.
145         """
146         if self._pending_dataset:
147             if self._pending_tasks:
148                 logger.error(
149                     "Cannot get new dataset when there are pending tasks"
150                 )
151                 return None
152             self._reset()
153             # We use a task to perform warm-up for data reader in order
154             # to collect useful metadata. Note that we only performs
155             # data fetching for this task and `break` instantly to make
156             # sure `read_records()` is executed without iterating all the
157             # records so this should not be time consuming.
158             if self._warm_up_task is None and not self._has_warmed_up:
159                 while True:
160                     task = self._worker.get_task()
161                     if task.type != elasticdl_pb2.WAIT:
162                         break
163                     time.sleep(2)
164                 if task.type == elasticdl_pb2.TRAIN_END_CALLBACK:
165                     self._pending_train_end_callback_task = task
166                     return None
167                 elif not task.shard_name:
168                     logger.info("No more task, stopping")
169                     return None
170                 else:
171                     self._warm_up_task = task
172                     for _ in self.data_reader.read_records(task):
173                         break
174                     self._has_warmed_up = True
175             ds = tf.data.Dataset.from_generator(
176                 self._gen, self.data_reader.records_output_types
177             )
178             self._pending_dataset = False
179             return ds
180         else:
181             return None
182 
183     def _gen(self):
184         """
185         A generator supports the iter() protocol (e.g. a generator function),
186         used to create a `tf.data.Dataset` object from a list of tasks.
187         """
188         while True:
189             # Make sure we also generate data from the warm-up task.
190             if self._warm_up_task is not None and self._has_warmed_up:
191                 task = self._warm_up_task
192                 self._warm_up_task = None
193             else:
194                 task = self._worker.get_task()
195             if not task.shard_name:
196                 if task.type == elasticdl_pb2.WAIT:
197                     self._pending_dataset = True
198                     logger.info("No tasks for now, maybe more later")
199                 else:
200                     logger.info("No more task, stopping")
201                 break
202             with self._lock:
203                 if task.type == elasticdl_pb2.TRAIN_END_CALLBACK:
204                     self._pending_train_end_callback_task = task
205                     continue
206 
207                 self._pending_tasks.append(task)
208                 if len(self._pending_tasks) == 1:
209                     self._current_task = task
210             for data in self.data_reader.read_records(task):
211                 if data:
212                     yield data
213
```
--- 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/elasticdl/python/worker/task_data_service.py b/elasticdl/python/worker/task_data_service.py
--- a/elasticdl/python/worker/task_data_service.py
+++ b/elasticdl/python/worker/task_data_service.py
@@ -83,6 +83,9 @@
         if err_msg:
             self._failed_record_count += count
 
+        # TODO(qijun) This is a workaround for #1829
+        if not self._pending_tasks:
+            return False
         task = self._pending_tasks[0]
         total_record_num = task.end - task.start
         if self._reported_record_count >= total_record_num:

{"golden_diff": "diff --git a/elasticdl/python/worker/task_data_service.py b/elasticdl/python/worker/task_data_service.py\n--- a/elasticdl/python/worker/task_data_service.py\n+++ b/elasticdl/python/worker/task_data_service.py\n@@ -83,6 +83,9 @@\n         if err_msg:\n             self._failed_record_count += count\n \n+        # TODO(qijun) This is a workaround for #1829\n+        if not self._pending_tasks:\n+            return False\n         task = self._pending_tasks[0]\n         total_record_num = task.end - task.start\n         if self._reported_record_count >= total_record_num:\n", "issue": "worker pod crashes when get task\n```\r\n2020-03-12 22:03:32.300671: W tensorflow/core/kernels/data/generator_dataset_op.cc:103] Error occurred when finalizing GeneratorDataset iterator: Cancelled: Operation was cancelled\r\nTraceback (most recent call last):\r\n  File \"/usr/lib/python3.6/runpy.py\", line 193, in _run_module_as_main\r\n    \"__main__\", mod_spec)\r\n  File \"/usr/lib/python3.6/runpy.py\", line 85, in _run_code\r\n    exec(code, run_globals)\r\n  File \"/elasticdl/python/worker/main.py\", line 44, in <module>\r\n    main()\r\n  File \"/elasticdl/python/worker/main.py\", line 40, in main\r\n    worker.run()\r\n  File \"/elasticdl/python/worker/worker.py\", line 1144, in run\r\n    self._train_and_evaluate()\r\n  File \"/elasticdl/python/worker/worker.py\", line 1074, in _train_and_evaluate\r\n    self._minibatch_size, err_msg\r\n  File \"/elasticdl/python/worker/task_data_service.py\", line 86, in report_record_done\r\n    task = self._pending_tasks[0]\r\nIndexError: deque index out of range\r\n```\n", "before_files": [{"content": "import threading\nimport time\nfrom collections import deque\n\nimport tensorflow as tf\n\nfrom elasticdl.proto import elasticdl_pb2\nfrom elasticdl.python.common.constants import TaskExecCounterKey\nfrom elasticdl.python.common.log_utils import default_logger as logger\nfrom elasticdl.python.data.reader.data_reader_factory import create_data_reader\n\n\nclass TaskDataService(object):\n    def __init__(\n        self, worker, training_with_evaluation, data_reader_params=None\n    ):\n        self._worker = worker\n        self._create_data_reader_fn = create_data_reader\n        if self._worker._custom_data_reader is not None:\n            self._create_data_reader_fn = self._worker._custom_data_reader\n        self._training_with_evaluation = training_with_evaluation\n        self._lock = threading.Lock()\n        self._pending_dataset = True\n        self._pending_train_end_callback_task = None\n        if data_reader_params:\n            self.data_reader = self._create_data_reader_fn(\n                data_origin=None, **data_reader_params\n            )\n        else:\n            self.data_reader = self._create_data_reader_fn(data_origin=None)\n        self._warm_up_task = None\n        self._has_warmed_up = False\n        self._failed_record_count = 0\n        self._reported_record_count = 0\n        self._current_task = None\n        self._pending_tasks = deque()\n\n    def _reset(self):\n        \"\"\"\n        Reset pending tasks and record counts\n        \"\"\"\n        self._reported_record_count = 0\n        self._failed_record_count = 0\n        self._pending_tasks = deque()\n        self._current_task = None\n\n    def get_current_task(self):\n        return self._current_task\n\n    def _do_report_task(self, task, err_msg=\"\"):\n        if self._failed_record_count != 0:\n            exec_counters = {\n                TaskExecCounterKey.FAIL_COUNT: self._failed_record_count\n            }\n        else:\n            exec_counters = None\n        self._worker.report_task_result(\n            task.task_id, err_msg, exec_counters=exec_counters\n        )\n\n    def _log_fail_records(self, task, err_msg):\n        task_len = task.end - task.start\n        msg = (\n            \"records ({f}/{t}) failure, possible \"\n            \"in task_id: {task_id} \"\n            'reason \"{err_msg}\"'\n        ).format(\n            task_id=task.task_id,\n            err_msg=err_msg,\n            f=self._failed_record_count,\n            t=task_len,\n        )\n        logger.warning(msg)\n\n    def report_record_done(self, count, err_msg=\"\"):\n        \"\"\"\n        Report the number of records in the latest processed batch,\n        so TaskDataService knows if some pending tasks are finished\n        and report_task_result to the master.\n        Return True if there are some finished tasks, False otherwise.\n        \"\"\"\n        self._reported_record_count += count\n        if err_msg:\n            self._failed_record_count += count\n\n        task = self._pending_tasks[0]\n        total_record_num = task.end - task.start\n        if self._reported_record_count >= total_record_num:\n            if err_msg:\n                self._log_fail_records(task, err_msg)\n\n            # Keep popping tasks until the reported record count is less\n            # than the size of the current data since `batch_size` may be\n            # larger than `task.end - task.start`\n            with self._lock:\n                while self._pending_tasks and self._reported_record_count >= (\n                    self._pending_tasks[0].end - self._pending_tasks[0].start\n                ):\n                    task = self._pending_tasks[0]\n                    self._reported_record_count -= task.end - task.start\n                    self._pending_tasks.popleft()\n                    self._do_report_task(task, err_msg)\n                    self._failed_record_count = 0\n                if self._pending_tasks:\n                    self._current_task = self._pending_tasks[0]\n            return True\n        return False\n\n    def get_dataset_gen(self, task):\n        \"\"\"\n        If a task exists, this creates a generator, which could be used to\n        creating a `tf.data.Dataset` object in further.\n        \"\"\"\n        if not task:\n            return None\n        tasks = [task]\n\n        def gen():\n            for task in tasks:\n                for data in self.data_reader.read_records(task):\n                    if data:\n                        yield data\n\n        return gen\n\n    def get_dataset_by_task(self, task):\n        if task is None:\n            return None\n        gen = self.get_dataset_gen(task)\n        dataset = tf.data.Dataset.from_generator(\n            gen, self.data_reader.records_output_types\n        )\n        return dataset\n\n    def get_train_end_callback_task(self):\n        return self._pending_train_end_callback_task\n\n    def clear_train_end_callback_task(self):\n        self._pending_train_end_callback_task = None\n\n    def get_dataset(self):\n        \"\"\"\n        If there's more data, this creates a `tf.data.Dataset` object.\n        Otherwise, this returns `None`.\n        \"\"\"\n        if self._pending_dataset:\n            if self._pending_tasks:\n                logger.error(\n                    \"Cannot get new dataset when there are pending tasks\"\n                )\n                return None\n            self._reset()\n            # We use a task to perform warm-up for data reader in order\n            # to collect useful metadata. Note that we only performs\n            # data fetching for this task and `break` instantly to make\n            # sure `read_records()` is executed without iterating all the\n            # records so this should not be time consuming.\n            if self._warm_up_task is None and not self._has_warmed_up:\n                while True:\n                    task = self._worker.get_task()\n                    if task.type != elasticdl_pb2.WAIT:\n                        break\n                    time.sleep(2)\n                if task.type == elasticdl_pb2.TRAIN_END_CALLBACK:\n                    self._pending_train_end_callback_task = task\n                    return None\n                elif not task.shard_name:\n                    logger.info(\"No more task, stopping\")\n                    return None\n                else:\n                    self._warm_up_task = task\n                    for _ in self.data_reader.read_records(task):\n                        break\n                    self._has_warmed_up = True\n            ds = tf.data.Dataset.from_generator(\n                self._gen, self.data_reader.records_output_types\n            )\n            self._pending_dataset = False\n            return ds\n        else:\n            return None\n\n    def _gen(self):\n        \"\"\"\n        A generator supports the iter() protocol (e.g. a generator function),\n        used to create a `tf.data.Dataset` object from a list of tasks.\n        \"\"\"\n        while True:\n            # Make sure we also generate data from the warm-up task.\n            if self._warm_up_task is not None and self._has_warmed_up:\n                task = self._warm_up_task\n                self._warm_up_task = None\n            else:\n                task = self._worker.get_task()\n            if not task.shard_name:\n                if task.type == elasticdl_pb2.WAIT:\n                    self._pending_dataset = True\n                    logger.info(\"No tasks for now, maybe more later\")\n                else:\n                    logger.info(\"No more task, stopping\")\n                break\n            with self._lock:\n                if task.type == elasticdl_pb2.TRAIN_END_CALLBACK:\n                    self._pending_train_end_callback_task = task\n                    continue\n\n                self._pending_tasks.append(task)\n                if len(self._pending_tasks) == 1:\n                    self._current_task = task\n            for data in self.data_reader.read_records(task):\n                if data:\n                    yield data\n", "path": "elasticdl/python/worker/task_data_service.py"}], "after_files": [{"content": "import threading\nimport time\nfrom collections import deque\n\nimport tensorflow as tf\n\nfrom elasticdl.proto import elasticdl_pb2\nfrom elasticdl.python.common.constants import TaskExecCounterKey\nfrom elasticdl.python.common.log_utils import default_logger as logger\nfrom elasticdl.python.data.reader.data_reader_factory import create_data_reader\n\n\nclass TaskDataService(object):\n    def __init__(\n        self, worker, training_with_evaluation, data_reader_params=None\n    ):\n        self._worker = worker\n        self._create_data_reader_fn = create_data_reader\n        if self._worker._custom_data_reader is not None:\n            self._create_data_reader_fn = self._worker._custom_data_reader\n        self._training_with_evaluation = training_with_evaluation\n        self._lock = threading.Lock()\n        self._pending_dataset = True\n        self._pending_train_end_callback_task = None\n        if data_reader_params:\n            self.data_reader = self._create_data_reader_fn(\n                data_origin=None, **data_reader_params\n            )\n        else:\n            self.data_reader = self._create_data_reader_fn(data_origin=None)\n        self._warm_up_task = None\n        self._has_warmed_up = False\n        self._failed_record_count = 0\n        self._reported_record_count = 0\n        self._current_task = None\n        self._pending_tasks = deque()\n\n    def _reset(self):\n        \"\"\"\n        Reset pending tasks and record counts\n        \"\"\"\n        self._reported_record_count = 0\n        self._failed_record_count = 0\n        self._pending_tasks = deque()\n        self._current_task = None\n\n    def get_current_task(self):\n        return self._current_task\n\n    def _do_report_task(self, task, err_msg=\"\"):\n        if self._failed_record_count != 0:\n            exec_counters = {\n                TaskExecCounterKey.FAIL_COUNT: self._failed_record_count\n            }\n        else:\n            exec_counters = None\n        self._worker.report_task_result(\n            task.task_id, err_msg, exec_counters=exec_counters\n        )\n\n    def _log_fail_records(self, task, err_msg):\n        task_len = task.end - task.start\n        msg = (\n            \"records ({f}/{t}) failure, possible \"\n            \"in task_id: {task_id} \"\n            'reason \"{err_msg}\"'\n        ).format(\n            task_id=task.task_id,\n            err_msg=err_msg,\n            f=self._failed_record_count,\n            t=task_len,\n        )\n        logger.warning(msg)\n\n    def report_record_done(self, count, err_msg=\"\"):\n        \"\"\"\n        Report the number of records in the latest processed batch,\n        so TaskDataService knows if some pending tasks are finished\n        and report_task_result to the master.\n        Return True if there are some finished tasks, False otherwise.\n        \"\"\"\n        self._reported_record_count += count\n        if err_msg:\n            self._failed_record_count += count\n\n        # TODO(qijun) This is a workaround for #1829\n        if not self._pending_tasks:\n            return False\n        task = self._pending_tasks[0]\n        total_record_num = task.end - task.start\n        if self._reported_record_count >= total_record_num:\n            if err_msg:\n                self._log_fail_records(task, err_msg)\n\n            # Keep popping tasks until the reported record count is less\n            # than the size of the current data since `batch_size` may be\n            # larger than `task.end - task.start`\n            with self._lock:\n                while self._pending_tasks and self._reported_record_count >= (\n                    self._pending_tasks[0].end - self._pending_tasks[0].start\n                ):\n                    task = self._pending_tasks[0]\n                    self._reported_record_count -= task.end - task.start\n                    self._pending_tasks.popleft()\n                    self._do_report_task(task, err_msg)\n                    self._failed_record_count = 0\n                if self._pending_tasks:\n                    self._current_task = self._pending_tasks[0]\n            return True\n        return False\n\n    def get_dataset_gen(self, task):\n        \"\"\"\n        If a task exists, this creates a generator, which could be used to\n        creating a `tf.data.Dataset` object in further.\n        \"\"\"\n        if not task:\n            return None\n        tasks = [task]\n\n        def gen():\n            for task in tasks:\n                for data in self.data_reader.read_records(task):\n                    if data:\n                        yield data\n\n        return gen\n\n    def get_dataset_by_task(self, task):\n        if task is None:\n            return None\n        gen = self.get_dataset_gen(task)\n        dataset = tf.data.Dataset.from_generator(\n            gen, self.data_reader.records_output_types\n        )\n        return dataset\n\n    def get_train_end_callback_task(self):\n        return self._pending_train_end_callback_task\n\n    def clear_train_end_callback_task(self):\n        self._pending_train_end_callback_task = None\n\n    def get_dataset(self):\n        \"\"\"\n        If there's more data, this creates a `tf.data.Dataset` object.\n        Otherwise, this returns `None`.\n        \"\"\"\n        if self._pending_dataset:\n            if self._pending_tasks:\n                logger.error(\n                    \"Cannot get new dataset when there are pending tasks\"\n                )\n                return None\n            self._reset()\n            # We use a task to perform warm-up for data reader in order\n            # to collect useful metadata. Note that we only performs\n            # data fetching for this task and `break` instantly to make\n            # sure `read_records()` is executed without iterating all the\n            # records so this should not be time consuming.\n            if self._warm_up_task is None and not self._has_warmed_up:\n                while True:\n                    task = self._worker.get_task()\n                    if task.type != elasticdl_pb2.WAIT:\n                        break\n                    time.sleep(2)\n                if task.type == elasticdl_pb2.TRAIN_END_CALLBACK:\n                    self._pending_train_end_callback_task = task\n                    return None\n                elif not task.shard_name:\n                    logger.info(\"No more task, stopping\")\n                    return None\n                else:\n                    self._warm_up_task = task\n                    for _ in self.data_reader.read_records(task):\n                        break\n                    self._has_warmed_up = True\n            ds = tf.data.Dataset.from_generator(\n                self._gen, self.data_reader.records_output_types\n            )\n            self._pending_dataset = False\n            return ds\n        else:\n            return None\n\n    def _gen(self):\n        \"\"\"\n        A generator supports the iter() protocol (e.g. a generator function),\n        used to create a `tf.data.Dataset` object from a list of tasks.\n        \"\"\"\n        while True:\n            # Make sure we also generate data from the warm-up task.\n            if self._warm_up_task is not None and self._has_warmed_up:\n                task = self._warm_up_task\n                self._warm_up_task = None\n            else:\n                task = self._worker.get_task()\n            if not task.shard_name:\n                if task.type == elasticdl_pb2.WAIT:\n                    self._pending_dataset = True\n                    logger.info(\"No tasks for now, maybe more later\")\n                else:\n                    logger.info(\"No more task, stopping\")\n                break\n            with self._lock:\n                if task.type == elasticdl_pb2.TRAIN_END_CALLBACK:\n                    self._pending_train_end_callback_task = task\n                    continue\n\n                self._pending_tasks.append(task)\n                if len(self._pending_tasks) == 1:\n                    self._current_task = task\n            for data in self.data_reader.read_records(task):\n                if data:\n                    yield data\n", "path": "elasticdl/python/worker/task_data_service.py"}]}