Tensorflow如何减少/广播集群中有多个worker的tensor？

Question

我只是想用一种同步的方式，在tensorflow中求一个工人之间的张量之和。感觉这应该是非常容易的，但我没有找到一个方法。你知道吗

我的结论是tf.分配太抽象了，我需要一些低级操作，因为我只想在我的程序中操作一个张量。如果里面有什么，请纠正我tf.分配这让我减少/广播一个张量，每个worker上有一个实例。你知道吗

下面是我尝试使用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

我希望他们两个都打印sum [-2, -6]，但是他们被阻塞调用less.run(res)

当我只启动两个进程中的一个时，它会一直打印“2019-08-01 12:05:24.324155:I tensorflow/core/distributed\u runtime”/抄送：268]CreateSession仍在等待工作进程的响应：/作业：辅助进程/副本:0/任务：0", i、 e.等待其他工人的回应。 当我启动第二个进程时，上面的日志记录停止了，但是什么也没有发生。你知道吗

我还试着用 with tf.device(tf.train.replica_device_setter(worker_device="/job:worker/task:0/device:CPU:0", cluster=cluster)): 但是我在定义张量时遇到了一个错误： “ValueError:集体操作需要设备分配”

Answer 1

问题在github上提出，并收到团队成员的响应。有一个代码片段可以解决这个图形之间的通信问题。你知道吗