2018-08-13multi-threads

settings: pretreatment gonna slower the training proceedings

multithreads.png

queue

#create
tf.FIFOQueue(num_elements, dtype)
tf.RandomShuffleQueue()
#initialize
tf.enqueue_many([])
#out
tf.dequeue()
#in
tf.enqueue([])

tf.Coordinator and tf.QueueRunner

#enable the thread
tf.Coordinator()
# the work thread will do
def Loop(coord,thread_id)
# check and ask whether it should be stoped
     while not coord.should_stop():
         if ...
            coord.request_stop()
         else:
            print(thread_id)
         time.sleep(1)
coord=tf.train.Coordinator()
threads=[
threading.Thread(target=Loop,args=[coord,i,]))  for i in xrange(5)
for t in threads: t,start()
coord.join(threads)

manage the queue

queue= tf.FIFQueue(100,'float')
enqueue_op=queue.enqueue([tf.random_normal([1])])
#start 5 threads to execute the enqueue_op
qr=tf.train.QueueRunner(queue,[enqueue_op*5])
# add the QueueRunner to the specified collection on the graph, if no,then default
tf.train.add_queue_runner(qr)
out_tensor=queue.dequeue()
with tf.session() as sess:
       coord=tf.train.Coordinator()
       threads=tf.train.start_queue_runners(sess=sess,coord=coord)
       for _ in range(3): print sess.run(out_tensor) [0]
       coord.request_stop ()
       coord_join(threads)

input_file queue

tf.train.match_filenames_once()#get the files matched
tf.train.string_input_producer()#generate the input_file queue

files=tf.train.match_filenames_once('path_regex')
filename_queue=tf.train.string_input_producer(files,shuffle=False)
reader=tf.TFRecorder()
_,serialized_example=reader.read(filename_queue)
features=tf.parse_single_example(
serialized_example,
features={
'': tf.FixedLenFeature([],dtype)
...
}
with tf.session() as sess:
       tf.initialize_all_variables().run()
       print(sess.run(files))
       coord=tf.train.Coordinator()
       threads=tf.train.start_queue_runners(sess=sess,coord=coord)
       for i in range(6):
            print(sess.run([features['i'],features['j']]))
       coord.request_stop()
       coord.join(threads)
)

batching

tf.train.batch()
tf.train.shuffle_batch()

input_data processing framework

framework.png