title: How to add new op in distributed Tensorflow?
body: <p>I'm trying to add a new op in distributed tensorflow. Following code is how I create a new operator which can add two array (following this tutorial: <a href="https://www.tensorflow.org/api_docs/python/tf/py_func" rel="nofollow noreferrer">https://www.tensorflow.org/api_docs/python/tf/py_func</a>).</p>
  
  <pre><code>import tensorflow as tf
  import numpy as np
  
  array1 = np.array([[1, 2], [3, 4]], dtype=np.float32)
  array2 = np.array([[5, 6], [7, 8]], dtype=np.float32)
  
  def add(array1, array2):
      return array1 + array2
  
  add_op = tf.py_func(add, [array1, array2], [tf.float32])
  
  
  sess = tf.Session()
  print sess.run(add_op)
  </code></pre>
  
  <p>Then I follow this tutorial(<a href="https://www.tensorflow.org/deploy/distributed" rel="nofollow noreferrer">https://www.tensorflow.org/deploy/distributed</a>) to create a cluster of TensorFlow servers and try put an operator in it. I start the nodes, use the following code and command line.  </p>
  
  <pre><code>#worker.py
  import tensorflow as tf
  
  # Configuration of cluster 
  ps_hosts = [ "127.0.0.1:8887" ]
  worker_hosts = [ "127.0.0.1:8888", "127.0.0.1:8889" ]
  cluster = tf.train.ClusterSpec({"ps": ps_hosts, "worker": worker_hosts})      
  
  tf.app.flags.DEFINE_string("job_name", "worker", "One of 'ps', 'worker'")
  tf.app.flags.DEFINE_integer("task_index", 0, "Index of task within the job")
  
  FLAGS = tf.app.flags.FLAGS
  
  def main(_):
      server = tf.train.Server(cluster,
                               job_name=FLAGS.job_name,
                               task_index=FLAGS.task_index)
      server.join()
  
  if __name__ == "__main__":
      tf.app.run()
  </code></pre>
  
  <p>Command line:</p>
  
  <pre><code># On worker0
  python worker.py --job_name=worker --task_index=0
  
  # On ps0
  python worker.py --job_name=ps --task_index=0
  </code></pre>
  
  <p>Then execute this file to specify whether ops run.</p>
  
  <pre><code>#master.py
  
  import tensorflow as tf
  import numpy as np
  
  def add(array1, array2):
      return array1 + array2    
  
  with tf.device("/job:ps/task:0"):
      array1 = np.array([[1, 2], [3, 4]], dtype=np.float32)
      array2 = np.array([[5, 6], [7, 8]], dtype=np.float32)
  
  with tf.device("/job:worker/task:0"):
      add_op = tf.py_func(add, [array1, array2], [tf.float32])
  
  
  with tf.Session("grpc://127.0.0.1:8887") as sess:
      print sess.run(add_op)
  </code></pre>
  
  <p>But this code is not working, following is the error </p>
  
  <pre><code>Traceback (most recent call last):
    File "master.py", line 19, in &lt;module&gt;
      print sess.run(add_op)
    File "/Users/WakeUp/Repository/tensorflow/lib/python2.7/site-packages/tensorflow/python/client/session.py", line 767, in run
      run_metadata_ptr)
    File "/Users/WakeUp/Repository/tensorflow/lib/python2.7/site-packages/tensorflow/python/client/session.py", line 965, in _run
      feed_dict_string, options, run_metadata)
    File "/Users/WakeUp/Repository/tensorflow/lib/python2.7/site-packages/tensorflow/python/client/session.py", line 1015, in _do_run
      target_list, options, run_metadata)
    File "/Users/WakeUp/Repository/tensorflow/lib/python2.7/site-packages/tensorflow/python/client/session.py", line 1035, in _do_call
      raise type(e)(node_def, op, message)
  tensorflow.python.framework.errors_impl.InternalError: Failed to run py callback pyfunc_0: see error log.
       [[Node: PyFunc = PyFunc[Tin=[DT_FLOAT, DT_FLOAT], Tout=[DT_FLOAT], token="pyfunc_0", _device="/job:worker/replica:0/task:0/cpu:0"](PyFunc/input_0, PyFunc/input_1)]]
       [[Node: PyFunc_S1 = _Recv[client_terminated=false, recv_device="/job:ps/replica:0/task:0/cpu:0", send_device="/job:worker/replica:0/task:0/cpu:0", send_device_incarnation=-6409934414370243221, tensor_name="edge_8_PyFunc", tensor_type=DT_FLOAT, _device="/job:ps/replica:0/task:0/cpu:0"]()]]
  
  Caused by op u'PyFunc', defined at:
    File "master.py", line 15, in &lt;module&gt;
      add_op = tf.py_func(add, [array1, array2], [tf.float32])
    File "/Users/WakeUp/Repository/tensorflow/lib/python2.7/site-packages/tensorflow/python/ops/script_ops.py", line 189, in py_func
      input=inp, token=token, Tout=Tout, name=name)
    File "/Users/WakeUp/Repository/tensorflow/lib/python2.7/site-packages/tensorflow/python/ops/gen_script_ops.py", line 40, in _py_func
      name=name)
    File "/Users/WakeUp/Repository/tensorflow/lib/python2.7/site-packages/tensorflow/python/framework/op_def_library.py", line 763, in apply_op
      op_def=op_def)
    File "/Users/WakeUp/Repository/tensorflow/lib/python2.7/site-packages/tensorflow/python/framework/ops.py", line 2327, in create_op
      original_op=self._default_original_op, op_def=op_def)
    File "/Users/WakeUp/Repository/tensorflow/lib/python2.7/site-packages/tensorflow/python/framework/ops.py", line 1226, in __init__
      self._traceback = _extract_stack()
  
  InternalError (see above for traceback): Failed to run py callback pyfunc_0: see error log.
       [[Node: PyFunc = PyFunc[Tin=[DT_FLOAT, DT_FLOAT], Tout=[DT_FLOAT], token="pyfunc_0", _device="/job:worker/replica:0/task:0/cpu:0"](PyFunc/input_0, PyFunc/input_1)]]
       [[Node: PyFunc_S1 = _Recv[client_terminated=false, recv_device="/job:ps/replica:0/task:0/cpu:0", send_device="/job:worker/replica:0/task:0/cpu:0", send_device_incarnation=-6409934414370243221, tensor_name="edge_8_PyFunc", tensor_type=DT_FLOAT, _device="/job:ps/replica:0/task:0/cpu:0"]()]]
  </code></pre>
  
  <p>According to the N.B. in this page (<a href="https://www.tensorflow.org/api_docs/python/tf/py_func" rel="nofollow noreferrer">https://www.tensorflow.org/api_docs/python/tf/py_func</a>), I think I didn't violate any rule of py_func function. Such as serializing the model or forgetting run <code>tf.train.Server</code> and <code>tf.device()</code>. Is there still something I didn't notice?</p>