python - Tensorflow QueueRunner 与 py_func enqueue_op : How to use?

Question

我正在尝试使用自定义 py_func enqueue_op与 TensorFlow RandomShuffleQueue和QueueRunner 。我对TensorFlow很陌生并且很困惑。这是我现在拥有的:

def compute_data(symbol, time):
    data = np.zeros((1330,))
    return data

key_1 = [str(x) for x in range(3000)]
key_2 = [str(y) for y in range(4800)]
tf_k1 = tf.constant([k for k in k1])
tf_k2 = tf.constant([k for k in k2])
tf_k1_index = tf.random_uniform((1,), minval=0, maxval=len(k1), dtype=tf.int32, name='k1_index')
tf_k2_index = tf.random_uniform((1,), minval=0, maxval=len(k2), dtype=tf.int32, name='k2_index')
tf_k1_variable = tf.gather_nd(tf_symbols, tf_k1_index)
tf_k2_variable = tf.gather_nd(tf_times, tf_k2_index)
tf_compute_data = tf.py_func(compute_data, [tf_k1_variable, tf_k2_variable], tf.float32, name='py_func_compute_data')

基本上，我在这里想要实现的是给定两组 key ，每次随机采样这两个 key 的组合，并根据这两个 key 生成一段数据。数据生成过程涉及大量文件读取，暂时跳过，因为我想首先正确构建图表。

下面是应将 tf_compute_data 的结果排入队列的其余代码进入queue .

queue = tf.RandomShuffleQueue(
    capacity=20000, 
    min_after_dequeue=2000, 
    dtypes=[tf.float32], 
    shapes=[[1330]], 
    name='data_queue'
    )

enqueue_op = queue.enqueue(tf_compute_data)
tf_data = queue.dequeue_many(batch_size)

...

qr = tf.train.QueueRunner(queue, [enqueue_op] * 4)
sv = tf.train.Supervisor(logdir="logdir")
with sv.managed_session(config=config, start_standard_services=True) as sess:
    coord = tf.train.Coordinator()
    enqueue_threads = qr.create_threads(sess, coord=coord, start=True)

    for step in xrange(1000000):
        if coord.should_stop():
            break
        sess.run(train_op)
        print step

    coord.request_stop()
    coord.join(enqueue_threads)

当我运行脚本时，错误显示如下:

W tensorflow/core/framework/op_kernel.cc:993] Out of range: RandomShuffleQueue '_0_data_queue' is closed and has insufficient elements (requested 64, current size 0)
     [[Node: data_queue_DequeueMany = QueueDequeueManyV2[component_types=[DT_FLOAT], timeout_ms=-1, _device="/job:localhost/replica:0/task:0/cpu:0"](data_queue, data_queue_DequeueMany/n)]]
W tensorflow/core/framework/op_kernel.cc:993] Out of range: RandomShuffleQueue '_0_data_queue' is closed and has insufficient elements (requested 64, current size 0)
     [[Node: data_queue_DequeueMany = QueueDequeueManyV2[component_types=[DT_FLOAT], timeout_ms=-1, _device="/job:localhost/replica:0/task:0/cpu:0"](data_queue, data_queue_DequeueMany/n)]]

当我将日志记录添加到 compute_data 时函数显示它只运行了 4 次，每个线程一次。我如何让它运行只要 coord.should_stop()是 False ？

Answer 1

总结一下评论，有两个问题:

首先，with tf.Graph().as_default() 从头开始​​，因此所有内容都需要在新图中重新定义。

其次，py_func 返回的 dtype 有点棘手，因为 numpy 默认为 float64，而大多数 TensorFlow 函数默认为 >float32。因此，在定义 py_func 时，可能需要将 numpy 数组的 dtype 显式设置为 float32。有一条错误消息，但我认为它已写入不同的流(因此，如果您到达此页面时遇到类似的队列错误和 py_func dtype 匹配不是问题，请确保检查 stdout 和 stderr 是否存在根本错误)。