We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[master]Use tf.ResourceVariable to store model
Currently we store model as a <string, ndarray> map. when using tf.optimizer.apply_gradient() to update model, we need to convert the map to ResourceVariable and back. It is better to change model to a <string, ResourceVariable> map to avoid copy and conversion.

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

{"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@@ -1,5 +1,7 @@\n import threading\n+import numpy as np\n \n+import tensorflow as tf\n from proto import master_pb2\n from proto import master_pb2_grpc\n from util.converter import NdarrayToTensor, TensorToNdarray\n@@ -12,12 +14,21 @@\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@@ -38,7 +49,7 @@\n         with self._lock:\n             res.version = self._version\n             for k, v in self._model.items():\n-                res.param[k].CopyFrom(NdarrayToTensor(v))\n+                res.param[k].CopyFrom(NdarrayToTensor(v.numpy()))\n         return res\n \n     def ReportTaskResult(self, request, context):\n@@ -76,7 +87,7 @@\n                         \"Gradient key: %s is not part of model\", k\n                     )\n                 arr = TensorToNdarray(v)\n-                if arr.shape != self._model[k].shape:\n+                if arr.shape != self._model[k].numpy().shape:\n                     raise ValueError(\n                         \"Gradient key: %s has incompatible dimension\", k\n                     )\n", "issue": "[master]Use tf.ResourceVariable to store model\nCurrently we store model as a <string, ndarray> map. when using tf.optimizer.apply_gradient() to update model, we need to convert the map to ResourceVariable and back. It is better to change model to a <string, ResourceVariable> map to avoid copy and conversion.\n", "before_files": [{"content": "import threading\n\nfrom proto import master_pb2\nfrom proto import master_pb2_grpc\nfrom util.converter import NdarrayToTensor, TensorToNdarray\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        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 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(NdarrayToTensor(v))\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 = TensorToNdarray(v)\n                if arr.shape != self._model[k].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\nfrom proto import master_pb2\nfrom proto import master_pb2_grpc\nfrom util.converter import NdarrayToTensor, TensorToNdarray\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(NdarrayToTensor(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 = TensorToNdarray(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"}]}