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()

```

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

{"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"}]}