prompt: We are currently solving the following issue within our repository. Here is the issue text:
  --- BEGIN ISSUE ---
  Implement additional proto definitions and service for reporting evaluation metrics
  This is part of https://github.com/wangkuiyi/elasticdl/issues/384. Implement `MasterServicer.ReportEvaluationMetrics()` and additional proto definitions such as `ReportEvaluationMetricsReply` and `ReportEvaluationMetricsRequest`.
  Implement additional proto definitions and service for reporting evaluation metrics
  This is part of https://github.com/wangkuiyi/elasticdl/issues/384. Implement `MasterServicer.ReportEvaluationMetrics()` and additional proto definitions such as `ReportEvaluationMetricsReply` and `ReportEvaluationMetricsRequest`.
  
  --- 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/master/servicer.py`
  Content:
  ```
  1 import threading
  2 import numpy as np
  3 
  4 import tensorflow as tf
  5 
  6 assert tf.executing_eagerly()
  7 
  8 from google.protobuf import empty_pb2
  9 
  10 from elasticdl.proto import elasticdl_pb2
  11 from elasticdl.proto import elasticdl_pb2_grpc
  12 from elasticdl.python.elasticdl.common.ndarray import ndarray_to_tensor, tensor_to_ndarray
  13 
  14 
  15 class MasterServicer(elasticdl_pb2_grpc.MasterServicer):
  16     """Master service implementation"""
  17 
  18     def __init__(
  19         self,
  20         logger,
  21         grads_to_wait,
  22         minibatch_size,
  23         optimizer,
  24         task_q,
  25         *,
  26         init_var=[]
  27     ):
  28         # TODO: group params together into a single object.
  29         self.logger = logger
  30         self._opt = optimizer
  31         self._task_q = task_q
  32         self._lock = threading.Lock()
  33         # A <string, tf.ResourceVariable> map. We use tf.ResourceVariable
  34         # instead ndarray to avoid copying and conversion when calling
  35         # optimizer's apply_gradients() function.
  36         self._model = {}
  37         self._version = 0
  38         self._gradient_sum = {}
  39         self._grad_to_wait = grads_to_wait
  40         self._grad_n = 0
  41         self._minibatch_size = minibatch_size
  42         for var in init_var:
  43             self.set_model_var(var.name, var.numpy())
  44 
  45     def set_model_var(self, name, value):
  46         """Add or set model variable. Value should be a float32 ndarray"""
  47         if value.dtype != np.float32:
  48             raise ValueError("Value should be a float32 numpy array")
  49         self._model[name] = tf.Variable(
  50             value, name=MasterServicer.var_name_encode(name)
  51         )
  52 
  53     @staticmethod
  54     def var_name_encode(name):
  55         return name.replace(":", "-")
  56 
  57     def GetTask(self, request, _):
  58         res = elasticdl_pb2.Task()
  59         res.model_version = self._version
  60         res.minibatch_size = self._minibatch_size
  61         task_id, task = self._task_q.get(request.worker_id)
  62         if task:
  63             res.task_id = task_id
  64             res.shard_file_name = task.file_name
  65             res.start = task.start
  66             res.end = task.end
  67         return res
  68 
  69     def GetModel(self, request, _):
  70         if request.min_version > self._version:
  71             err_msg = (
  72                 "Requested version %d not available yet, current version: %d"
  73                 % (request.min_version, self._version)
  74             )
  75             self.logger.warning(err_msg)
  76             raise ValueError(err_msg)
  77 
  78         res = elasticdl_pb2.Model()
  79         with self._lock:
  80             res.version = self._version
  81             for k, v in self._model.items():
  82                 res.param[k].CopyFrom(ndarray_to_tensor(v.numpy()))
  83         return res
  84 
  85     def _update_model(self):
  86         assert self._lock.locked()
  87         grad_var = []
  88         for k in self._gradient_sum:
  89             self._gradient_sum[k] = self._gradient_sum[k] / self._grad_to_wait
  90             grad_var.append((self._gradient_sum[k], self._model[k]))
  91         self._opt.apply_gradients(grad_var)
  92         self._version += 1
  93         self._gradient_sum.clear()
  94         self._grad_n = 0
  95 
  96     def ReportGradient(self, request, _):
  97         if request.model_version > self._version:
  98             err_msg = "Model version %d out of range, current version: %d" % (
  99                 request.model_version,
  100                 self._version,
  101             )
  102             self.logger.warning(err_msg)
  103             raise ValueError(err_msg)
  104 
  105         res = elasticdl_pb2.ReportGradientReply()
  106         if request.model_version < self._version:
  107             self.logger.warning(
  108                 "Task result for outdated version %d dropped",
  109                 request.model_version,
  110             )
  111             res.accepted = False
  112             res.model_version = self._version
  113             return res
  114 
  115         # TODO: Update task queue with task_id
  116         with self._lock:
  117             tmp = {}
  118             # Do sanity check before accumulating gradients.
  119             for k, v in request.gradient.items():
  120                 if k not in self._model:
  121                     raise ValueError(
  122                         "Gradient key: %s is not part of model", k
  123                     )
  124                 arr = tensor_to_ndarray(v)
  125                 if arr.shape != self._model[k].numpy().shape:
  126                     raise ValueError(
  127                         "Gradient key: %s has incompatible dimension", k
  128                     )
  129                 tmp[k] = arr
  130 
  131             for k, v in tmp.items():
  132                 if k in self._gradient_sum:
  133                     self._gradient_sum[k] = self._gradient_sum[k] + v
  134                 else:
  135                     self._gradient_sum[k] = v
  136 
  137             self._grad_n += 1
  138             if self._grad_n >= self._grad_to_wait:
  139                 self._update_model()
  140         res.accepted = True
  141         res.model_version = self._version
  142         return res
  143 
  144     def ReportTaskResult(self, request, _):
  145         if request.err_message:
  146             self.logger.warning(
  147                 "Worker reported error: " + request.err_message
  148             )
  149             self._task_q.report(request.task_id, False)
  150         else:
  151             self._task_q.report(request.task_id, True)
  152         return empty_pb2.Empty()
  153
  ```
  --- 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()
  
  ```
  
golden_diff: diff --git a/elasticdl/python/elasticdl/master/servicer.py b/elasticdl/python/elasticdl/master/servicer.py
  --- a/elasticdl/python/elasticdl/master/servicer.py
  +++ b/elasticdl/python/elasticdl/master/servicer.py
  @@ -39,6 +39,7 @@
           self._grad_to_wait = grads_to_wait
           self._grad_n = 0
           self._minibatch_size = minibatch_size
  +        self._evaluation_metrics = {}
           for var in init_var:
               self.set_model_var(var.name, var.numpy())
   
  @@ -67,13 +68,7 @@
           return res
   
       def GetModel(self, request, _):
  -        if request.min_version > self._version:
  -            err_msg = (
  -                "Requested version %d not available yet, current version: %d"
  -                % (request.min_version, self._version)
  -            )
  -            self.logger.warning(err_msg)
  -            raise ValueError(err_msg)
  +        _ = self._validate_model_version(request.min_version)
   
           res = elasticdl_pb2.Model()
           with self._lock:
  @@ -93,21 +88,28 @@
           self._gradient_sum.clear()
           self._grad_n = 0
   
  -    def ReportGradient(self, request, _):
  -        if request.model_version > self._version:
  -            err_msg = "Model version %d out of range, current version: %d" % (
  -                request.model_version,
  +    def _validate_model_version(self, request_model_version):
  +        if request_model_version > self._version:
  +            err_msg = "Model version %d not available yet, current version: %d" % (
  +                request_model_version,
                   self._version,
               )
               self.logger.warning(err_msg)
               raise ValueError(err_msg)
   
  -        res = elasticdl_pb2.ReportGradientReply()
  -        if request.model_version < self._version:
  +        invalid_model_version = request_model_version < self._version
  +        if invalid_model_version:
               self.logger.warning(
                   "Task result for outdated version %d dropped",
  -                request.model_version,
  +                request_model_version,
               )
  +        return invalid_model_version
  +
  +    def ReportGradient(self, request, _):
  +        invalid_model_version = self._validate_model_version(request.model_version)
  +
  +        res = elasticdl_pb2.ReportGradientResponse()
  +        if invalid_model_version:
               res.accepted = False
               res.model_version = self._version
               return res
  @@ -150,3 +152,22 @@
           else:
               self._task_q.report(request.task_id, True)
           return empty_pb2.Empty()
  +
  +    def ReportEvaluationMetrics(self, request, _):
  +        invalid_model_version = self._validate_model_version(request.model_version)
  +
  +        res = elasticdl_pb2.ReportEvaluationMetricsResponse()
  +        if invalid_model_version:
  +            res.accepted = False
  +            res.model_version = self._version
  +            return res
  +
  +        with self._lock:
  +            for k, v in request.evaluation_metrics.items():
  +                arr = tensor_to_ndarray(v)
  +                self._evaluation_metrics[k] = arr
  +
  +            self._update_model()
  +        res.accepted = True
  +        res.model_version = self._version
  +        return res
  
verification_info: {"golden_diff": "diff --git a/elasticdl/python/elasticdl/master/servicer.py b/elasticdl/python/elasticdl/master/servicer.py\n--- a/elasticdl/python/elasticdl/master/servicer.py\n+++ b/elasticdl/python/elasticdl/master/servicer.py\n@@ -39,6 +39,7 @@\n         self._grad_to_wait = grads_to_wait\n         self._grad_n = 0\n         self._minibatch_size = minibatch_size\n+        self._evaluation_metrics = {}\n         for var in init_var:\n             self.set_model_var(var.name, var.numpy())\n \n@@ -67,13 +68,7 @@\n         return res\n \n     def GetModel(self, request, _):\n-        if request.min_version > self._version:\n-            err_msg = (\n-                \"Requested version %d not available yet, current version: %d\"\n-                % (request.min_version, self._version)\n-            )\n-            self.logger.warning(err_msg)\n-            raise ValueError(err_msg)\n+        _ = self._validate_model_version(request.min_version)\n \n         res = elasticdl_pb2.Model()\n         with self._lock:\n@@ -93,21 +88,28 @@\n         self._gradient_sum.clear()\n         self._grad_n = 0\n \n-    def ReportGradient(self, request, _):\n-        if request.model_version > self._version:\n-            err_msg = \"Model version %d out of range, current version: %d\" % (\n-                request.model_version,\n+    def _validate_model_version(self, request_model_version):\n+        if request_model_version > self._version:\n+            err_msg = \"Model version %d not available yet, current version: %d\" % (\n+                request_model_version,\n                 self._version,\n             )\n             self.logger.warning(err_msg)\n             raise ValueError(err_msg)\n \n-        res = elasticdl_pb2.ReportGradientReply()\n-        if request.model_version < self._version:\n+        invalid_model_version = request_model_version < self._version\n+        if invalid_model_version:\n             self.logger.warning(\n                 \"Task result for outdated version %d dropped\",\n-                request.model_version,\n+                request_model_version,\n             )\n+        return invalid_model_version\n+\n+    def ReportGradient(self, request, _):\n+        invalid_model_version = self._validate_model_version(request.model_version)\n+\n+        res = elasticdl_pb2.ReportGradientResponse()\n+        if invalid_model_version:\n             res.accepted = False\n             res.model_version = self._version\n             return res\n@@ -150,3 +152,22 @@\n         else:\n             self._task_q.report(request.task_id, True)\n         return empty_pb2.Empty()\n+\n+    def ReportEvaluationMetrics(self, request, _):\n+        invalid_model_version = self._validate_model_version(request.model_version)\n+\n+        res = elasticdl_pb2.ReportEvaluationMetricsResponse()\n+        if invalid_model_version:\n+            res.accepted = False\n+            res.model_version = self._version\n+            return res\n+\n+        with self._lock:\n+            for k, v in request.evaluation_metrics.items():\n+                arr = tensor_to_ndarray(v)\n+                self._evaluation_metrics[k] = arr\n+\n+            self._update_model()\n+        res.accepted = True\n+        res.model_version = self._version\n+        return res\n", "issue": "Implement additional proto definitions and service for reporting evaluation metrics\nThis is part of https://github.com/wangkuiyi/elasticdl/issues/384. Implement `MasterServicer.ReportEvaluationMetrics()` and additional proto definitions such as `ReportEvaluationMetricsReply` and `ReportEvaluationMetricsRequest`.\nImplement additional proto definitions and service for reporting evaluation metrics\nThis is part of https://github.com/wangkuiyi/elasticdl/issues/384. Implement `MasterServicer.ReportEvaluationMetrics()` and additional proto definitions such as `ReportEvaluationMetricsReply` and `ReportEvaluationMetricsRequest`.\n", "before_files": [{"content": "import threading\nimport numpy as np\n\nimport tensorflow as tf\n\nassert tf.executing_eagerly()\n\nfrom google.protobuf import empty_pb2\n\nfrom elasticdl.proto import elasticdl_pb2\nfrom elasticdl.proto import elasticdl_pb2_grpc\nfrom elasticdl.python.elasticdl.common.ndarray import ndarray_to_tensor, tensor_to_ndarray\n\n\nclass MasterServicer(elasticdl_pb2_grpc.MasterServicer):\n    \"\"\"Master service implementation\"\"\"\n\n    def __init__(\n        self,\n        logger,\n        grads_to_wait,\n        minibatch_size,\n        optimizer,\n        task_q,\n        *,\n        init_var=[]\n    ):\n        # TODO: group params together into a single object.\n        self.logger = logger\n        self._opt = optimizer\n        self._task_q = task_q\n        self._lock = threading.Lock()\n        # A <string, tf.ResourceVariable> map. We use tf.ResourceVariable\n        # instead ndarray to avoid copying and conversion when calling\n        # optimizer's apply_gradients() function.\n        self._model = {}\n        self._version = 0\n        self._gradient_sum = {}\n        self._grad_to_wait = grads_to_wait\n        self._grad_n = 0\n        self._minibatch_size = minibatch_size\n        for var in init_var:\n            self.set_model_var(var.name, var.numpy())\n\n    def set_model_var(self, name, value):\n        \"\"\"Add or set model variable. Value should be a float32 ndarray\"\"\"\n        if value.dtype != np.float32:\n            raise ValueError(\"Value should be a float32 numpy array\")\n        self._model[name] = tf.Variable(\n            value, name=MasterServicer.var_name_encode(name)\n        )\n\n    @staticmethod\n    def var_name_encode(name):\n        return name.replace(\":\", \"-\")\n\n    def GetTask(self, request, _):\n        res = elasticdl_pb2.Task()\n        res.model_version = self._version\n        res.minibatch_size = self._minibatch_size\n        task_id, task = self._task_q.get(request.worker_id)\n        if task:\n            res.task_id = task_id\n            res.shard_file_name = task.file_name\n            res.start = task.start\n            res.end = task.end\n        return res\n\n    def GetModel(self, request, _):\n        if request.min_version > self._version:\n            err_msg = (\n                \"Requested version %d not available yet, current version: %d\"\n                % (request.min_version, self._version)\n            )\n            self.logger.warning(err_msg)\n            raise ValueError(err_msg)\n\n        res = elasticdl_pb2.Model()\n        with self._lock:\n            res.version = self._version\n            for k, v in self._model.items():\n                res.param[k].CopyFrom(ndarray_to_tensor(v.numpy()))\n        return res\n\n    def _update_model(self):\n        assert self._lock.locked()\n        grad_var = []\n        for k in self._gradient_sum:\n            self._gradient_sum[k] = self._gradient_sum[k] / self._grad_to_wait\n            grad_var.append((self._gradient_sum[k], self._model[k]))\n        self._opt.apply_gradients(grad_var)\n        self._version += 1\n        self._gradient_sum.clear()\n        self._grad_n = 0\n\n    def ReportGradient(self, request, _):\n        if request.model_version > self._version:\n            err_msg = \"Model version %d out of range, current version: %d\" % (\n                request.model_version,\n                self._version,\n            )\n            self.logger.warning(err_msg)\n            raise ValueError(err_msg)\n\n        res = elasticdl_pb2.ReportGradientReply()\n        if request.model_version < self._version:\n            self.logger.warning(\n                \"Task result for outdated version %d dropped\",\n                request.model_version,\n            )\n            res.accepted = False\n            res.model_version = self._version\n            return res\n\n        # TODO: Update task queue with task_id\n        with self._lock:\n            tmp = {}\n            # Do sanity check before accumulating gradients.\n            for k, v in request.gradient.items():\n                if k not in self._model:\n                    raise ValueError(\n                        \"Gradient key: %s is not part of model\", k\n                    )\n                arr = tensor_to_ndarray(v)\n                if arr.shape != self._model[k].numpy().shape:\n                    raise ValueError(\n                        \"Gradient key: %s has incompatible dimension\", k\n                    )\n                tmp[k] = arr\n\n            for k, v in tmp.items():\n                if k in self._gradient_sum:\n                    self._gradient_sum[k] = self._gradient_sum[k] + v\n                else:\n                    self._gradient_sum[k] = v\n\n            self._grad_n += 1\n            if self._grad_n >= self._grad_to_wait:\n                self._update_model()\n        res.accepted = True\n        res.model_version = self._version\n        return res\n\n    def ReportTaskResult(self, request, _):\n        if request.err_message:\n            self.logger.warning(\n                \"Worker reported error: \" + request.err_message\n            )\n            self._task_q.report(request.task_id, False)\n        else:\n            self._task_q.report(request.task_id, True)\n        return empty_pb2.Empty()\n", "path": "elasticdl/python/elasticdl/master/servicer.py"}], "after_files": [{"content": "import threading\nimport numpy as np\n\nimport tensorflow as tf\n\nassert tf.executing_eagerly()\n\nfrom google.protobuf import empty_pb2\n\nfrom elasticdl.proto import elasticdl_pb2\nfrom elasticdl.proto import elasticdl_pb2_grpc\nfrom elasticdl.python.elasticdl.common.ndarray import ndarray_to_tensor, tensor_to_ndarray\n\n\nclass MasterServicer(elasticdl_pb2_grpc.MasterServicer):\n    \"\"\"Master service implementation\"\"\"\n\n    def __init__(\n        self,\n        logger,\n        grads_to_wait,\n        minibatch_size,\n        optimizer,\n        task_q,\n        *,\n        init_var=[]\n    ):\n        # TODO: group params together into a single object.\n        self.logger = logger\n        self._opt = optimizer\n        self._task_q = task_q\n        self._lock = threading.Lock()\n        # A <string, tf.ResourceVariable> map. We use tf.ResourceVariable\n        # instead ndarray to avoid copying and conversion when calling\n        # optimizer's apply_gradients() function.\n        self._model = {}\n        self._version = 0\n        self._gradient_sum = {}\n        self._grad_to_wait = grads_to_wait\n        self._grad_n = 0\n        self._minibatch_size = minibatch_size\n        self._evaluation_metrics = {}\n        for var in init_var:\n            self.set_model_var(var.name, var.numpy())\n\n    def set_model_var(self, name, value):\n        \"\"\"Add or set model variable. Value should be a float32 ndarray\"\"\"\n        if value.dtype != np.float32:\n            raise ValueError(\"Value should be a float32 numpy array\")\n        self._model[name] = tf.Variable(\n            value, name=MasterServicer.var_name_encode(name)\n        )\n\n    @staticmethod\n    def var_name_encode(name):\n        return name.replace(\":\", \"-\")\n\n    def GetTask(self, request, _):\n        res = elasticdl_pb2.Task()\n        res.model_version = self._version\n        res.minibatch_size = self._minibatch_size\n        task_id, task = self._task_q.get(request.worker_id)\n        if task:\n            res.task_id = task_id\n            res.shard_file_name = task.file_name\n            res.start = task.start\n            res.end = task.end\n        return res\n\n    def GetModel(self, request, _):\n        _ = self._validate_model_version(request.min_version)\n\n        res = elasticdl_pb2.Model()\n        with self._lock:\n            res.version = self._version\n            for k, v in self._model.items():\n                res.param[k].CopyFrom(ndarray_to_tensor(v.numpy()))\n        return res\n\n    def _update_model(self):\n        assert self._lock.locked()\n        grad_var = []\n        for k in self._gradient_sum:\n            self._gradient_sum[k] = self._gradient_sum[k] / self._grad_to_wait\n            grad_var.append((self._gradient_sum[k], self._model[k]))\n        self._opt.apply_gradients(grad_var)\n        self._version += 1\n        self._gradient_sum.clear()\n        self._grad_n = 0\n\n    def _validate_model_version(self, request_model_version):\n        if request_model_version > self._version:\n            err_msg = \"Model version %d not available yet, current version: %d\" % (\n                request_model_version,\n                self._version,\n            )\n            self.logger.warning(err_msg)\n            raise ValueError(err_msg)\n\n        invalid_model_version = request_model_version < self._version\n        if invalid_model_version:\n            self.logger.warning(\n                \"Task result for outdated version %d dropped\",\n                request_model_version,\n            )\n        return invalid_model_version\n\n    def ReportGradient(self, request, _):\n        invalid_model_version = self._validate_model_version(request.model_version)\n\n        res = elasticdl_pb2.ReportGradientResponse()\n        if invalid_model_version:\n            res.accepted = False\n            res.model_version = self._version\n            return res\n\n        # TODO: Update task queue with task_id\n        with self._lock:\n            tmp = {}\n            # Do sanity check before accumulating gradients.\n            for k, v in request.gradient.items():\n                if k not in self._model:\n                    raise ValueError(\n                        \"Gradient key: %s is not part of model\", k\n                    )\n                arr = tensor_to_ndarray(v)\n                if arr.shape != self._model[k].numpy().shape:\n                    raise ValueError(\n                        \"Gradient key: %s has incompatible dimension\", k\n                    )\n                tmp[k] = arr\n\n            for k, v in tmp.items():\n                if k in self._gradient_sum:\n                    self._gradient_sum[k] = self._gradient_sum[k] + v\n                else:\n                    self._gradient_sum[k] = v\n\n            self._grad_n += 1\n            if self._grad_n >= self._grad_to_wait:\n                self._update_model()\n        res.accepted = True\n        res.model_version = self._version\n        return res\n\n    def ReportTaskResult(self, request, _):\n        if request.err_message:\n            self.logger.warning(\n                \"Worker reported error: \" + request.err_message\n            )\n            self._task_q.report(request.task_id, False)\n        else:\n            self._task_q.report(request.task_id, True)\n        return empty_pb2.Empty()\n\n    def ReportEvaluationMetrics(self, request, _):\n        invalid_model_version = self._validate_model_version(request.model_version)\n\n        res = elasticdl_pb2.ReportEvaluationMetricsResponse()\n        if invalid_model_version:\n            res.accepted = False\n            res.model_version = self._version\n            return res\n\n        with self._lock:\n            for k, v in request.evaluation_metrics.items():\n                arr = tensor_to_ndarray(v)\n                self._evaluation_metrics[k] = arr\n\n            self._update_model()\n        res.accepted = True\n        res.model_version = self._version\n        return res\n", "path": "elasticdl/python/elasticdl/master/servicer.py"}]}