We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
The status of the worker pod should be Error when there is any error happened.
The status of the worker pod is still Running and will never stop even if there are some errors in function `handle_task` like:
```bash
Traceback (most recent call last):
  File "/elasticdl/python/elasticdl/worker/worker.py", line 236, in run
    self._handle_task(task)
  File "/elasticdl/python/elasticdl/worker/worker.py", line 190, in _handle_task
    task, record_buf, min_model_version
  File "/elasticdl/python/elasticdl/worker/worker.py", line 211, in _process_minibatch
    features, labels
  File "/elasticdl/python/elasticdl/worker/worker.py", line 168, in _run_training_task
    accepted, min_model_version = self.report_gradient(grads)
  File "/elasticdl/python/elasticdl/worker/worker.py", line 115, in report_gradient
    res = self._stub.ReportGradient(req)
  File "/usr/local/lib/python3.6/dist-packages/grpc/_channel.py", line 565, in __call__
    return _end_unary_response_blocking(state, call, False, None)
  File "/usr/local/lib/python3.6/dist-packages/grpc/_channel.py", line 467, in _end_unary_response_blocking
    raise _Rendezvous(state, None, None, deadline)

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

{"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@@ -234,7 +234,11 @@\n             err_msg = \"\"\n             try:\n                 self._handle_task(task)\n+            except RuntimeError as err:\n+                err_msg = str(err)\n+                traceback.print_exc()\n             except Exception as ex:\n                 err_msg = str(ex)\n                 traceback.print_exc()\n+                raise ex\n             self.report_task_result(task.task_id, err_msg)\n", "issue": "The status of the worker pod should be Error when there is any error happened.\nThe status of the worker pod is still Running and will never stop even if there are some errors in function `handle_task` like:\r\n```bash\r\nTraceback (most recent call last):\r\n  File \"/elasticdl/python/elasticdl/worker/worker.py\", line 236, in run\r\n    self._handle_task(task)\r\n  File \"/elasticdl/python/elasticdl/worker/worker.py\", line 190, in _handle_task\r\n    task, record_buf, min_model_version\r\n  File \"/elasticdl/python/elasticdl/worker/worker.py\", line 211, in _process_minibatch\r\n    features, labels\r\n  File \"/elasticdl/python/elasticdl/worker/worker.py\", line 168, in _run_training_task\r\n    accepted, min_model_version = self.report_gradient(grads)\r\n  File \"/elasticdl/python/elasticdl/worker/worker.py\", line 115, in report_gradient\r\n    res = self._stub.ReportGradient(req)\r\n  File \"/usr/local/lib/python3.6/dist-packages/grpc/_channel.py\", line 565, in __call__\r\n    return _end_unary_response_blocking(state, call, False, None)\r\n  File \"/usr/local/lib/python3.6/dist-packages/grpc/_channel.py\", line 467, in _end_unary_response_blocking\r\n    raise _Rendezvous(state, None, None, deadline)\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 load_module\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_file=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_module(model_file)\n        self._model = model_module.model\n        self._feature_columns = model_module.feature_columns()\n        self._var_created = self._model.built\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\n        # Initilize codec\n        codec_module = load_module(codec_file)\n        codec_module.codec.init(all_columns)\n        self._codec = codec_module.codec\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\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, version, method):\n        \"\"\"\n        get model from master, and update model_version\n        \"\"\"\n        req = elasticdl_pb2.GetModelRequest()\n        req.version = version\n        req.method = method\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_variable(self):\n        \"\"\"\n        report variable to ps.\n        \"\"\"\n        req = elasticdl_pb2.ReportVariableRequest()\n        for v in self._model.trainable_variables:\n            req.variable[v.name].CopyFrom(ndarray_to_tensor(v.numpy()))\n        self._stub.ReportVariable(req)\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 scalar, convert to numpy 1D array with size 1\n            if not v_np.shape:\n                v_np = v_np.reshape(1)\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 _create_variable_and_report(self, features):\n        # Use model.call to create variables, then report to ps\n        _ = self._model.call(features)\n        self.report_variable()\n        self._var_created = True\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        features, labels = self._get_features_and_labels(record_buf)\n        if not self._var_created:\n            self._create_variable_and_report(features)\n        for _ in range(self._max_minibatch_retry_num):\n            if task.type == elasticdl_pb2.EVALUATION:\n                self.get_model(min_model_version, elasticdl_pb2.FIXED)\n                accepted, _ = self._run_evaluation_task(features, labels)\n                if accepted:\n                    break\n            elif task.type == elasticdl_pb2.TRAINING:\n                # TODO: optimize the logic to avoid unnecessary\n                #       get_model call.\n                self.get_model(\n                    max(self._model_version, min_model_version),\n                    elasticdl_pb2.MINIMUM,\n                )\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"}], "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 load_module\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_file=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_module(model_file)\n        self._model = model_module.model\n        self._feature_columns = model_module.feature_columns()\n        self._var_created = self._model.built\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\n        # Initilize codec\n        codec_module = load_module(codec_file)\n        codec_module.codec.init(all_columns)\n        self._codec = codec_module.codec\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\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, version, method):\n        \"\"\"\n        get model from master, and update model_version\n        \"\"\"\n        req = elasticdl_pb2.GetModelRequest()\n        req.version = version\n        req.method = method\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_variable(self):\n        \"\"\"\n        report variable to ps.\n        \"\"\"\n        req = elasticdl_pb2.ReportVariableRequest()\n        for v in self._model.trainable_variables:\n            req.variable[v.name].CopyFrom(ndarray_to_tensor(v.numpy()))\n        self._stub.ReportVariable(req)\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 scalar, convert to numpy 1D array with size 1\n            if not v_np.shape:\n                v_np = v_np.reshape(1)\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 _create_variable_and_report(self, features):\n        # Use model.call to create variables, then report to ps\n        _ = self._model.call(features)\n        self.report_variable()\n        self._var_created = True\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        features, labels = self._get_features_and_labels(record_buf)\n        if not self._var_created:\n            self._create_variable_and_report(features)\n        for _ in range(self._max_minibatch_retry_num):\n            if task.type == elasticdl_pb2.EVALUATION:\n                self.get_model(min_model_version, elasticdl_pb2.FIXED)\n                accepted, _ = self._run_evaluation_task(features, labels)\n                if accepted:\n                    break\n            elif task.type == elasticdl_pb2.TRAINING:\n                # TODO: optimize the logic to avoid unnecessary\n                #       get_model call.\n                self.get_model(\n                    max(self._model_version, min_model_version),\n                    elasticdl_pb2.MINIMUM,\n                )\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 RuntimeError as err:\n                err_msg = str(err)\n                traceback.print_exc()\n            except Exception as ex:\n                err_msg = str(ex)\n                traceback.print_exc()\n                raise ex\n            self.report_task_result(task.task_id, err_msg)\n", "path": "elasticdl/python/elasticdl/worker/worker.py"}]}