We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
keep track of the correct min_model_version value
`min_model_version` should be initialized at the beginning of `_handle_task`, and updated with minibatch training.

fix #659 

--- 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/elasticdl/worker/worker.py`
Content:
```
1 import logging
2 import traceback
3 
4 import tensorflow as tf
5 
6 assert tf.executing_eagerly()  # noqa
7 
8 import recordio
9 
10 from contextlib import closing
11 from elasticdl.proto import elasticdl_pb2_grpc
12 from elasticdl.proto import elasticdl_pb2
13 from elasticdl.python.elasticdl.common.ndarray import (
14     ndarray_to_tensor,
15     tensor_to_ndarray,
16 )
17 from elasticdl.python.elasticdl.common.model_helper import (
18     load_user_model,
19     build_model,
20 )
21 from elasticdl.python.data.codec import TFExampleCodec
22 from elasticdl.python.data.codec import BytesCodec
23 
24 # The default maximum number of a minibatch retry as its results
25 # (e.g. gradients) are not accepted by master.
26 DEFAULT_MAX_MINIBATCH_RETRY_NUM = 64
27 
28 
29 class Worker(object):
30     """ElasticDL worker"""
31 
32     def __init__(
33         self,
34         worker_id,
35         model_file,
36         channel=None,
37         max_minibatch_retry_num=DEFAULT_MAX_MINIBATCH_RETRY_NUM,
38         codec_type=None,
39     ):
40         """
41         Arguments:
42             model_file: A module to define the model
43             channel: grpc channel
44             max_minibatch_retry_num: The maximum number of a minibatch retry
45                 as its results (e.g. gradients) are not accepted by master.
46         """
47         self._logger = logging.getLogger(__name__)
48         self._worker_id = worker_id
49         model_module = load_user_model(model_file)
50         self._model = model_module.model
51         self._feature_columns = model_module.feature_columns()
52         build_model(self._model, self._feature_columns)
53         self._input_fn = model_module.input_fn
54         self._opt_fn = model_module.optimizer
55         self._loss = model_module.loss
56         self._eval_metrics_fn = model_module.eval_metrics_fn
57         all_columns = self._feature_columns + model_module.label_columns()
58         if codec_type == "tf_example":
59             self._codec = TFExampleCodec(all_columns)
60         elif codec_type == "bytes":
61             self._codec = BytesCodec(all_columns)
62         else:
63             raise ValueError("invalid codec_type: " + codec_type)
64 
65         if channel is None:
66             self._stub = None
67         else:
68             self._stub = elasticdl_pb2_grpc.MasterStub(channel)
69         self._max_minibatch_retry_num = max_minibatch_retry_num
70         self._model_version = -1
71         self._codec_type = codec_type
72 
73     def get_task(self):
74         """
75         get task from master
76         """
77         req = elasticdl_pb2.GetTaskRequest()
78         req.worker_id = self._worker_id
79 
80         return self._stub.GetTask(req)
81 
82     def get_model(self, min_version):
83         """
84         get model from master, and update model_version
85         """
86         req = elasticdl_pb2.GetModelRequest()
87         req.min_version = min_version
88         model = self._stub.GetModel(req)
89 
90         for var in self._model.trainable_variables:
91             # Assumes all trainable variables exist in model.param.
92             var.assign(tensor_to_ndarray(model.param[var.name]))
93         self._model_version = model.version
94 
95     def report_task_result(self, task_id, err_msg):
96         """
97         report task result to master
98         """
99         report = elasticdl_pb2.ReportTaskResultRequest()
100         report.task_id = task_id
101         report.err_message = err_msg
102         return self._stub.ReportTaskResult(report)
103 
104     def report_gradient(self, grads):
105         """
106         report gradient to ps, return (accepted, model_version) from rpc call.
107         """
108         req = elasticdl_pb2.ReportGradientRequest()
109         for g, v in zip(grads, self._model.trainable_variables):
110             req.gradient[v.name].CopyFrom(ndarray_to_tensor(g.numpy()))
111         req.model_version = self._model_version
112         res = self._stub.ReportGradient(req)
113         return res.accepted, res.model_version
114 
115     def report_evaluation_metrics(self, evaluation_metrics):
116         """
117         report evaluation metrics to ps, return (accepted, model_version)
118         from rpc call.
119         """
120         req = elasticdl_pb2.ReportEvaluationMetricsRequest()
121         for k, v in evaluation_metrics.items():
122             v_np = v.numpy()
123             if v_np.size != 1:
124                 raise Exception(
125                     "Only metric result of length 1 is " "supported currently"
126                 )
127             req.evaluation_metrics[k].CopyFrom(ndarray_to_tensor(v_np))
128         req.model_version = self._model_version
129         res = self._stub.ReportEvaluationMetrics(req)
130         return res.accepted, res.model_version
131 
132     @staticmethod
133     def _get_batch(reader, batch_size, decode):
134         res = []
135         for i in range(batch_size):
136             record = reader.record()
137             if record is None:
138                 break
139             res.append(decode(record))
140         return res
141 
142     def _get_features_and_labels(self, record_buf):
143         batch_input_data, batch_labels = self._input_fn(record_buf)
144         features = [
145             batch_input_data[f_col.key] for f_col in self._feature_columns
146         ]
147         if len(features) == 1:
148             features = features[0]
149         return features, batch_labels
150 
151     def _run_training_task(self, features, labels):
152         with tf.GradientTape() as tape:
153             outputs = self._model.call(features, training=True)
154             loss = self._loss(outputs, labels)
155 
156             # TODO:  Add regularization loss if any,
157             #        which should be divided by the
158             #        number of contributing workers.
159         grads = tape.gradient(loss, self._model.trainable_variables)
160         accepted, min_model_version = self.report_gradient(grads)
161         return accepted, min_model_version, loss
162 
163     def _run_evaluation_task(self, features, labels):
164         outputs = self._model.call(features, training=False)
165         evaluation_metrics = self._eval_metrics_fn(outputs, labels)
166         return self.report_evaluation_metrics(evaluation_metrics)
167 
168     def _handle_task(self, task):
169         with closing(
170             recordio.Scanner(
171                 task.shard_file_name, task.start, task.end - task.start
172             )
173         ) as reader:
174             while True:
175                 record_buf = self._get_batch(
176                     reader, task.minibatch_size, self._codec.decode
177                 )
178                 if not record_buf:
179                     break
180                 self._process_minibatch(task, record_buf)
181 
182     def _process_minibatch(self, task, record_buf):
183         min_model_version = task.model_version
184         for _ in range(self._max_minibatch_retry_num):
185             # TODO: optimize the logic to avoid unnecessary
186             #       get_model call.
187             self.get_model(max(self._model_version, min_model_version))
188             features, labels = self._get_features_and_labels(record_buf)
189             if task.type == elasticdl_pb2.EVALUATION:
190                 accepted, min_model_version = self._run_evaluation_task(
191                     features, labels
192                 )
193                 if accepted:
194                     break
195             elif task.type == elasticdl_pb2.TRAINING:
196                 accepted, min_model_version, loss = self._run_training_task(
197                     features, labels
198                 )
199                 if accepted:
200                     self._logger.info("Loss is %f" % loss.numpy())
201                     break
202             else:
203                 raise RuntimeError("Unrecognized task type, %s" % task.type)
204         else:
205             # Worker got stuck, fail the task.
206             # TODO: stop the worker if it fails to make any
207             #       progress for some time.
208             raise RuntimeError("Worker got stuck")
209 
210     def run(self):
211         """
212         Fetches task from master and performs training or evaluation.
213         """
214         while True:
215             task = self.get_task()
216             if not task.shard_file_name:
217                 # No more task
218                 break
219             err_msg = ""
220             try:
221                 self._handle_task(task)
222             except Exception as ex:
223                 err_msg = str(ex)
224                 traceback.print_exc()
225             self.report_task_result(task.task_id, err_msg)
226
```
--- 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/elasticdl/worker/worker.py b/elasticdl/python/elasticdl/worker/worker.py
--- a/elasticdl/python/elasticdl/worker/worker.py
+++ b/elasticdl/python/elasticdl/worker/worker.py
@@ -166,6 +166,7 @@
         return self.report_evaluation_metrics(evaluation_metrics)
 
     def _handle_task(self, task):
+        min_model_version = task.model_version
         with closing(
             recordio.Scanner(
                 task.shard_file_name, task.start, task.end - task.start
@@ -177,10 +178,11 @@
                 )
                 if not record_buf:
                     break
-                self._process_minibatch(task, record_buf)
+                min_model_version = self._process_minibatch(
+                    task, record_buf, min_model_version
+                )
 
-    def _process_minibatch(self, task, record_buf):
-        min_model_version = task.model_version
+    def _process_minibatch(self, task, record_buf, min_model_version):
         for _ in range(self._max_minibatch_retry_num):
             # TODO: optimize the logic to avoid unnecessary
             #       get_model call.
@@ -206,6 +208,7 @@
             # TODO: stop the worker if it fails to make any
             #       progress for some time.
             raise RuntimeError("Worker got stuck")
+        return min_model_version
 
     def run(self):
         """

{"golden_diff": "diff --git a/elasticdl/python/elasticdl/worker/worker.py b/elasticdl/python/elasticdl/worker/worker.py\n--- a/elasticdl/python/elasticdl/worker/worker.py\n+++ b/elasticdl/python/elasticdl/worker/worker.py\n@@ -166,6 +166,7 @@\n         return self.report_evaluation_metrics(evaluation_metrics)\n \n     def _handle_task(self, task):\n+        min_model_version = task.model_version\n         with closing(\n             recordio.Scanner(\n                 task.shard_file_name, task.start, task.end - task.start\n@@ -177,10 +178,11 @@\n                 )\n                 if not record_buf:\n                     break\n-                self._process_minibatch(task, record_buf)\n+                min_model_version = self._process_minibatch(\n+                    task, record_buf, min_model_version\n+                )\n \n-    def _process_minibatch(self, task, record_buf):\n-        min_model_version = task.model_version\n+    def _process_minibatch(self, task, record_buf, min_model_version):\n         for _ in range(self._max_minibatch_retry_num):\n             # TODO: optimize the logic to avoid unnecessary\n             #       get_model call.\n@@ -206,6 +208,7 @@\n             # TODO: stop the worker if it fails to make any\n             #       progress for some time.\n             raise RuntimeError(\"Worker got stuck\")\n+        return min_model_version\n \n     def run(self):\n         \"\"\"\n", "issue": "keep track of the correct min_model_version value\n`min_model_version` should be initialized at the beginning of `_handle_task`, and updated with minibatch training.\r\n\r\nfix #659 \n", "before_files": [{"content": "import logging\nimport traceback\n\nimport tensorflow as tf\n\nassert tf.executing_eagerly()  # noqa\n\nimport recordio\n\nfrom contextlib import closing\nfrom elasticdl.proto import elasticdl_pb2_grpc\nfrom elasticdl.proto import elasticdl_pb2\nfrom elasticdl.python.elasticdl.common.ndarray import (\n    ndarray_to_tensor,\n    tensor_to_ndarray,\n)\nfrom elasticdl.python.elasticdl.common.model_helper import (\n    load_user_model,\n    build_model,\n)\nfrom elasticdl.python.data.codec import TFExampleCodec\nfrom elasticdl.python.data.codec import BytesCodec\n\n# The default maximum number of a minibatch retry as its results\n# (e.g. gradients) are not accepted by master.\nDEFAULT_MAX_MINIBATCH_RETRY_NUM = 64\n\n\nclass Worker(object):\n    \"\"\"ElasticDL worker\"\"\"\n\n    def __init__(\n        self,\n        worker_id,\n        model_file,\n        channel=None,\n        max_minibatch_retry_num=DEFAULT_MAX_MINIBATCH_RETRY_NUM,\n        codec_type=None,\n    ):\n        \"\"\"\n        Arguments:\n            model_file: A module to define the model\n            channel: grpc channel\n            max_minibatch_retry_num: The maximum number of a minibatch retry\n                as its results (e.g. gradients) are not accepted by master.\n        \"\"\"\n        self._logger = logging.getLogger(__name__)\n        self._worker_id = worker_id\n        model_module = load_user_model(model_file)\n        self._model = model_module.model\n        self._feature_columns = model_module.feature_columns()\n        build_model(self._model, self._feature_columns)\n        self._input_fn = model_module.input_fn\n        self._opt_fn = model_module.optimizer\n        self._loss = model_module.loss\n        self._eval_metrics_fn = model_module.eval_metrics_fn\n        all_columns = self._feature_columns + model_module.label_columns()\n        if codec_type == \"tf_example\":\n            self._codec = TFExampleCodec(all_columns)\n        elif codec_type == \"bytes\":\n            self._codec = BytesCodec(all_columns)\n        else:\n            raise ValueError(\"invalid codec_type: \" + codec_type)\n\n        if channel is None:\n            self._stub = None\n        else:\n            self._stub = elasticdl_pb2_grpc.MasterStub(channel)\n        self._max_minibatch_retry_num = max_minibatch_retry_num\n        self._model_version = -1\n        self._codec_type = codec_type\n\n    def get_task(self):\n        \"\"\"\n        get task from master\n        \"\"\"\n        req = elasticdl_pb2.GetTaskRequest()\n        req.worker_id = self._worker_id\n\n        return self._stub.GetTask(req)\n\n    def get_model(self, min_version):\n        \"\"\"\n        get model from master, and update model_version\n        \"\"\"\n        req = elasticdl_pb2.GetModelRequest()\n        req.min_version = min_version\n        model = self._stub.GetModel(req)\n\n        for var in self._model.trainable_variables:\n            # Assumes all trainable variables exist in model.param.\n            var.assign(tensor_to_ndarray(model.param[var.name]))\n        self._model_version = model.version\n\n    def report_task_result(self, task_id, err_msg):\n        \"\"\"\n        report task result to master\n        \"\"\"\n        report = elasticdl_pb2.ReportTaskResultRequest()\n        report.task_id = task_id\n        report.err_message = err_msg\n        return self._stub.ReportTaskResult(report)\n\n    def report_gradient(self, grads):\n        \"\"\"\n        report gradient to ps, return (accepted, model_version) from rpc call.\n        \"\"\"\n        req = elasticdl_pb2.ReportGradientRequest()\n        for g, v in zip(grads, self._model.trainable_variables):\n            req.gradient[v.name].CopyFrom(ndarray_to_tensor(g.numpy()))\n        req.model_version = self._model_version\n        res = self._stub.ReportGradient(req)\n        return res.accepted, res.model_version\n\n    def report_evaluation_metrics(self, evaluation_metrics):\n        \"\"\"\n        report evaluation metrics to ps, return (accepted, model_version)\n        from rpc call.\n        \"\"\"\n        req = elasticdl_pb2.ReportEvaluationMetricsRequest()\n        for k, v in evaluation_metrics.items():\n            v_np = v.numpy()\n            if v_np.size != 1:\n                raise Exception(\n                    \"Only metric result of length 1 is \" \"supported currently\"\n                )\n            req.evaluation_metrics[k].CopyFrom(ndarray_to_tensor(v_np))\n        req.model_version = self._model_version\n        res = self._stub.ReportEvaluationMetrics(req)\n        return res.accepted, res.model_version\n\n    @staticmethod\n    def _get_batch(reader, batch_size, decode):\n        res = []\n        for i in range(batch_size):\n            record = reader.record()\n            if record is None:\n                break\n            res.append(decode(record))\n        return res\n\n    def _get_features_and_labels(self, record_buf):\n        batch_input_data, batch_labels = self._input_fn(record_buf)\n        features = [\n            batch_input_data[f_col.key] for f_col in self._feature_columns\n        ]\n        if len(features) == 1:\n            features = features[0]\n        return features, batch_labels\n\n    def _run_training_task(self, features, labels):\n        with tf.GradientTape() as tape:\n            outputs = self._model.call(features, training=True)\n            loss = self._loss(outputs, labels)\n\n            # TODO:  Add regularization loss if any,\n            #        which should be divided by the\n            #        number of contributing workers.\n        grads = tape.gradient(loss, self._model.trainable_variables)\n        accepted, min_model_version = self.report_gradient(grads)\n        return accepted, min_model_version, loss\n\n    def _run_evaluation_task(self, features, labels):\n        outputs = self._model.call(features, training=False)\n        evaluation_metrics = self._eval_metrics_fn(outputs, labels)\n        return self.report_evaluation_metrics(evaluation_metrics)\n\n    def _handle_task(self, task):\n        with closing(\n            recordio.Scanner(\n                task.shard_file_name, task.start, task.end - task.start\n            )\n        ) as reader:\n            while True:\n                record_buf = self._get_batch(\n                    reader, task.minibatch_size, self._codec.decode\n                )\n                if not record_buf:\n                    break\n                self._process_minibatch(task, record_buf)\n\n    def _process_minibatch(self, task, record_buf):\n        min_model_version = task.model_version\n        for _ in range(self._max_minibatch_retry_num):\n            # TODO: optimize the logic to avoid unnecessary\n            #       get_model call.\n            self.get_model(max(self._model_version, min_model_version))\n            features, labels = self._get_features_and_labels(record_buf)\n            if task.type == elasticdl_pb2.EVALUATION:\n                accepted, min_model_version = self._run_evaluation_task(\n                    features, labels\n                )\n                if accepted:\n                    break\n            elif task.type == elasticdl_pb2.TRAINING:\n                accepted, min_model_version, loss = self._run_training_task(\n                    features, labels\n                )\n                if accepted:\n                    self._logger.info(\"Loss is %f\" % loss.numpy())\n                    break\n            else:\n                raise RuntimeError(\"Unrecognized task type, %s\" % task.type)\n        else:\n            # Worker got stuck, fail the task.\n            # TODO: stop the worker if it fails to make any\n            #       progress for some time.\n            raise RuntimeError(\"Worker got stuck\")\n\n    def run(self):\n        \"\"\"\n        Fetches task from master and performs training or evaluation.\n        \"\"\"\n        while True:\n            task = self.get_task()\n            if not task.shard_file_name:\n                # No more task\n                break\n            err_msg = \"\"\n            try:\n                self._handle_task(task)\n            except Exception as ex:\n                err_msg = str(ex)\n                traceback.print_exc()\n            self.report_task_result(task.task_id, err_msg)\n", "path": "elasticdl/python/elasticdl/worker/worker.py"}], "after_files": [{"content": "import logging\nimport traceback\n\nimport tensorflow as tf\n\nassert tf.executing_eagerly()  # noqa\n\nimport recordio\n\nfrom contextlib import closing\nfrom elasticdl.proto import elasticdl_pb2_grpc\nfrom elasticdl.proto import elasticdl_pb2\nfrom elasticdl.python.elasticdl.common.ndarray import (\n    ndarray_to_tensor,\n    tensor_to_ndarray,\n)\nfrom elasticdl.python.elasticdl.common.model_helper import (\n    load_user_model,\n    build_model,\n)\nfrom elasticdl.python.data.codec import TFExampleCodec\nfrom elasticdl.python.data.codec import BytesCodec\n\n# The default maximum number of a minibatch retry as its results\n# (e.g. gradients) are not accepted by master.\nDEFAULT_MAX_MINIBATCH_RETRY_NUM = 64\n\n\nclass Worker(object):\n    \"\"\"ElasticDL worker\"\"\"\n\n    def __init__(\n        self,\n        worker_id,\n        model_file,\n        channel=None,\n        max_minibatch_retry_num=DEFAULT_MAX_MINIBATCH_RETRY_NUM,\n        codec_type=None,\n    ):\n        \"\"\"\n        Arguments:\n            model_file: A module to define the model\n            channel: grpc channel\n            max_minibatch_retry_num: The maximum number of a minibatch retry\n                as its results (e.g. gradients) are not accepted by master.\n        \"\"\"\n        self._logger = logging.getLogger(__name__)\n        self._worker_id = worker_id\n        model_module = load_user_model(model_file)\n        self._model = model_module.model\n        self._feature_columns = model_module.feature_columns()\n        build_model(self._model, self._feature_columns)\n        self._input_fn = model_module.input_fn\n        self._opt_fn = model_module.optimizer\n        self._loss = model_module.loss\n        self._eval_metrics_fn = model_module.eval_metrics_fn\n        all_columns = self._feature_columns + model_module.label_columns()\n        if codec_type == \"tf_example\":\n            self._codec = TFExampleCodec(all_columns)\n        elif codec_type == \"bytes\":\n            self._codec = BytesCodec(all_columns)\n        else:\n            raise ValueError(\"invalid codec_type: \" + codec_type)\n\n        if channel is None:\n            self._stub = None\n        else:\n            self._stub = elasticdl_pb2_grpc.MasterStub(channel)\n        self._max_minibatch_retry_num = max_minibatch_retry_num\n        self._model_version = -1\n        self._codec_type = codec_type\n\n    def get_task(self):\n        \"\"\"\n        get task from master\n        \"\"\"\n        req = elasticdl_pb2.GetTaskRequest()\n        req.worker_id = self._worker_id\n\n        return self._stub.GetTask(req)\n\n    def get_model(self, min_version):\n        \"\"\"\n        get model from master, and update model_version\n        \"\"\"\n        req = elasticdl_pb2.GetModelRequest()\n        req.min_version = min_version\n        model = self._stub.GetModel(req)\n\n        for var in self._model.trainable_variables:\n            # Assumes all trainable variables exist in model.param.\n            var.assign(tensor_to_ndarray(model.param[var.name]))\n        self._model_version = model.version\n\n    def report_task_result(self, task_id, err_msg):\n        \"\"\"\n        report task result to master\n        \"\"\"\n        report = elasticdl_pb2.ReportTaskResultRequest()\n        report.task_id = task_id\n        report.err_message = err_msg\n        return self._stub.ReportTaskResult(report)\n\n    def report_gradient(self, grads):\n        \"\"\"\n        report gradient to ps, return (accepted, model_version) from rpc call.\n        \"\"\"\n        req = elasticdl_pb2.ReportGradientRequest()\n        for g, v in zip(grads, self._model.trainable_variables):\n            req.gradient[v.name].CopyFrom(ndarray_to_tensor(g.numpy()))\n        req.model_version = self._model_version\n        res = self._stub.ReportGradient(req)\n        return res.accepted, res.model_version\n\n    def report_evaluation_metrics(self, evaluation_metrics):\n        \"\"\"\n        report evaluation metrics to ps, return (accepted, model_version)\n        from rpc call.\n        \"\"\"\n        req = elasticdl_pb2.ReportEvaluationMetricsRequest()\n        for k, v in evaluation_metrics.items():\n            v_np = v.numpy()\n            if v_np.size != 1:\n                raise Exception(\n                    \"Only metric result of length 1 is \" \"supported currently\"\n                )\n            req.evaluation_metrics[k].CopyFrom(ndarray_to_tensor(v_np))\n        req.model_version = self._model_version\n        res = self._stub.ReportEvaluationMetrics(req)\n        return res.accepted, res.model_version\n\n    @staticmethod\n    def _get_batch(reader, batch_size, decode):\n        res = []\n        for i in range(batch_size):\n            record = reader.record()\n            if record is None:\n                break\n            res.append(decode(record))\n        return res\n\n    def _get_features_and_labels(self, record_buf):\n        batch_input_data, batch_labels = self._input_fn(record_buf)\n        features = [\n            batch_input_data[f_col.key] for f_col in self._feature_columns\n        ]\n        if len(features) == 1:\n            features = features[0]\n        return features, batch_labels\n\n    def _run_training_task(self, features, labels):\n        with tf.GradientTape() as tape:\n            outputs = self._model.call(features, training=True)\n            loss = self._loss(outputs, labels)\n\n            # TODO:  Add regularization loss if any,\n            #        which should be divided by the\n            #        number of contributing workers.\n        grads = tape.gradient(loss, self._model.trainable_variables)\n        accepted, min_model_version = self.report_gradient(grads)\n        return accepted, min_model_version, loss\n\n    def _run_evaluation_task(self, features, labels):\n        outputs = self._model.call(features, training=False)\n        evaluation_metrics = self._eval_metrics_fn(outputs, labels)\n        return self.report_evaluation_metrics(evaluation_metrics)\n\n    def _handle_task(self, task):\n        min_model_version = task.model_version\n        with closing(\n            recordio.Scanner(\n                task.shard_file_name, task.start, task.end - task.start\n            )\n        ) as reader:\n            while True:\n                record_buf = self._get_batch(\n                    reader, task.minibatch_size, self._codec.decode\n                )\n                if not record_buf:\n                    break\n                min_model_version = self._process_minibatch(\n                    task, record_buf, min_model_version\n                )\n\n    def _process_minibatch(self, task, record_buf, min_model_version):\n        for _ in range(self._max_minibatch_retry_num):\n            # TODO: optimize the logic to avoid unnecessary\n            #       get_model call.\n            self.get_model(max(self._model_version, min_model_version))\n            features, labels = self._get_features_and_labels(record_buf)\n            if task.type == elasticdl_pb2.EVALUATION:\n                accepted, min_model_version = self._run_evaluation_task(\n                    features, labels\n                )\n                if accepted:\n                    break\n            elif task.type == elasticdl_pb2.TRAINING:\n                accepted, min_model_version, loss = self._run_training_task(\n                    features, labels\n                )\n                if accepted:\n                    self._logger.info(\"Loss is %f\" % loss.numpy())\n                    break\n            else:\n                raise RuntimeError(\"Unrecognized task type, %s\" % task.type)\n        else:\n            # Worker got stuck, fail the task.\n            # TODO: stop the worker if it fails to make any\n            #       progress for some time.\n            raise RuntimeError(\"Worker got stuck\")\n        return min_model_version\n\n    def run(self):\n        \"\"\"\n        Fetches task from master and performs training or evaluation.\n        \"\"\"\n        while True:\n            task = self.get_task()\n            if not task.shard_file_name:\n                # No more task\n                break\n            err_msg = \"\"\n            try:\n                self._handle_task(task)\n            except Exception as ex:\n                err_msg = str(ex)\n                traceback.print_exc()\n            self.report_task_result(task.task_id, err_msg)\n", "path": "elasticdl/python/elasticdl/worker/worker.py"}]}