Dequeue sofort nach dem Starten von Threads fehlgeschlagen

Question

So habe ich gelesen, wie Warteschlangen und Auspacken funktioniert in Tensorflow, und ich habe herum mit dem Erstellen einer Warteschlange Dateiname und ziehen daraus.

Leider bekomme ich einen Fehler, wenn ich versuche, sofort nach dem Starten der Threads aus der Warteschlange zu entfernen. Gibt es einen Grund dafür? Wenn ich einen 1-Sekunden-Sleep-Timer einlege, wird er aus der Warteschlange genommen. Wenn nicht, es funktioniert manchmal, aber oft wird eine Ausnahme (siehe unten Code) wirft

import tensorflow as tf 
import time 
with tf.Graph().as_default(): 
    filename_list = ['data_batch_{}.mat'.format(i+1) for i in range(5)] 
    filename_queue = tf.train.string_input_producer(filename_list) 

    with tf.Session() as sess: 
     coord = tf.train.Coordinator() 
     threads = tf.train.start_queue_runners(sess=sess, coord=coord) 

     #time.sleep(1) # If I uncomment this it works 
     for i in range(5): 
      print(sess.run(filename_queue.dequeue())) 

     coord.request_stop() 
     coord.join(threads) 

Und die Ausnahme geworfen:

--------------------------------------------------------------------------- 
NotFoundError        Traceback (most recent call last) 
<ipython-input-28-cf6ab7b71f22> in <module>() 
    10   #time.sleep(1) 
    11   for i in range(5): 
---> 12    print(sess.run(filename_queue.dequeue())) 
    13 
    14   coord.request_stop() 

/usr/local/lib/python2.7/dist-packages/tensorflow/python/client/session.pyc in run(self, fetches, feed_dict, options, run_metadata) 
    331  try: 
    332  result = self._run(None, fetches, feed_dict, options_ptr, 
--> 333       run_metadata_ptr) 
    334  if run_metadata: 
    335   proto_data = tf_session.TF_GetBuffer(run_metadata_ptr) 

/usr/local/lib/python2.7/dist-packages/tensorflow/python/client/session.pyc in _run(self, handle, fetches, feed_dict, options, run_metadata) 
    571  try: 
    572  results = self._do_run(handle, target_list, unique_fetches, 
--> 573        feed_dict_string, options, run_metadata) 
    574  finally: 
    575  # The movers are no longer used. Delete them. 

/usr/local/lib/python2.7/dist-packages/tensorflow/python/client/session.pyc in _do_run(self, handle, target_list, fetch_list, feed_dict, options, run_metadata) 
    646  if handle is None: 
    647  return self._do_call(_run_fn, self._session, feed_dict, fetch_list, 
--> 648       target_list, options, run_metadata) 
    649  else: 
    650  return self._do_call(_prun_fn, self._session, handle, feed_dict, 

/usr/local/lib/python2.7/dist-packages/tensorflow/python/client/session.pyc in _do_call(self, fn, *args) 
    666   except KeyError: 
    667   pass 
--> 668  raise type(e)(node_def, op, message) 
    669 
    670 def _extend_graph(self): 

NotFoundError: FetchOutputs node input_producer_Dequeue:0: not found 

Answer 1

Danke, dass wir über dieses Problem wissen. Ich habe eine entsprechende GitHub issue eingereicht und eine Reparatur vorbereitet, die bald im Repository erscheinen sollte.

In der Zwischenzeit sollte der folgende Code funktioniert, indem eine einzige dequeue() op Erstellen, bevor die Sitzung starten:

import tensorflow as tf 
with tf.Graph().as_default(): 
    filename_list = ['data_batch_{}.mat'.format(i+1) for i in range(5)] 
    filename_queue = tf.train.string_input_producer(filename_list) 
    dequeued_t = filename_queue.dequeue() 

    with tf.Session() as sess: 
     coord = tf.train.Coordinator() 
     threads = tf.train.start_queue_runners(sess=sess, coord=coord) 

     for i in range(5): 
      print(sess.run(dequeued_t)) 

     coord.request_stop() 
     coord.join(threads)