Enqueing und Threading-Fehler mit Tensorflow beim Abrufen von Daten

Question

Ich verwende Tensorflow für die Feinabstimmung eines Modells und verwendet Threads zum Einreihen der Warteschlange Warteschlange. Der Code ist für mehrere 1000 Iterationen in Ordnung, aber gibt nach einigen Iterationen folgende Fehler weiter. Der Code wird jedoch normalerweise nach mehreren Iterationen mit einigen dieser Fehler beendet, und ich muss ihn manuell neu starten. Könnten Sie mir bitte helfen, den Fehler zu verstehen?

Exception in thread Thread-29: 
Traceback (most recent call last): 
    File "/usr/lib/python2.7/threading.py", line 810, in __bootstrap_inner 
    self.run() 
    File "/usr/lib/python2.7/threading.py", line 763, in run 
    self.__target(*self.__args, **self.__kwargs) 
    File "data_feeder.py", line 212, in enqueue_op_online_val 
    model_vars['labels_val']: labels_minibatch}) 
    File "/usr/local/lib/python2.7/dist-packages/tensorflow/python/client/session.py", line 717, in run 
    run_metadata_ptr) 
    File "/usr/local/lib/python2.7/dist-packages/tensorflow/python/client/session.py", line 915, in _run 
    feed_dict_string, options, run_metadata) 
    File "/usr/local/lib/python2.7/dist-packages/tensorflow/python/client/session.py", line 965, in _do_run 
    target_list, options, run_metadata) 
    File "/usr/local/lib/python2.7/dist-packages/tensorflow/python/client/session.py", line 985, in _do_call 
    raise type(e)(node_def, op, message) 
NotFoundError: PruneForTargets: Some target nodes not found: fifo_queue_enqueue/fifo_queue_enqueue_4/fifo_queue_enqueue_10/fifo_queue_enqueue_10/fifo_queue_enqueue_8/fifo_queue_enqueue_2/fifo_queue_enqueue_4/fifo_queue_enqueue/fifo_queue_enqueue_3/fifo_queue_enqueue_3/fifo_queue_enqueue/fifo_queue_enqueue_1/fifo_queue_enqueue_5/fifo_queue_enqueue/fifo_queue_enqueue_2/fifo_queue_enqueue/fifo_queue_enqueue_4/fifo_queue_enqueue_6/fifo_queue_enqueue_20/fifo_queue_enqueue_29/fifo_queue_enqueue/fifo_queue_enqueue_7/fifo_queue_enqueue_8/fifo_queue_enqueue_9/fifo_queue_enqueue_7/fifo_queue_enqueue/fifo_queue_enqueue_9/fifo_queue_enqueue_8/fifo_queue_enqueue_11/fifo_queue_enqueue_5/fifo_queue_enqueue_7/fifo_queue_enqueue_8/fifo_queue_enqueue_12/fifo_queue_enqueue_3/fifo_queue_enqueue_2/fifo_queue_enqueue_5/fifo_queue_enqueue_4/fifo_queue_enqueue_7/fifo_queue_enqueue_3/fifo_queue_enqueue/fifo_queue_enqueue_9/fifo_queue_enqueue_1/fifo_queue_enqueue_25/fifo_queue_enqueue_23/fifo_queue_enqueue_30/fifo_queue_enqueue_5/fifo_queue_enqueue_3/fifo_queue_enqueue_7/fifo_queue_enqueue_3/fifo_queue_enqueue/fifo_queue_enqueue_2/fifo_queue_enqueue_8/fifo_queue_enqueue_11/fifo_queue_enqueue_11/fifo_queue_enqueue/fifo_queue_enqueue_4 

Hier ist der enqueue op

def enqueue_op(self, sess, model_vars, coord): 
     queue = model_vars['queue_train'] 
     random.shuffle(self.keys) 
     img_minibatch = np.zeros((self.batch_size, 224, 224, 3)) 
     #t = time.time() 

     for indx in xrange(self.num_batches): 
      labels_minibatch = [] 
      # Load all files in this batch 
      for i,k in\ 
      enumerate(self.keys[indx*self.batch_size:(indx+1)*self.batch_size]): 
       img = misc.imread(self.img_path + self.data[k]['filename']) 
       img_minibatch[i, :, :, :] = img_proc(img) 
       labels_minibatch.append(self.data[k]['labels']) 

      labels_minibatch =\ 
      self.mlb.fit_transform(labels_minibatch).astype(float) 
      sess.run([queue.enqueue([model_vars['input'], 
            model_vars['labels']])],\ 
        feed_dict={model_vars['input']: img_minibatch, 
           model_vars['labels']: labels_minibatch}) 

      if coord.should_stop(): 
       break 

Hier werden die Thread-Erzeugung Schnipsel

sind

thr_train = [] 
for i in xrange(12):    

thr_train.append(threading.Thread(target=primary_mb_feeder.enqueue_op, 
             args=(sess, model_vars, coord))) 
       thr_train[-1].setDaemon(True) 
       thr_train[-1].start() 

Answer 1

Es sieht aus wie das Problem durch die enqueue_op() Verfahren verursacht wird, der (i) mehrere in läuft Threads, und (ii) erstellt einen neuen tf.Operation für jede Minibatch, durch den Aufruf queue.enqueue(). TensorFlow unterstützt nicht mehrere Threads, die gleichzeitig Knoten zum Graphen hinzufügen, und dies ist im Allgemeinen ineffizient (TensorFlow arbeitet am effizientesten, wenn sich das Diagramm nicht ändert).

Um dieses Problem zu beheben, sollten Sie Ihren Code so umgestalten, dass Sie nur queue.enqueue() einmal aufrufen, den zurückgegebenen Vorgang zwischen allen Eingabe-Threads freigeben und in jedem der neu eingehenden sess.run()-Aufrufe verwenden Daten.