prompt: You will be provided with a partial code base and an issue statement explaining a problem to resolve.
  <issue>
  Change PS to run in graph mode.
  在multi-thread下面， 如果ps enable eager mode, 在同一个process下面，所有thread都是在eager mode下面了。 eager mode 可以run graph, 反之则不行。
  
  </issue>
  <code>
  [start of python/elasticdl/tflib/ps/ps.py]
  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 of python/elasticdl/tflib/ps/ps.py]
  </code>
  
  I need you to solve this issue by generating a single patch file that I can apply directly to this repository using git apply. Please respond with a single patch file in the following format.
  <patch>
  diff --git a/file.py b/file.py
  --- a/file.py
  +++ b/file.py
  @@ -1,27 +1,35 @@
   def euclidean(a, b):
  -    while b:
  -        a, b = b, a % b
  -    return a
  +    if b == 0:
  +        return a
  +    return euclidean(b, a % b)
   
   
   def bresenham(x0, y0, x1, y1):
       points = []
       dx = abs(x1 - x0)
       dy = abs(y1 - y0)
  -    sx = 1 if x0 < x1 else -1
  -    sy = 1 if y0 < y1 else -1
  -    err = dx - dy
  +    x, y = x0, y0
  +    sx = -1 if x0 > x1 else 1
  +    sy = -1 if y0 > y1 else 1
   
  -    while True:
  -        points.append((x0, y0))
  -        if x0 == x1 and y0 == y1:
  -            break
  -        e2 = 2 * err
  -        if e2 > -dy:
  -            err -= dy
  -            x0 += sx
  -        if e2 < dx:
  -            err += dx
  -            y0 += sy
  +    if dx > dy:
  +        err = dx / 2.0
  +        while x != x1:
  +            points.append((x, y))
  +            err -= dy
  +            if err < 0:
  +                y += sy
  +                err += dx
  +            x += sx
  +    else:
  +        err = dy / 2.0
  +        while y != y1:
  +            points.append((x, y))
  +            err -= dx
  +            if err < 0:
  +                x += sx
  +                err += dy
  +            y += sy
  +
  +    points.append((x, y))
       return points
  
  </patch>
golden_diff: 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()
  
verification_info: {"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"}]}