TensorFlow-如何用群集中的多个工人减少/广播张量

Question

我只是在试图以同步方式以张量的工人概括张量。感觉这应该非常容易，但是我还没有找到方法。

我得出的结论是，tf.distribute中的任何内容都太抽象了，我需要一些较低级别的操作，因为我只想在程序中的单个张量上操作。如果TF。分发中有任何内容，请纠正我

这是我尝试使用collective_ops.all_reduce

的一些代码

import sys
import tensorflow as tf
from tensorflow.python.ops import collective_ops
task_id = int(sys.argv[1])
cluster = tf.train.ClusterSpec({"worker": ["localhost:2222", "localhost:2223"]})
server = tf.train.Server(cluster, job_name="worker", task_index=task_id)
with tf.device("/job:worker/task:{}/device:CPU:0".format(task_id)):
    t = tf.constant([-1,-3], name='testtensor')
    res = collective_ops.all_reduce(t, group_size=2, group_key=123, instance_key=123, merge_op='Add', final_op='Id', subdiv_offsets=(0,))
with tf.Session(server.target) as sess:
    print('running reduce..')
    print(sess.run(res))

如果我在一个终端中运行上述脚本： python myscript.py 0以及另一个终端中的另一个实例： python myscript.py 1

我希望他们两个都可以打印总和[-2, -6]，但它们被卡在less.run(res)

的封锁调用中

当我仅启动两个过程之一时，它会不断打印" 2019-08-01 12：05：24.324155：i Tensorflow/core/core/distractuted_runtime/master.cc：268]创建仍在等待工作的响应：/job：wrorker/epplica：0/task：0"，即等待其他工人的回应。当我启动第二个过程时，上述记录停止，但无论如何都没有发生。

我还尝试使用使用张量 with tf.device(tf.train.replica_device_setter(worker_device="/job:worker/task:0/device:CPU:0", cluster=cluster)):但是后来我遇到了这个错误，试图定义张量：" ValueError：集体OPS所需的设备分配"

Answer 1

在Github上提出了问题，并从团队成员那里收到了响应。有一个代码段来解决此图间通信问题。