TensorFlow:从多个线程中排队和出列队列

Question

我试图解决的问题如下:我有一个trainimgs文件名列表.我已经定义了一个

• tf.RandomShuffleQueue与它capacity=len(trainimgs)和min_after_dequeue=0.
• tf.RandomShuffleQueue预计这将填充trainimgs指定epochlimit的次数.
• 预计许多线程将并行工作.每个线程从一个元素中取出一个元素tf.RandomShuffleQueue并对其进行一些操作并将其排入另一个队列.我有那个部分是正确的.
• 然而,一旦1 epoch的trainimgs已被处理和tf.RandomShuffleQueue为空,前提是当前时期e < epochlimit,队列必须再次被填满和线程必须重新工作.

好消息是:我已经让它在某种情况下工作(最后见PS !!)

坏消息是:我认为有更好的方法可以做到这一点.

我现在使用的方法如下(我已经简化了功能并删除了基于预处理和后续排队的e图像处理,但处理的核心保持不变!!):

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. 这段代码的逻辑有什么问题吗？我对多线程场景不是很有经验,所以任何引起我注意的明显错误都对我很有帮助.

PS:看来这个代码毕竟不是完美的.当我运行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