Threading w rurociągu wejściowym tensorflow za

Question

Tło

Typowe wejście rurociąg w tensorflow wygląda następująco:

    tf.train.string_input_producer(list_of_filenames) 
         (creates queue of filenames) 
            | 
            \|/ 
      fixed length reader reads records from the files 
            | 
            \|/ 
    Read records are decoded and processed(eg if dealing with images then cropping,flipping etc) 
            | 
            \|/ 
      tf.train.shuffle_batch(tensors,num_threads) 
     (creates a shuffling queue and returns batches of tensors) 

Problem

Q1) nie istnieje argument NUM_THREADS w TF funkcji .train.string_input_producer(). Czy to oznacza, że ​​tylko pojedynczy wątek jest przeznaczony do odczytu nazw plików z kolejki nazw plików?

Q2) Jaki jest zakres argumentu num_threads funkcji tf.train.shuffle_batch() np. Liczba wymienionych tu wątków jest używana do odczytywania, dekodowania i przetwarzania plików lub są one używane do tworzenia partii tensorów?

Q3) Czy istnieje sposób drukowania, który wątek odczytuje nazwy plików lub zapisy z określonego pliku, np. Rodzaj zapisu pracy wykonanej przez każdy wątek?

Answer 1

Wszystkie operacje ładowania danych są wykonywane w ramach wykresu tensorflow, co należy zrobić, to uruchomić jeden lub więcej wątków do iteracji nad operacjami odczytu/kolejkowania. Tensorflow zapewnia klasę QueueRunner, która dokładnie to robi. Klasa Coordinator pozwala dość trywialnie zarządzać wątkami.

https://www.tensorflow.org/programmers_guide/threading_and_queues

To jest przykład kodu z linku powyżej:

# Create a queue runner that will run 4 threads in parallel to enqueue 
# examples. 
qr = tf.train.QueueRunner(queue, [enqueue_op] * 4) 

# Launch the graph. 
sess = tf.Session() 
# Create a coordinator, launch the queue runner threads. 
coord = tf.train.Coordinator() 
enqueue_threads = qr.create_threads(sess, coord=coord, start=True) 
# Run the training loop, controlling termination with the coordinator. 
for step in xrange(1000000): 
    if coord.should_stop(): 
     break 
    sess.run(train_op) 
# When done, ask the threads to stop. 
coord.request_stop() 
# And wait for them to actually do it. 
coord.join(enqueue_threads) 

Jeśli ładowali/przerób próbek poza wykresu (w swoim własnym kodzie, nie używając operacji TF), a następnie nie używałbyś QueueRunner, zamiast tego używałbyś swojej własnej klasy do kolejkowania danych za pomocą polecenia sess.run(enqueue_op, feed_dict={...}) w pętli.

Q1: Ilość wątków jest obsługiwane z: qr.create_threads(sess, coord=coord, start=True)

Q2: sesje TF są bezpieczne dla wątków, każde wywołanie tf.run(...) widzi spójną migawkę aktualnych zmiennych począwszy od kiedy zacząć. Twoje kolejki QueueRunner mogą uruchamiać dowolną liczbę wątków. Będą ustawiać się w kolejce w sposób bezpieczny dla wątków.

Q3: Nie używałem tf.train.string_input_producer siebie, ale myślę, że trzeba zwrócić tensora później na wykresie że dequeued dane, wystarczy dodać, że tensor do listy wniosków w sess.run([train_op, dequeue_op])