prompt: We are currently solving the following issue within our repository. Here is the issue text:
  --- BEGIN ISSUE ---
  [master] PS model update 
  
  
  --- 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/master/servicer.py`
  Content:
  ```
  1 import threading
  2 import numpy as np
  3 
  4 import tensorflow as tf
  5 from proto import master_pb2
  6 from proto import master_pb2_grpc
  7 from util.ndarray import ndarray_to_tensor, tensor_to_ndarray
  8 
  9 
  10 class MasterServicer(master_pb2_grpc.MasterServicer):
  11     """Master service implementation"""
  12 
  13     def __init__(self, logger, grads_to_wait):
  14         self.logger = logger
  15         self._lock = threading.Lock()
  16         # TODO: random initialization
  17         # A <string, tf.ResourceVariable> map. We use tf.ResourceVariable
  18         # instead ndarray to avoid copying and conversion when calling
  19         # optimizer's apply_gradients() function.
  20         self._model = {}
  21         self._version = 0
  22         self._gradient_sum = {}
  23         self._grad_to_wait = grads_to_wait
  24         self._grad_n = 0
  25 
  26     def _set_model_var(self, name, value):
  27         """Add or set model variable. Value should be a float32 ndarray"""
  28         if value.dtype != np.float32:
  29             raise ValueError("Value should be a float32 numpy array")
  30         self._model[name] = tf.Variable(value, name=name, use_resource=True)
  31 
  32     def GetTask(self, request, context):
  33         # TODO: implent task queues. Return an empty task for now.
  34         res = master_pb2.Task()
  35         res.shard_file_name = ""
  36         res.model_version = self._version
  37         return res
  38 
  39     def GetModel(self, request, context):
  40         if request.min_version > self._version:
  41             err_msg = (
  42                 "Requested version %d not available yet, current version: %d"
  43                 % (request.min_version, self._version)
  44             )
  45             self.logger.warning(err_msg)
  46             raise ValueError(err_msg)
  47 
  48         res = master_pb2.Model()
  49         with self._lock:
  50             res.version = self._version
  51             for k, v in self._model.items():
  52                 res.param[k].CopyFrom(ndarray_to_tensor(v.numpy()))
  53         return res
  54 
  55     def ReportTaskResult(self, request, context):
  56         if request.model_version > self._version:
  57             err_msg = "Model version %d out of range, current version: %d" % (
  58                 request.model_version,
  59                 self._version,
  60             )
  61             self.logger.warning(err_msg)
  62             raise ValueError(err_msg)
  63 
  64         res = master_pb2.ReportTaskResultReply()
  65         if request.model_version < self._version:
  66             self.logger.warning(
  67                 "Task result for outdated version %d dropped",
  68                 request.model_version,
  69             )
  70             res.accepted = False
  71             res.model_version = self._version
  72             return res
  73 
  74         if request.err_message:
  75             self.logger.warning("Worker error: %s" % request.err_message)
  76             res.accepted = False
  77             res.model_version = self._version
  78             return res
  79 
  80         # TODO: Update task queue with task_id
  81         with self._lock:
  82             tmp = {}
  83             # Do sanity check before accumulating gradients.
  84             for k, v in request.gradient.items():
  85                 if k not in self._model:
  86                     raise ValueError(
  87                         "Gradient key: %s is not part of model", k
  88                     )
  89                 arr = tensor_to_ndarray(v)
  90                 if arr.shape != self._model[k].numpy().shape:
  91                     raise ValueError(
  92                         "Gradient key: %s has incompatible dimension", k
  93                     )
  94                 tmp[k] = arr
  95 
  96             for k, v in tmp.items():
  97                 if k in self._gradient_sum:
  98                     self._gradient_sum[k] = self._gradient_sum[k] + v
  99                 else:
  100                     self._gradient_sum[k] = v
  101 
  102             self._grad_n += 1
  103             if self._grad_n >= self._grad_to_wait:
  104                 # TODO: update model
  105                 self._version += 1
  106                 self._gradient_sum.clear()
  107                 self._grad_n = 0
  108         res.accepted = True
  109         res.model_version = self._version
  110         return res
  111
  ```
  --- 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/master/servicer.py b/elasticdl/master/servicer.py
  --- a/elasticdl/master/servicer.py
  +++ b/elasticdl/master/servicer.py
  @@ -2,6 +2,8 @@
   import numpy as np
   
   import tensorflow as tf
  +assert tf.executing_eagerly()
  +
   from proto import master_pb2
   from proto import master_pb2_grpc
   from util.ndarray import ndarray_to_tensor, tensor_to_ndarray
  @@ -10,8 +12,9 @@
   class MasterServicer(master_pb2_grpc.MasterServicer):
       """Master service implementation"""
   
  -    def __init__(self, logger, grads_to_wait):
  +    def __init__(self, logger, grads_to_wait, optimizer):
           self.logger = logger
  +        self._opt = optimizer
           self._lock = threading.Lock()
           # TODO: random initialization
           # A <string, tf.ResourceVariable> map. We use tf.ResourceVariable
  @@ -52,6 +55,17 @@
                   res.param[k].CopyFrom(ndarray_to_tensor(v.numpy()))
           return res
   
  +    def _update_model(self):
  +        assert self._lock.locked()
  +        grad_var = []
  +        for k in self._gradient_sum:
  +            self._gradient_sum[k] = self._gradient_sum[k] / self._grad_to_wait
  +            grad_var.append((self._gradient_sum[k], self._model[k]))
  +        self._opt.apply_gradients(grad_var)
  +        self._version += 1
  +        self._gradient_sum.clear()
  +        self._grad_n = 0
  +
       def ReportTaskResult(self, request, context):
           if request.model_version > self._version:
               err_msg = "Model version %d out of range, current version: %d" % (
  @@ -101,10 +115,7 @@
   
               self._grad_n += 1
               if self._grad_n >= self._grad_to_wait:
  -                # TODO: update model
  -                self._version += 1
  -                self._gradient_sum.clear()
  -                self._grad_n = 0
  +                self._update_model()
           res.accepted = True
           res.model_version = self._version
           return res
  
verification_info: {"golden_diff": "diff --git a/elasticdl/master/servicer.py b/elasticdl/master/servicer.py\n--- a/elasticdl/master/servicer.py\n+++ b/elasticdl/master/servicer.py\n@@ -2,6 +2,8 @@\n import numpy as np\n \n import tensorflow as tf\n+assert tf.executing_eagerly()\n+\n from proto import master_pb2\n from proto import master_pb2_grpc\n from util.ndarray import ndarray_to_tensor, tensor_to_ndarray\n@@ -10,8 +12,9 @@\n class MasterServicer(master_pb2_grpc.MasterServicer):\n     \"\"\"Master service implementation\"\"\"\n \n-    def __init__(self, logger, grads_to_wait):\n+    def __init__(self, logger, grads_to_wait, optimizer):\n         self.logger = logger\n+        self._opt = optimizer\n         self._lock = threading.Lock()\n         # TODO: random initialization\n         # A <string, tf.ResourceVariable> map. We use tf.ResourceVariable\n@@ -52,6 +55,17 @@\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 ReportTaskResult(self, request, context):\n         if request.model_version > self._version:\n             err_msg = \"Model version %d out of range, current version: %d\" % (\n@@ -101,10 +115,7 @@\n \n             self._grad_n += 1\n             if self._grad_n >= self._grad_to_wait:\n-                # TODO: update model\n-                self._version += 1\n-                self._gradient_sum.clear()\n-                self._grad_n = 0\n+                self._update_model()\n         res.accepted = True\n         res.model_version = self._version\n         return res\n", "issue": "[master] PS model update \n\n", "before_files": [{"content": "import threading\nimport numpy as np\n\nimport tensorflow as tf\nfrom proto import master_pb2\nfrom proto import master_pb2_grpc\nfrom util.ndarray import ndarray_to_tensor, tensor_to_ndarray\n\n\nclass MasterServicer(master_pb2_grpc.MasterServicer):\n    \"\"\"Master service implementation\"\"\"\n\n    def __init__(self, logger, grads_to_wait):\n        self.logger = logger\n        self._lock = threading.Lock()\n        # TODO: random initialization\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\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(value, name=name, use_resource=True)\n\n    def GetTask(self, request, context):\n        # TODO: implent task queues. Return an empty task for now.\n        res = master_pb2.Task()\n        res.shard_file_name = \"\"\n        res.model_version = self._version\n        return res\n\n    def GetModel(self, request, context):\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 = master_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 ReportTaskResult(self, request, context):\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 = master_pb2.ReportTaskResultReply()\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        if request.err_message:\n            self.logger.warning(\"Worker error: %s\" % request.err_message)\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                # TODO: update model\n                self._version += 1\n                self._gradient_sum.clear()\n                self._grad_n = 0\n        res.accepted = True\n        res.model_version = self._version\n        return res\n", "path": "elasticdl/master/servicer.py"}], "after_files": [{"content": "import threading\nimport numpy as np\n\nimport tensorflow as tf\nassert tf.executing_eagerly()\n\nfrom proto import master_pb2\nfrom proto import master_pb2_grpc\nfrom util.ndarray import ndarray_to_tensor, tensor_to_ndarray\n\n\nclass MasterServicer(master_pb2_grpc.MasterServicer):\n    \"\"\"Master service implementation\"\"\"\n\n    def __init__(self, logger, grads_to_wait, optimizer):\n        self.logger = logger\n        self._opt = optimizer\n        self._lock = threading.Lock()\n        # TODO: random initialization\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\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(value, name=name, use_resource=True)\n\n    def GetTask(self, request, context):\n        # TODO: implent task queues. Return an empty task for now.\n        res = master_pb2.Task()\n        res.shard_file_name = \"\"\n        res.model_version = self._version\n        return res\n\n    def GetModel(self, request, context):\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 = master_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 ReportTaskResult(self, request, context):\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 = master_pb2.ReportTaskResultReply()\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        if request.err_message:\n            self.logger.warning(\"Worker error: %s\" % request.err_message)\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", "path": "elasticdl/master/servicer.py"}]}