Tensor Flow shuffle_batch() блокирует в конце эпохи

Question

Я использую tf.train.shuffle_batch() для создания партий входных изображений. Он включает параметр min_after_dequeue, который гарантирует, что во внутренней очереди указано определенное количество элементов, и блокирует все остальное, если этого не происходит.

images, label_batch = tf.train.shuffle_batch(
    [image, label], 
    batch_size=FLAGS.batch_size, 
    num_threads=num_preprocess_threads, 
    capacity=FLAGS.min_queue_size + 3 * FLAGS.batch_size, 
    min_after_dequeue=FLAGS.min_queue_size) 

В конце эпохи, когда я делаю оценки (я уверен, что это верно и в обучении, но я не проверял), все блоки. Я понял, что в тот же момент внутренняя очередь партии в случайном порядке будет оставлена ​​с минимальными элементами min_after_dequeue. В это время в программе я бы идеально хотел просто удалить оставшиеся элементы, но я не уверен, как это сделать.

Очевидно, что этот тип блокировки внутри очередей TF может быть отключен, когда вы знаете, что больше нет элементов, связанных с методом .close(). Однако, поскольку основная очередь скрыта внутри функции, как я могу вызвать этот метод?

Answer 1

Вот код, который я в конце концов получил работу, хотя и с кучей предупреждения о том, что элементы, которые я сделал, были отменены.

lv = tf.constant(label_list) 

label_fifo = tf.FIFOQueue(len(filenames),tf.int32,shapes=[[]]) 
# if eval_data: 
    # num_epochs = 1 
# else: 
    # num_epochs = None 
file_fifo = tf.train.string_input_producer(filenames, shuffle=False, capacity=len(filenames)) 
label_enqueue = label_fifo.enqueue_many([lv]) 


reader = tf.WholeFileReader() 
result.key, value = reader.read(file_fifo) 
image = tf.image.decode_jpeg(value, channels=3) 
image.set_shape([128,128,3]) 
result.uint8image = image 
result.label = label_fifo.dequeue() 

images, label_batch = tf.train.shuffle_batch(
    [result.uint8image, result.label], 
    batch_size=FLAGS.batch_size, 
    num_threads=num_preprocess_threads, 
    capacity=FLAGS.min_queue_size + 3 * FLAGS.batch_size, 
    min_after_dequeue=FLAGS.min_queue_size) 

#in eval file: 
label_enqueue, images, labels = load_input.inputs() 
#restore from checkpoint in between 
coord = tf.train.Coordinator() 
try: 
    threads = [] 
    for qr in tf.get_collection(tf.GraphKeys.QUEUE_RUNNERS): 
    threads.extend(qr.create_threads(sess, coord=coord, daemon=True, 
            start=True)) 

    num_iter = int(math.ceil(FLAGS.num_examples/FLAGS.batch_size)) 
    true_count = 0 # Counts the number of correct predictions. 
    total_sample_count = num_iter * FLAGS.batch_size 

    sess.run(label_enqueue) 
    step = 0 
    while step < num_iter and not coord.should_stop(): 
    end_epoch = False 
    if step > 0: 
     for qr in tf.get_collection(tf.GraphKeys.QUEUE_RUNNERS): 
      #check if not enough elements in queue 
      size = qr._queue.size().eval() 
      if size - FLAGS.batch_size < FLAGS.min_queue_size: 
       end_epoch = True 
    if end_epoch: 
     #enqueue more so that we can finish 
     sess.run(label_enqueue) 
    #actually run step 
    predictions = sess.run([top_k_op]) 

Answer 2

Вы правы, что операция RandomShuffleQueue.close() остановит блокировку потоков, вызывающих блокировку, когда в очереди осталось меньше min_after_dequeue элементов.

Функция tf.train.shuffle_batch() создает tf.train.QueueRunner, которая выполняет операции в очереди в фоновом потоке. Если вы запустите его следующим образом, переходя к tf.train.Coordinator, вы сможете закрыть очередь чисто (на примере here):

sess = tf.Session() 
coord = tf.train.Coordinator() 
tf.train.start_queue_runners(sess, coord=coord) 

while not coord.should_stop(): 
    sess.run(train_op) 
# When done, ask the threads to stop. 
coord.request_stop() 
# And wait for them to actually do it. 
coord.join(threads) 

Answer 3

Существует дополнительный аргумент allow_smaller_final_batch

«allow_smaller_final_batch:.. (Необязательно) Boolean Если True, то дайте окончательный партии быть меньше, если нет достаточного количества предметов, оставленных в очереди»