prompt: 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()
  
  ```
  
golden_diff: 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
                       )
  
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@@ -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"}]}