We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Only one task is created for each epoch for ODPS data source
There are more than 100 observations in the ODPS table and `--minibatch_size=10`, `--records_per_task=10`, and `--num_epochs=5` are used. However, only one task is created for each epoch. This only happens for ODPS data source though. RecordIO data source is working correctly as usual.

Here are the logs from master pod for a job with :
```
[2019-09-25 14:20:31,653] [INFO] [task_dispatcher.py:80:create_tasks] Creating a new set of training tasks for model version -1
[2019-09-25 14:20:31,654] [INFO] [task_dispatcher.py:118:get] Starting epoch 1
[2019-09-25 14:20:32,201] [INFO] [task_dispatcher.py:151:report] Task:1 completed, 1 remaining tasks
[2019-09-25 14:20:32,563] [INFO] [task_dispatcher.py:80:create_tasks] Creating a new set of training tasks for model version -1
[2019-09-25 14:20:32,563] [INFO] [task_dispatcher.py:118:get] Starting epoch 2
[2019-09-25 14:20:32,575] [INFO] [task_dispatcher.py:151:report] Task:2 completed, 1 remaining tasks
[2019-09-25 14:20:33,320] [INFO] [task_dispatcher.py:80:create_tasks] Creating a new set of training tasks for model version -1
[2019-09-25 14:20:33,320] [INFO] [task_dispatcher.py:118:get] Starting epoch 3
[2019-09-25 14:20:33,325] [INFO] [task_dispatcher.py:151:report] Task:3 completed, 1 remaining tasks
[2019-09-25 14:20:34,163] [INFO] [task_dispatcher.py:80:create_tasks] Creating a new set of training tasks for model version -1
[2019-09-25 14:20:34,163] [INFO] [task_dispatcher.py:118:get] Starting epoch 4
[2019-09-25 14:20:34,169] [INFO] [task_dispatcher.py:151:report] Task:4 completed, 1 remaining tasks
[2019-09-25 14:20:34,910] [INFO] [task_dispatcher.py:151:report] Task:5 completed, 0 remaining tasks
[2019-09-25 14:20:35,762] [INFO] [k8s_worker_manager.py:120:_event_cb] Got event MODIFIED, phase Succeeded for pod: elasticdl-test-iris-custom-model-odps-worker-0
[2019-09-25 14:20:35,784] [INFO] [k8s_worker_manager.py:120:_event_cb] Got event MODIFIED, phase Succeeded for pod: elasticdl-test-iris-custom-model-odps-worker-0
[2019-09-25 14:20:35,951] [INFO] [k8s_worker_manager.py:120:_event_cb] Got event MODIFIED, phase Succeeded for pod: elasticdl-test-iris-custom-model-odps-worker-0
[2019-09-25 14:20:35,983] [INFO] [k8s_worker_manager.py:120:_event_cb] Got event MODIFIED, phase Succeeded for pod: elasticdl-test-iris-custom-model-odps-worker-0
[2019-09-25 14:20:46,186] [INFO] [k8s_worker_manager.py:120:_event_cb] Got event MODIFIED, phase Running for pod: elasticdl-test-iris-custom-model-odps-master
[2019-09-25 14:20:46,188] [INFO] [main.py:294:main] Stopping RPC server
[2019-09-25 14:20:46,188] [INFO] [main.py:311:main] Master stopped
```

--- 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/master/task_dispatcher.py`
Content:
```
1 """TaskQueue Implementation"""
2 
3 import random
4 import threading
5 
6 from elasticdl.proto import elasticdl_pb2
7 from elasticdl.python.common.log_util import default_logger as logger
8 
9 
10 class _Task(object):
11     """Internal representation of a task"""
12 
13     def __init__(self, *, shard_name, start, end, type, model_version=-1):
14         self.shard_name = shard_name
15         self.start = start
16         self.end = end
17         self.type = type
18         self.model_version = model_version
19 
20     def _info(self):
21         return (
22             self.shard_name,
23             self.start,
24             self.end,
25             self.type,
26             self.model_version,
27         )
28 
29 
30 class _TaskDispatcher(object):
31     """Creates and dispatches Tasks. Keep track of a Task's lifecycle."""
32 
33     def __init__(
34         self,
35         training_shards,
36         evaluation_shards,
37         prediction_shards,
38         records_per_task,
39         num_epochs,
40     ):
41         """
42         Arguments:
43             training_shards: A dictionary from RecordIO file name to the
44                 number of training records.
45             evaluation_shards: A dictionary from RecordIO file name to
46                 the number of evaluation records.
47             prediction_shards: A dictionary from RecordIO file name to
48                 the number of prediction records.
49             records_per_task: The number of records per task.
50             num_epochs: The total number of epochs for the tasks where
51                 an epoch is a complete iteration over the shards.
52         """
53         self._lock = threading.Lock()
54 
55         self._num_epochs = num_epochs
56         self._epoch = 0
57         self._training_shards = training_shards
58         self._evaluation_shards = evaluation_shards
59         self._prediction_shards = prediction_shards
60         self._records_per_task = records_per_task
61 
62         self._todo = []
63         # dictionary from task id to Task.
64         self._doing = {}
65         self._task_id = 0
66         self._evaluation_service = None
67 
68         if self._training_shards:
69             logger.info("Starting epoch %d", self._epoch)
70             self.create_tasks(elasticdl_pb2.TRAINING)
71         elif self._evaluation_shards:
72             self.create_tasks(elasticdl_pb2.EVALUATION)
73         elif self._prediction_shards:
74             self.create_tasks(elasticdl_pb2.PREDICTION)
75 
76     def create_tasks(self, task_type, model_version=-1):
77         logger.info(
78             "Creating a new set of %s tasks for model version %d",
79             elasticdl_pb2._TASKTYPE.values_by_number[task_type].name.lower(),
80             model_version,
81         )
82         if task_type == elasticdl_pb2.TRAINING:
83             shards = self._training_shards
84         elif task_type == elasticdl_pb2.EVALUATION:
85             shards = self._evaluation_shards
86         else:
87             shards = self._prediction_shards
88         tasks = []
89         for shard_name, (start_ind, num_records) in shards.items():
90             for start in range(start_ind, num_records, self._records_per_task):
91                 tasks.append(
92                     _Task(
93                         shard_name=shard_name,
94                         start=start,
95                         end=min(start + self._records_per_task, num_records),
96                         type=task_type,
97                         model_version=model_version,
98                     )
99                 )
100         if task_type == elasticdl_pb2.TRAINING:
101             random.shuffle(tasks)
102             self._todo.extend(tasks)
103         else:
104             with self._lock:
105                 self._todo.extend(tasks)
106         return tasks
107 
108     def get(self, worker_id):
109         """Return next (task_id, Task) tuple"""
110 
111         with self._lock:
112             # TODO: check if task queue doesn't have training task,
113             #       to avoid the queue is overwhelmed by evaluation tasks.
114             if not self._todo and self._epoch < self._num_epochs - 1:
115                 # Start a new epoch
116                 self._epoch += 1
117                 self.create_tasks(elasticdl_pb2.TRAINING)
118                 logger.info("Starting epoch %d", self._epoch)
119 
120             if not self._todo:
121                 # No more tasks
122                 return -1, None
123 
124             self._task_id += 1
125             task = self._todo.pop()
126             # TODO: Handle timeout of tasks.
127             self._doing[self._task_id] = (worker_id, task)
128 
129             return self._task_id, task
130 
131     def report(self, task_id, success):
132         """Report if the task is successful or not"""
133 
134         evaluation_task_completed = False
135         with self._lock:
136             _, task = self._doing.pop(task_id, (-1, None))
137             if not task:
138                 logger.warning("Unknown task_id: %d" % task_id)
139             elif not success:
140                 # TODO: keep count of retries.
141                 self._todo.append(task)
142             elif (
143                 task.type == elasticdl_pb2.EVALUATION
144                 and self._evaluation_service is not None
145             ):
146                 evaluation_task_completed = True
147             else:
148                 logger.info(
149                     "Task:%d completed, %d remaining tasks",
150                     task_id,
151                     len(self._todo) + len(self._doing),
152                 )
153         if evaluation_task_completed:
154             self._evaluation_service.complete_task()
155 
156     def finished(self):
157         """Return if all tasks are done"""
158         return not self._todo and not self._doing
159 
160     def recover_tasks(self, worker_id):
161         """Recover doing tasks for a dead worker"""
162 
163         with self._lock:
164             ids = [
165                 id for id, (wid, _) in self._doing.items() if wid == worker_id
166             ]
167         for id in ids:
168             self.report(id, False)
169 
170     # TODO: need to re-check after refactoring servicer.py
171     def set_evaluation_service(self, evaluation_service):
172         with self._lock:
173             self._evaluation_service = evaluation_service
174             if self._evaluation_shards and not self._training_shards:
175                 evaluation_service.init_eval_only_job(len(self._todo))
176
```
--- 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/master/task_dispatcher.py b/elasticdl/python/master/task_dispatcher.py
--- a/elasticdl/python/master/task_dispatcher.py
+++ b/elasticdl/python/master/task_dispatcher.py
@@ -87,12 +87,24 @@
             shards = self._prediction_shards
         tasks = []
         for shard_name, (start_ind, num_records) in shards.items():
-            for start in range(start_ind, num_records, self._records_per_task):
+            for start in range(
+                start_ind, start_ind + num_records, self._records_per_task
+            ):
+                end_ind = min(start + self._records_per_task, num_records)
+                # If the start index is not smaller than end index,
+                # we need to take the start index into account
+                if start >= end_ind:
+                    end_ind = min(
+                        start + self._records_per_task, start + num_records
+                    )
+                # Note that only records in [start, end) of this task
+                # will be consumed later in the worker that handles
+                # this task.
                 tasks.append(
                     _Task(
                         shard_name=shard_name,
                         start=start,
-                        end=min(start + self._records_per_task, num_records),
+                        end=end_ind,
                         type=task_type,
                         model_version=model_version,
                     )

{"golden_diff": "diff --git a/elasticdl/python/master/task_dispatcher.py b/elasticdl/python/master/task_dispatcher.py\n--- a/elasticdl/python/master/task_dispatcher.py\n+++ b/elasticdl/python/master/task_dispatcher.py\n@@ -87,12 +87,24 @@\n             shards = self._prediction_shards\n         tasks = []\n         for shard_name, (start_ind, num_records) in shards.items():\n-            for start in range(start_ind, num_records, self._records_per_task):\n+            for start in range(\n+                start_ind, start_ind + num_records, self._records_per_task\n+            ):\n+                end_ind = min(start + self._records_per_task, num_records)\n+                # If the start index is not smaller than end index,\n+                # we need to take the start index into account\n+                if start >= end_ind:\n+                    end_ind = min(\n+                        start + self._records_per_task, start + num_records\n+                    )\n+                # Note that only records in [start, end) of this task\n+                # will be consumed later in the worker that handles\n+                # this task.\n                 tasks.append(\n                     _Task(\n                         shard_name=shard_name,\n                         start=start,\n-                        end=min(start + self._records_per_task, num_records),\n+                        end=end_ind,\n                         type=task_type,\n                         model_version=model_version,\n                     )\n", "issue": "Only one task is created for each epoch for ODPS data source\nThere are more than 100 observations in the ODPS table and `--minibatch_size=10`, `--records_per_task=10`, and `--num_epochs=5` are used. However, only one task is created for each epoch. This only happens for ODPS data source though. RecordIO data source is working correctly as usual.\r\n\r\nHere are the logs from master pod for a job with :\r\n```\r\n[2019-09-25 14:20:31,653] [INFO] [task_dispatcher.py:80:create_tasks] Creating a new set of training tasks for model version -1\r\n[2019-09-25 14:20:31,654] [INFO] [task_dispatcher.py:118:get] Starting epoch 1\r\n[2019-09-25 14:20:32,201] [INFO] [task_dispatcher.py:151:report] Task:1 completed, 1 remaining tasks\r\n[2019-09-25 14:20:32,563] [INFO] [task_dispatcher.py:80:create_tasks] Creating a new set of training tasks for model version -1\r\n[2019-09-25 14:20:32,563] [INFO] [task_dispatcher.py:118:get] Starting epoch 2\r\n[2019-09-25 14:20:32,575] [INFO] [task_dispatcher.py:151:report] Task:2 completed, 1 remaining tasks\r\n[2019-09-25 14:20:33,320] [INFO] [task_dispatcher.py:80:create_tasks] Creating a new set of training tasks for model version -1\r\n[2019-09-25 14:20:33,320] [INFO] [task_dispatcher.py:118:get] Starting epoch 3\r\n[2019-09-25 14:20:33,325] [INFO] [task_dispatcher.py:151:report] Task:3 completed, 1 remaining tasks\r\n[2019-09-25 14:20:34,163] [INFO] [task_dispatcher.py:80:create_tasks] Creating a new set of training tasks for model version -1\r\n[2019-09-25 14:20:34,163] [INFO] [task_dispatcher.py:118:get] Starting epoch 4\r\n[2019-09-25 14:20:34,169] [INFO] [task_dispatcher.py:151:report] Task:4 completed, 1 remaining tasks\r\n[2019-09-25 14:20:34,910] [INFO] [task_dispatcher.py:151:report] Task:5 completed, 0 remaining tasks\r\n[2019-09-25 14:20:35,762] [INFO] [k8s_worker_manager.py:120:_event_cb] Got event MODIFIED, phase Succeeded for pod: elasticdl-test-iris-custom-model-odps-worker-0\r\n[2019-09-25 14:20:35,784] [INFO] [k8s_worker_manager.py:120:_event_cb] Got event MODIFIED, phase Succeeded for pod: elasticdl-test-iris-custom-model-odps-worker-0\r\n[2019-09-25 14:20:35,951] [INFO] [k8s_worker_manager.py:120:_event_cb] Got event MODIFIED, phase Succeeded for pod: elasticdl-test-iris-custom-model-odps-worker-0\r\n[2019-09-25 14:20:35,983] [INFO] [k8s_worker_manager.py:120:_event_cb] Got event MODIFIED, phase Succeeded for pod: elasticdl-test-iris-custom-model-odps-worker-0\r\n[2019-09-25 14:20:46,186] [INFO] [k8s_worker_manager.py:120:_event_cb] Got event MODIFIED, phase Running for pod: elasticdl-test-iris-custom-model-odps-master\r\n[2019-09-25 14:20:46,188] [INFO] [main.py:294:main] Stopping RPC server\r\n[2019-09-25 14:20:46,188] [INFO] [main.py:311:main] Master stopped\r\n```\n", "before_files": [{"content": "\"\"\"TaskQueue Implementation\"\"\"\n\nimport random\nimport threading\n\nfrom elasticdl.proto import elasticdl_pb2\nfrom elasticdl.python.common.log_util import default_logger as logger\n\n\nclass _Task(object):\n    \"\"\"Internal representation of a task\"\"\"\n\n    def __init__(self, *, shard_name, start, end, type, model_version=-1):\n        self.shard_name = shard_name\n        self.start = start\n        self.end = end\n        self.type = type\n        self.model_version = model_version\n\n    def _info(self):\n        return (\n            self.shard_name,\n            self.start,\n            self.end,\n            self.type,\n            self.model_version,\n        )\n\n\nclass _TaskDispatcher(object):\n    \"\"\"Creates and dispatches Tasks. Keep track of a Task's lifecycle.\"\"\"\n\n    def __init__(\n        self,\n        training_shards,\n        evaluation_shards,\n        prediction_shards,\n        records_per_task,\n        num_epochs,\n    ):\n        \"\"\"\n        Arguments:\n            training_shards: A dictionary from RecordIO file name to the\n                number of training records.\n            evaluation_shards: A dictionary from RecordIO file name to\n                the number of evaluation records.\n            prediction_shards: A dictionary from RecordIO file name to\n                the number of prediction records.\n            records_per_task: The number of records per task.\n            num_epochs: The total number of epochs for the tasks where\n                an epoch is a complete iteration over the shards.\n        \"\"\"\n        self._lock = threading.Lock()\n\n        self._num_epochs = num_epochs\n        self._epoch = 0\n        self._training_shards = training_shards\n        self._evaluation_shards = evaluation_shards\n        self._prediction_shards = prediction_shards\n        self._records_per_task = records_per_task\n\n        self._todo = []\n        # dictionary from task id to Task.\n        self._doing = {}\n        self._task_id = 0\n        self._evaluation_service = None\n\n        if self._training_shards:\n            logger.info(\"Starting epoch %d\", self._epoch)\n            self.create_tasks(elasticdl_pb2.TRAINING)\n        elif self._evaluation_shards:\n            self.create_tasks(elasticdl_pb2.EVALUATION)\n        elif self._prediction_shards:\n            self.create_tasks(elasticdl_pb2.PREDICTION)\n\n    def create_tasks(self, task_type, model_version=-1):\n        logger.info(\n            \"Creating a new set of %s tasks for model version %d\",\n            elasticdl_pb2._TASKTYPE.values_by_number[task_type].name.lower(),\n            model_version,\n        )\n        if task_type == elasticdl_pb2.TRAINING:\n            shards = self._training_shards\n        elif task_type == elasticdl_pb2.EVALUATION:\n            shards = self._evaluation_shards\n        else:\n            shards = self._prediction_shards\n        tasks = []\n        for shard_name, (start_ind, num_records) in shards.items():\n            for start in range(start_ind, num_records, self._records_per_task):\n                tasks.append(\n                    _Task(\n                        shard_name=shard_name,\n                        start=start,\n                        end=min(start + self._records_per_task, num_records),\n                        type=task_type,\n                        model_version=model_version,\n                    )\n                )\n        if task_type == elasticdl_pb2.TRAINING:\n            random.shuffle(tasks)\n            self._todo.extend(tasks)\n        else:\n            with self._lock:\n                self._todo.extend(tasks)\n        return tasks\n\n    def get(self, worker_id):\n        \"\"\"Return next (task_id, Task) tuple\"\"\"\n\n        with self._lock:\n            # TODO: check if task queue doesn't have training task,\n            #       to avoid the queue is overwhelmed by evaluation tasks.\n            if not self._todo and self._epoch < self._num_epochs - 1:\n                # Start a new epoch\n                self._epoch += 1\n                self.create_tasks(elasticdl_pb2.TRAINING)\n                logger.info(\"Starting epoch %d\", self._epoch)\n\n            if not self._todo:\n                # No more tasks\n                return -1, None\n\n            self._task_id += 1\n            task = self._todo.pop()\n            # TODO: Handle timeout of tasks.\n            self._doing[self._task_id] = (worker_id, task)\n\n            return self._task_id, task\n\n    def report(self, task_id, success):\n        \"\"\"Report if the task is successful or not\"\"\"\n\n        evaluation_task_completed = False\n        with self._lock:\n            _, task = self._doing.pop(task_id, (-1, None))\n            if not task:\n                logger.warning(\"Unknown task_id: %d\" % task_id)\n            elif not success:\n                # TODO: keep count of retries.\n                self._todo.append(task)\n            elif (\n                task.type == elasticdl_pb2.EVALUATION\n                and self._evaluation_service is not None\n            ):\n                evaluation_task_completed = True\n            else:\n                logger.info(\n                    \"Task:%d completed, %d remaining tasks\",\n                    task_id,\n                    len(self._todo) + len(self._doing),\n                )\n        if evaluation_task_completed:\n            self._evaluation_service.complete_task()\n\n    def finished(self):\n        \"\"\"Return if all tasks are done\"\"\"\n        return not self._todo and not self._doing\n\n    def recover_tasks(self, worker_id):\n        \"\"\"Recover doing tasks for a dead worker\"\"\"\n\n        with self._lock:\n            ids = [\n                id for id, (wid, _) in self._doing.items() if wid == worker_id\n            ]\n        for id in ids:\n            self.report(id, False)\n\n    # TODO: need to re-check after refactoring servicer.py\n    def set_evaluation_service(self, evaluation_service):\n        with self._lock:\n            self._evaluation_service = evaluation_service\n            if self._evaluation_shards and not self._training_shards:\n                evaluation_service.init_eval_only_job(len(self._todo))\n", "path": "elasticdl/python/master/task_dispatcher.py"}], "after_files": [{"content": "\"\"\"TaskQueue Implementation\"\"\"\n\nimport random\nimport threading\n\nfrom elasticdl.proto import elasticdl_pb2\nfrom elasticdl.python.common.log_util import default_logger as logger\n\n\nclass _Task(object):\n    \"\"\"Internal representation of a task\"\"\"\n\n    def __init__(self, *, shard_name, start, end, type, model_version=-1):\n        self.shard_name = shard_name\n        self.start = start\n        self.end = end\n        self.type = type\n        self.model_version = model_version\n\n    def _info(self):\n        return (\n            self.shard_name,\n            self.start,\n            self.end,\n            self.type,\n            self.model_version,\n        )\n\n\nclass _TaskDispatcher(object):\n    \"\"\"Creates and dispatches Tasks. Keep track of a Task's lifecycle.\"\"\"\n\n    def __init__(\n        self,\n        training_shards,\n        evaluation_shards,\n        prediction_shards,\n        records_per_task,\n        num_epochs,\n    ):\n        \"\"\"\n        Arguments:\n            training_shards: A dictionary from RecordIO file name to the\n                number of training records.\n            evaluation_shards: A dictionary from RecordIO file name to\n                the number of evaluation records.\n            prediction_shards: A dictionary from RecordIO file name to\n                the number of prediction records.\n            records_per_task: The number of records per task.\n            num_epochs: The total number of epochs for the tasks where\n                an epoch is a complete iteration over the shards.\n        \"\"\"\n        self._lock = threading.Lock()\n\n        self._num_epochs = num_epochs\n        self._epoch = 0\n        self._training_shards = training_shards\n        self._evaluation_shards = evaluation_shards\n        self._prediction_shards = prediction_shards\n        self._records_per_task = records_per_task\n\n        self._todo = []\n        # dictionary from task id to Task.\n        self._doing = {}\n        self._task_id = 0\n        self._evaluation_service = None\n\n        if self._training_shards:\n            logger.info(\"Starting epoch %d\", self._epoch)\n            self.create_tasks(elasticdl_pb2.TRAINING)\n        elif self._evaluation_shards:\n            self.create_tasks(elasticdl_pb2.EVALUATION)\n        elif self._prediction_shards:\n            self.create_tasks(elasticdl_pb2.PREDICTION)\n\n    def create_tasks(self, task_type, model_version=-1):\n        logger.info(\n            \"Creating a new set of %s tasks for model version %d\",\n            elasticdl_pb2._TASKTYPE.values_by_number[task_type].name.lower(),\n            model_version,\n        )\n        if task_type == elasticdl_pb2.TRAINING:\n            shards = self._training_shards\n        elif task_type == elasticdl_pb2.EVALUATION:\n            shards = self._evaluation_shards\n        else:\n            shards = self._prediction_shards\n        tasks = []\n        for shard_name, (start_ind, num_records) in shards.items():\n            for start in range(\n                start_ind, start_ind + num_records, self._records_per_task\n            ):\n                end_ind = min(start + self._records_per_task, num_records)\n                # If the start index is not smaller than end index,\n                # we need to take the start index into account\n                if start >= end_ind:\n                    end_ind = min(\n                        start + self._records_per_task, start + num_records\n                    )\n                # Note that only records in [start, end) of this task\n                # will be consumed later in the worker that handles\n                # this task.\n                tasks.append(\n                    _Task(\n                        shard_name=shard_name,\n                        start=start,\n                        end=end_ind,\n                        type=task_type,\n                        model_version=model_version,\n                    )\n                )\n        if task_type == elasticdl_pb2.TRAINING:\n            random.shuffle(tasks)\n            self._todo.extend(tasks)\n        else:\n            with self._lock:\n                self._todo.extend(tasks)\n        return tasks\n\n    def get(self, worker_id):\n        \"\"\"Return next (task_id, Task) tuple\"\"\"\n\n        with self._lock:\n            # TODO: check if task queue doesn't have training task,\n            #       to avoid the queue is overwhelmed by evaluation tasks.\n            if not self._todo and self._epoch < self._num_epochs - 1:\n                # Start a new epoch\n                self._epoch += 1\n                self.create_tasks(elasticdl_pb2.TRAINING)\n                logger.info(\"Starting epoch %d\", self._epoch)\n\n            if not self._todo:\n                # No more tasks\n                return -1, None\n\n            self._task_id += 1\n            task = self._todo.pop()\n            # TODO: Handle timeout of tasks.\n            self._doing[self._task_id] = (worker_id, task)\n\n            return self._task_id, task\n\n    def report(self, task_id, success):\n        \"\"\"Report if the task is successful or not\"\"\"\n\n        evaluation_task_completed = False\n        with self._lock:\n            _, task = self._doing.pop(task_id, (-1, None))\n            if not task:\n                logger.warning(\"Unknown task_id: %d\" % task_id)\n            elif not success:\n                # TODO: keep count of retries.\n                self._todo.append(task)\n            elif (\n                task.type == elasticdl_pb2.EVALUATION\n                and self._evaluation_service is not None\n            ):\n                evaluation_task_completed = True\n            else:\n                logger.info(\n                    \"Task:%d completed, %d remaining tasks\",\n                    task_id,\n                    len(self._todo) + len(self._doing),\n                )\n        if evaluation_task_completed:\n            self._evaluation_service.complete_task()\n\n    def finished(self):\n        \"\"\"Return if all tasks are done\"\"\"\n        return not self._todo and not self._doing\n\n    def recover_tasks(self, worker_id):\n        \"\"\"Recover doing tasks for a dead worker\"\"\"\n\n        with self._lock:\n            ids = [\n                id for id, (wid, _) in self._doing.items() if wid == worker_id\n            ]\n        for id in ids:\n            self.report(id, False)\n\n    # TODO: need to re-check after refactoring servicer.py\n    def set_evaluation_service(self, evaluation_service):\n        with self._lock:\n            self._evaluation_service = evaluation_service\n            if self._evaluation_shards and not self._training_shards:\n                evaluation_service.init_eval_only_job(len(self._todo))\n", "path": "elasticdl/python/master/task_dispatcher.py"}]}