We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Change PS to run in graph mode.
在multi-thread下面， 如果ps enable eager mode, 在同一个process下面，所有thread都是在eager mode下面了。 eager mode 可以run graph, 反之则不行。

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `python/elasticdl/tflib/ps/ps.py`
Content:
```
1 import threading
2 import queue
3 import numpy as np
4 import tensorflow.contrib.eager as tfe
5 import tensorflow as tf
6 tf.enable_eager_execution()
7 
8 
9 class ParameterServer(object):
10     def __init__(self, optimizer, vars):
11         self._opt = optimizer
12         self._vars = {}
13         for k, v in vars.items():
14             if (not isinstance(v, np.ndarray)
15                     or v.dtype not in (np.float32, np.float64)):
16                 raise ValueError(
17                     'Initial value for variable %s is not of float type ndarray' %
18                     k)
19             self._vars[k] = tfe.Variable(v, name=k)
20         self._step = 0
21         self._grad_q = queue.Queue()
22         self._lock = threading.Lock()
23         self._runner = threading.Thread(target=self._run, name='ps-runner')
24         self._exiting = False
25         self._min_step_cv = threading.Condition()
26 
27     def pull(self, names=None, min_step=0, blocking=True, timeout=None):
28         with self._min_step_cv:
29             self._min_step_cv.wait_for(
30                 lambda: not blocking or min_step <= self._step,
31                 timeout=timeout)
32         with self._lock:
33             if min_step > self._step:
34                 raise LookupError(
35                     'Required step is not ready yet: %s' %
36                     min_step)
37             if names:
38                 res = {k: self._vars[k].numpy() for k in names}
39             else:
40                 res = {k: v.numpy() for k, v in self._vars.items()}
41             return self._step, res
42 
43     def push(self, base_step, sub_step, grads):
44         with self._lock:
45             if base_step > self._step:
46                 raise ValueError(
47                     'Illegal base step %s, parameter server step is %s' %
48                     (base_step, self._step))
49 
50         if sub_step < 0:
51             raise ValueError('Illegal sub step %s' % sub_step)
52 
53         for k, g in grads.items():
54             v = self._vars[k]
55             if g.dtype != v.dtype.as_numpy_dtype or g.shape != v.shape:
56                 raise ValueError('Incompatible gradient for variable %s' % k)
57         # TODO(l.zou): use @dataclass when python 3.7 is available.
58         self._grad_q.put((base_step, sub_step, grads))
59 
60     def _compute(self, grads):
61         grads_vars = [(g, self._vars[k]) for k, g in grads.items()]
62         with self._lock:
63             self._opt.apply_gradients(grads_vars)
64         with self._min_step_cv:
65             self._step += 1
66             self._min_step_cv.notify_all()
67 
68     def _run(self):
69         while not self._exiting:
70             # TODO(l.zou): How to properly accumulate and decay grads?
71             try:
72                 base_step, sub_step, grads = self._grad_q.get(timeout=1.0)
73                 self._compute(grads)
74             except queue.Empty:
75                 pass
76 
77     def start(self):
78         self._runner.start()
79 
80     def join(self):
81         self._exiting = True
82         self._runner.join()
83
```
--- 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/python/elasticdl/tflib/ps/ps.py b/python/elasticdl/tflib/ps/ps.py
--- a/python/elasticdl/tflib/ps/ps.py
+++ b/python/elasticdl/tflib/ps/ps.py
@@ -1,22 +1,12 @@
 import threading
 import queue
 import numpy as np
-import tensorflow.contrib.eager as tfe
 import tensorflow as tf
-tf.enable_eager_execution()
+from tensorflow.python.ops import array_ops
 
 
 class ParameterServer(object):
     def __init__(self, optimizer, vars):
-        self._opt = optimizer
-        self._vars = {}
-        for k, v in vars.items():
-            if (not isinstance(v, np.ndarray)
-                    or v.dtype not in (np.float32, np.float64)):
-                raise ValueError(
-                    'Initial value for variable %s is not of float type ndarray' %
-                    k)
-            self._vars[k] = tfe.Variable(v, name=k)
         self._step = 0
         self._grad_q = queue.Queue()
         self._lock = threading.Lock()
@@ -24,6 +14,23 @@
         self._exiting = False
         self._min_step_cv = threading.Condition()
 
+        self._grads_vars = {}
+        for k, v in vars.items():
+            if (not isinstance(v, np.ndarray)
+                    or v.dtype not in (np.float32, np.float64)):
+                raise ValueError(
+                    'Initial value for variable %s is not of float type ndarray' %
+                    k)
+            # TODO: In graph mode we don't need to keep track of variables by ourselves.
+            self._grads_vars[k] = (array_ops.placeholder(dtype=v.dtype), tf.Variable(v, name=k))
+
+        self._opt = optimizer
+        self._apply_grad_op = self._opt.apply_gradients(self._grads_vars.values())
+
+        self._sess = tf.Session()
+        init_op = tf.global_variables_initializer()
+        self._sess.run(init_op)
+
     def pull(self, names=None, min_step=0, blocking=True, timeout=None):
         with self._min_step_cv:
             self._min_step_cv.wait_for(
@@ -35,9 +42,9 @@
                     'Required step is not ready yet: %s' %
                     min_step)
             if names:
-                res = {k: self._vars[k].numpy() for k in names}
+                res = {k: self._grads_vars[k][1].eval(self._sess) for k in names}
             else:
-                res = {k: v.numpy() for k, v in self._vars.items()}
+                res = {k: v[1].eval(self._sess) for k, v in self._grads_vars.items()}
             return self._step, res
 
     def push(self, base_step, sub_step, grads):
@@ -51,16 +58,16 @@
             raise ValueError('Illegal sub step %s' % sub_step)
 
         for k, g in grads.items():
-            v = self._vars[k]
+            v = self._grads_vars[k][1]
             if g.dtype != v.dtype.as_numpy_dtype or g.shape != v.shape:
                 raise ValueError('Incompatible gradient for variable %s' % k)
         # TODO(l.zou): use @dataclass when python 3.7 is available.
         self._grad_q.put((base_step, sub_step, grads))
 
     def _compute(self, grads):
-        grads_vars = [(g, self._vars[k]) for k, g in grads.items()]
         with self._lock:
-            self._opt.apply_gradients(grads_vars)
+            feed_dict = {self._grads_vars[k][0]:v for k, v in grads.items()}
+            self._sess.run(self._apply_grad_op, feed_dict=feed_dict)
         with self._min_step_cv:
             self._step += 1
             self._min_step_cv.notify_all()
@@ -80,3 +87,4 @@
     def join(self):
         self._exiting = True
         self._runner.join()
+        self._sess.close()

{"golden_diff": "diff --git a/python/elasticdl/tflib/ps/ps.py b/python/elasticdl/tflib/ps/ps.py\n--- a/python/elasticdl/tflib/ps/ps.py\n+++ b/python/elasticdl/tflib/ps/ps.py\n@@ -1,22 +1,12 @@\n import threading\n import queue\n import numpy as np\n-import tensorflow.contrib.eager as tfe\n import tensorflow as tf\n-tf.enable_eager_execution()\n+from tensorflow.python.ops import array_ops\n \n \n class ParameterServer(object):\n     def __init__(self, optimizer, vars):\n-        self._opt = optimizer\n-        self._vars = {}\n-        for k, v in vars.items():\n-            if (not isinstance(v, np.ndarray)\n-                    or v.dtype not in (np.float32, np.float64)):\n-                raise ValueError(\n-                    'Initial value for variable %s is not of float type ndarray' %\n-                    k)\n-            self._vars[k] = tfe.Variable(v, name=k)\n         self._step = 0\n         self._grad_q = queue.Queue()\n         self._lock = threading.Lock()\n@@ -24,6 +14,23 @@\n         self._exiting = False\n         self._min_step_cv = threading.Condition()\n \n+        self._grads_vars = {}\n+        for k, v in vars.items():\n+            if (not isinstance(v, np.ndarray)\n+                    or v.dtype not in (np.float32, np.float64)):\n+                raise ValueError(\n+                    'Initial value for variable %s is not of float type ndarray' %\n+                    k)\n+            # TODO: In graph mode we don't need to keep track of variables by ourselves.\n+            self._grads_vars[k] = (array_ops.placeholder(dtype=v.dtype), tf.Variable(v, name=k))\n+\n+        self._opt = optimizer\n+        self._apply_grad_op = self._opt.apply_gradients(self._grads_vars.values())\n+\n+        self._sess = tf.Session()\n+        init_op = tf.global_variables_initializer()\n+        self._sess.run(init_op)\n+\n     def pull(self, names=None, min_step=0, blocking=True, timeout=None):\n         with self._min_step_cv:\n             self._min_step_cv.wait_for(\n@@ -35,9 +42,9 @@\n                     'Required step is not ready yet: %s' %\n                     min_step)\n             if names:\n-                res = {k: self._vars[k].numpy() for k in names}\n+                res = {k: self._grads_vars[k][1].eval(self._sess) for k in names}\n             else:\n-                res = {k: v.numpy() for k, v in self._vars.items()}\n+                res = {k: v[1].eval(self._sess) for k, v in self._grads_vars.items()}\n             return self._step, res\n \n     def push(self, base_step, sub_step, grads):\n@@ -51,16 +58,16 @@\n             raise ValueError('Illegal sub step %s' % sub_step)\n \n         for k, g in grads.items():\n-            v = self._vars[k]\n+            v = self._grads_vars[k][1]\n             if g.dtype != v.dtype.as_numpy_dtype or g.shape != v.shape:\n                 raise ValueError('Incompatible gradient for variable %s' % k)\n         # TODO(l.zou): use @dataclass when python 3.7 is available.\n         self._grad_q.put((base_step, sub_step, grads))\n \n     def _compute(self, grads):\n-        grads_vars = [(g, self._vars[k]) for k, g in grads.items()]\n         with self._lock:\n-            self._opt.apply_gradients(grads_vars)\n+            feed_dict = {self._grads_vars[k][0]:v for k, v in grads.items()}\n+            self._sess.run(self._apply_grad_op, feed_dict=feed_dict)\n         with self._min_step_cv:\n             self._step += 1\n             self._min_step_cv.notify_all()\n@@ -80,3 +87,4 @@\n     def join(self):\n         self._exiting = True\n         self._runner.join()\n+        self._sess.close()\n", "issue": "Change PS to run in graph mode.\n\u5728multi-thread\u4e0b\u9762\uff0c \u5982\u679cps enable eager mode, \u5728\u540c\u4e00\u4e2aprocess\u4e0b\u9762\uff0c\u6240\u6709thread\u90fd\u662f\u5728eager mode\u4e0b\u9762\u4e86\u3002 eager mode \u53ef\u4ee5run graph, \u53cd\u4e4b\u5219\u4e0d\u884c\u3002\n", "before_files": [{"content": "import threading\nimport queue\nimport numpy as np\nimport tensorflow.contrib.eager as tfe\nimport tensorflow as tf\ntf.enable_eager_execution()\n\n\nclass ParameterServer(object):\n    def __init__(self, optimizer, vars):\n        self._opt = optimizer\n        self._vars = {}\n        for k, v in vars.items():\n            if (not isinstance(v, np.ndarray)\n                    or v.dtype not in (np.float32, np.float64)):\n                raise ValueError(\n                    'Initial value for variable %s is not of float type ndarray' %\n                    k)\n            self._vars[k] = tfe.Variable(v, name=k)\n        self._step = 0\n        self._grad_q = queue.Queue()\n        self._lock = threading.Lock()\n        self._runner = threading.Thread(target=self._run, name='ps-runner')\n        self._exiting = False\n        self._min_step_cv = threading.Condition()\n\n    def pull(self, names=None, min_step=0, blocking=True, timeout=None):\n        with self._min_step_cv:\n            self._min_step_cv.wait_for(\n                lambda: not blocking or min_step <= self._step,\n                timeout=timeout)\n        with self._lock:\n            if min_step > self._step:\n                raise LookupError(\n                    'Required step is not ready yet: %s' %\n                    min_step)\n            if names:\n                res = {k: self._vars[k].numpy() for k in names}\n            else:\n                res = {k: v.numpy() for k, v in self._vars.items()}\n            return self._step, res\n\n    def push(self, base_step, sub_step, grads):\n        with self._lock:\n            if base_step > self._step:\n                raise ValueError(\n                    'Illegal base step %s, parameter server step is %s' %\n                    (base_step, self._step))\n\n        if sub_step < 0:\n            raise ValueError('Illegal sub step %s' % sub_step)\n\n        for k, g in grads.items():\n            v = self._vars[k]\n            if g.dtype != v.dtype.as_numpy_dtype or g.shape != v.shape:\n                raise ValueError('Incompatible gradient for variable %s' % k)\n        # TODO(l.zou): use @dataclass when python 3.7 is available.\n        self._grad_q.put((base_step, sub_step, grads))\n\n    def _compute(self, grads):\n        grads_vars = [(g, self._vars[k]) for k, g in grads.items()]\n        with self._lock:\n            self._opt.apply_gradients(grads_vars)\n        with self._min_step_cv:\n            self._step += 1\n            self._min_step_cv.notify_all()\n\n    def _run(self):\n        while not self._exiting:\n            # TODO(l.zou): How to properly accumulate and decay grads?\n            try:\n                base_step, sub_step, grads = self._grad_q.get(timeout=1.0)\n                self._compute(grads)\n            except queue.Empty:\n                pass\n\n    def start(self):\n        self._runner.start()\n\n    def join(self):\n        self._exiting = True\n        self._runner.join()\n", "path": "python/elasticdl/tflib/ps/ps.py"}], "after_files": [{"content": "import threading\nimport queue\nimport numpy as np\nimport tensorflow as tf\nfrom tensorflow.python.ops import array_ops\n\n\nclass ParameterServer(object):\n    def __init__(self, optimizer, vars):\n        self._step = 0\n        self._grad_q = queue.Queue()\n        self._lock = threading.Lock()\n        self._runner = threading.Thread(target=self._run, name='ps-runner')\n        self._exiting = False\n        self._min_step_cv = threading.Condition()\n\n        self._grads_vars = {}\n        for k, v in vars.items():\n            if (not isinstance(v, np.ndarray)\n                    or v.dtype not in (np.float32, np.float64)):\n                raise ValueError(\n                    'Initial value for variable %s is not of float type ndarray' %\n                    k)\n            # TODO: In graph mode we don't need to keep track of variables by ourselves.\n            self._grads_vars[k] = (array_ops.placeholder(dtype=v.dtype), tf.Variable(v, name=k))\n\n        self._opt = optimizer\n        self._apply_grad_op = self._opt.apply_gradients(self._grads_vars.values())\n\n        self._sess = tf.Session()\n        init_op = tf.global_variables_initializer()\n        self._sess.run(init_op)\n\n    def pull(self, names=None, min_step=0, blocking=True, timeout=None):\n        with self._min_step_cv:\n            self._min_step_cv.wait_for(\n                lambda: not blocking or min_step <= self._step,\n                timeout=timeout)\n        with self._lock:\n            if min_step > self._step:\n                raise LookupError(\n                    'Required step is not ready yet: %s' %\n                    min_step)\n            if names:\n                res = {k: self._grads_vars[k][1].eval(self._sess) for k in names}\n            else:\n                res = {k: v[1].eval(self._sess) for k, v in self._grads_vars.items()}\n            return self._step, res\n\n    def push(self, base_step, sub_step, grads):\n        with self._lock:\n            if base_step > self._step:\n                raise ValueError(\n                    'Illegal base step %s, parameter server step is %s' %\n                    (base_step, self._step))\n\n        if sub_step < 0:\n            raise ValueError('Illegal sub step %s' % sub_step)\n\n        for k, g in grads.items():\n            v = self._grads_vars[k][1]\n            if g.dtype != v.dtype.as_numpy_dtype or g.shape != v.shape:\n                raise ValueError('Incompatible gradient for variable %s' % k)\n        # TODO(l.zou): use @dataclass when python 3.7 is available.\n        self._grad_q.put((base_step, sub_step, grads))\n\n    def _compute(self, grads):\n        with self._lock:\n            feed_dict = {self._grads_vars[k][0]:v for k, v in grads.items()}\n            self._sess.run(self._apply_grad_op, feed_dict=feed_dict)\n        with self._min_step_cv:\n            self._step += 1\n            self._min_step_cv.notify_all()\n\n    def _run(self):\n        while not self._exiting:\n            # TODO(l.zou): How to properly accumulate and decay grads?\n            try:\n                base_step, sub_step, grads = self._grad_q.get(timeout=1.0)\n                self._compute(grads)\n            except queue.Empty:\n                pass\n\n    def start(self):\n        self._runner.start()\n\n    def join(self):\n        self._exiting = True\n        self._runner.join()\n        self._sess.close()\n", "path": "python/elasticdl/tflib/ps/ps.py"}]}