queue - 如何将张量推送到 TensorFlow 队列并从另一个进程中提取它们？

Question

我有一个已启动并正在运行的 TensorFlow 集群，我正在尝试使用一个客户端进程将数据入队，并将其从另一进程中出队。我无法让它工作，我做错了什么？

这是我推送数据的程序:

# queue_push.py
import tensorflow as tf
import time

with tf.container("qtest"):
    q = tf.FIFOQueue(capacity=10, dtypes=[tf.float32],
                     shapes=[[]], name="q")
    v = tf.placeholder(tf.float32, shape=())
    enqueue = q.enqueue([v])

with tf.Session("grpc://localhost:2210") as sess:
    while True:
        t = time.time()
        print(t)
        sess.run(enqueue, feed_dict={v: t})
        time.sleep(1)

我的程序来提取数据:

# queue_pull.py
import tensorflow as tf
import time

with tf.container("qtest"):
    q = tf.FIFOQueue(capacity=10, dtypes=[tf.float32],
                     shapes=[[]], name="q")
    dequeue = q.dequeue()

with tf.Session("grpc://localhost:2222") as sess:
    while True:
        v = sess.run(dequeue)
        print("Pulled:", v)
        time.sleep(0.5)

当我运行它们时，我得到的是:

$ python queue_push.py
1472420887.974484
1472420888.991067
1472420889.995756
1472420890.998365
1472420892.001799
1472420893.008567
1472420894.011109
1472420895.014532
1472420896.02017
1472420897.024806
1472420898.03187
(then blocked forever)

同时:

$ python queue_pull.py
(blocked forever)

Answer 1

分享tf.FIFOQueue (或其他 TensorFlow 队列)在多个 session 之间，您需要将可选的 shared_name 参数传递给 constructor ，并在每个实例中将其设置为相同的字符串。例如，您可以在两个脚本中创建 q，如下所示:

q = tf.FIFOQueue(capacity=10, dtypes=[tf.float32], shapes=[[]],
                 shared_name="shared_q", name="q")

shared_name 不必与队列的name 相同，但它确实需要在同一容器中创建的所有其他队列中是唯一的同一设备。您可以共享队列，而无需使用 with tf.container():堵塞; tf.container() block 提供了一种对有状态对象进行分组的方法，以便可以有选择地清除它们(使用 tf.Session.reset() )。

<小时/>

注意tf.Variable 的默认共享行为有所不同。对象，它们根据其 name 属性默认共享。否则，仅当您在构造函数中设置 shared_name 参数时，TensorFlow 图中的所有共享对象(队列、读取器等)才会被共享。