TensorFlow : Enqueuing and dequeuing a queue from multiple threads

Question

我试图解决的问题如下:我有一个 list trainimgs文件名。我定义了一个

• tf.RandomShuffleQueue及其capacity=len(trainimgs)和min_after_dequeue=0 。
• 这个tf.RandomShuffleQueue预计由 trainimgs 填补对于指定的epochlimit次数。
• 许多线程需要并行工作。每个线程从 tf.RandomShuffleQueue 中取出一个元素并对它进行一些操作并将其排入另一个队列。这部分我是对的。
• 但是一次1 epoch的trainimgs已处理，tf.RandomShuffleQueue为空，前提是当前纪元 e < epochlimit ，队列必须再次被填满，线程必须再次工作。

好消息是:我已经在某种情况下工作了(请参阅最后的PS!!)

坏消息是:我认为有更好的方法来做到这一点。

我现在用来执行此操作的方法如下(我简化了功能并删除了基于图像处理的预处理和后续排队，但处理的核心保持不变!):

with tf.Session() as sess:
    train_filename_queue = tf.RandomShuffleQueue(capacity=len(trainimgs), min_after_dequeue=0, dtypes=tf.string, seed=0)
    queue_size = train_filename_queue.size()
    trainimgtensor = tf.constant(trainimgs)
    close_queue = train_filename_queue.close()
    epoch = tf.Variable(initial_value=1, trainable=False, dtype=tf.int32)
    incrementepoch = tf.assign(epoch, epoch + 1, use_locking=True)
    supplyimages = train_filename_queue.enqueue_many(trainimgtensor)
    value = train_filename_queue.dequeue()

    init_op = tf.group(tf.global_variables_initializer(), tf.local_variables_initializer())
    sess.run(init_op)
    coord = tf.train.Coordinator()
    tf.train.start_queue_runners(sess, coord)
    sess.run(supplyimages)
    lock = threading.Lock()
    threads = [threading.Thread(target=work, args=(coord, value, sess, epoch, incrementepoch, supplyimages, queue_size, lock, close_queue)) for  i in range(200)] 
    for t in threads:
        t.start()
    coord.join(threads)

工作函数如下:

def work(coord, val, sess, epoch, incrementepoch, supplyimg, q, lock,\
         close_op):
while not coord.should_stop():
    if sess.run(q) > 0:
        filename, currepoch = sess.run([val, epoch])
        filename = filename.decode(encoding='UTF-8')
        print(filename + ' ' + str(currepoch))
    elif sess.run(epoch) < 2:
        lock.acquire()
        try:
            if sess.run(q) == 0:
                print("The previous epoch = %d"%(sess.run(epoch)))
                sess.run([incrementepoch, supplyimg])
                sz = sess.run(q)
                print("The new epoch = %d"%(sess.run(epoch)))
                print("The new queue size = %d"%(sz))
        finally:
            lock.release()
    else:
        try:
            sess.run(close_op)
        except tf.errors.CancelledError:
            print('Queue already closed.')
        coord.request_stop()
return None

所以，虽然这有效，但我有一种感觉，有一种更好、更干净的方法来实现这一目标。所以，简而言之，我的问题是:

1. 是否有更简单、更清晰的方法在 TensorFlow 中完成此任务？
2. 这段代码的逻辑有问题吗？我对多线程场景不是很有经验，所以任何我没有注意到的明显错误都会对我很有帮助。

P.S:看来这段代码毕竟并不完美。当我运行 120 万张图像和 200 个线程时，它运行了。但是，当我运行 10 个图像和 20 个线程时，出现以下错误:

CancelledError (see above for traceback): RandomShuffleQueue '_0_random_shuffle_queue' is closed.
     [[Node: random_shuffle_queue_EnqueueMany = QueueEnqueueManyV2[Tcomponents=[DT_STRING], timeout_ms=-1, _device="/job:localhost/replica:0/task:0/cpu:0"](random_shuffle_queue, Const)]]

我以为 except tf.errors.CancelledError 涵盖了这一点。这到底是怎么回事？

Answer 1

我终于找到答案了。问题在于多个线程在 work() 函数中的各个点上发生冲突。以下 work() 函数完美运行。

def work(coord, val, sess, epoch, maxepochs, incrementepoch, supplyimg, q, lock, close_op):
    print('I am thread number %s'%(threading.current_thread().name))
    print('I can see a queue with size %d'%(sess.run(q)))
    while not coord.should_stop():
        lock.acquire()
        if sess.run(q) > 0:
            filename, currepoch = sess.run([val, epoch])
            filename = filename.decode(encoding='UTF-8')
            tid = threading.current_thread().name
            print(filename + ' ' + str(currepoch) + ' thread ' + str(tid))
        elif sess.run(epoch) < maxepochs:
            print('Thread %s has acquired the lock'%(threading.current_thread().name))
            print("The previous epoch = %d"%(sess.run(epoch)))
            sess.run([incrementepoch, supplyimg])
            sz = sess.run(q)
            print("The new epoch = %d"%(sess.run(epoch)))
            print("The new queue size = %d"%(sz))
    else:
            coord.request_stop()
        lock.release()

    return None