一. 通過 ConfigProto 設(shè)置多線程

（具體參數(shù)功能及描述見 tensorflow/core/protobuf/config.proto）

在進(jìn)行 tf.ConfigProto() 初始化時(shí)，可以通過設(shè)置相應(yīng)的參數(shù)，來控制每個(gè)操作符 op 并行計(jì)算的線程個(gè)數(shù)或 session 線程池的線程數(shù)。主要涉及的參數(shù)有以下三個(gè):

1. intra_op_parallelism_threads 控制運(yùn)算符op內(nèi)部的并行
當(dāng)運(yùn)算符 op 為單一運(yùn)算符，并且內(nèi)部可以實(shí)現(xiàn)并行時(shí)，如矩陣乘法，reduce_sum 之類的操作，可以通過設(shè)置 intra_op_parallelism_threads 參數(shù)來并行。

2. inter_op_parallelism_threads 控制多個(gè)運(yùn)算符op之間的并行計(jì)算
當(dāng)有多個(gè)運(yùn)算符 op，并且他們之間比較獨(dú)立，運(yùn)算符和運(yùn)算符之間沒有直接的路徑 Path 相連。Tensorflow會(huì)嘗試并行地計(jì)算他們，使用由 inter_op_parallelism_threads 參數(shù)來控制數(shù)量的一個(gè)線程池。
在第一次創(chuàng)建會(huì)話將設(shè)置將來所有會(huì)話的線程數(shù)，除非是配置了 session_inter_op_thread_pool 選項(xiàng)。

3. session_inter_op_thread_pool 配置會(huì)話線程池。
如果會(huì)話線程池的 num_threads 為 0，使用 inter_op_parallelism_threads 選項(xiàng)。

二. 通過隊(duì)列進(jìn)行數(shù)據(jù)讀取時(shí)設(shè)置多線程

（具體函數(shù)功能及描述見 tensorflow/python/training/input.py）

1. 通過以下函數(shù)進(jìn)行樣本批處理時(shí)，可以通過設(shè)置 num_threads 來設(shè)置單個(gè) Reader 多線程讀取

1) batch(tensors, batch_size, num_threads=1, capacity=32,
enqueue_many=False, shapes=None, dynamic_pad=False,
allow_smaller_final_batch=False, shared_name=None, name=None)

2) maybe_batch(tensors, keep_input, batch_size, num_threads=1, capacity=32,
enqueue_many=False, shapes=None, dynamic_pad=False,
allow_smaller_final_batch=False, shared_name=None, name=None)

3) shuffle_batch(tensors, batch_size, capacity, min_after_dequeue,
num_threads=1, seed=None, enqueue_many=False, shapes=None,
allow_smaller_final_batch=False, shared_name=None, name=None)

4) maybe_shuffle_batch(tensors, batch_size, capacity, min_after_dequeue,
keep_input, num_threads=1, seed=None,
enqueue_many=False, shapes=None,
allow_smaller_final_batch=False, shared_name=None,
name=None)

例：

import tensorflow as tf 
 
 
filenames = ['A.csv', 'B.csv', 'C.csv'] 
# 生成一個(gè)先入先出隊(duì)列和一個(gè) QueueRunner，生成文件名隊(duì)列 
filename_queue = tf.train.string_input_producer(filenames, shuffle=False) 
 
 
# 定義 Reader 和 Decoder
reader = tf.TextLineReader() 
key, value = reader.read(filename_queue) 
example, label = tf.decode_csv(value, record_defaults=[['null'], ['null']])
 
 
# 使用tf.train.batch() 會(huì)為 graph 添加一個(gè)樣本隊(duì)列和一個(gè) QueueRunner。 
# 經(jīng)過 Reader 讀取文件和 Decoder 解碼后數(shù)據(jù)會(huì)進(jìn)入這個(gè)隊(duì)列，再批量出隊(duì)。
# tf.train.batch() 這里只有一個(gè) Reader，可以設(shè)置多線程
 
example_batch, label_batch = tf.train.batch([example, label], batch_size=5) 
 
 
with tf.Session() as sess: 
  coord = tf.train.Coordinator() 
  threads = tf.train.start_queue_runners(coord=coord) 
  for i in range(10): 
    e_val,l_val = sess.run([example_batch,label_batch]) 
    print e_val,l_val 
  coord.request_stop() 
  coord.join(threads) 

2. 通過以下函數(shù)進(jìn)行樣本批處理時(shí)，可以通過設(shè)置 Decoder 和 Reader 的個(gè)數(shù)來設(shè)置多 Reader 讀取，其中每個(gè) Reader 使用一個(gè)線程

1) batch_join(tensors_list, batch_size, capacity=32, enqueue_many=False,
shapes=None, dynamic_pad=False, allow_smaller_final_batch=False,
shared_name=None, name=None):

2) maybe_batch_join(tensors_list, keep_input, batch_size, capacity=32,
enqueue_many=False, shapes=None, dynamic_pad=False,
allow_smaller_final_batch=False, shared_name=None,
name=None)

3) shuffle_batch_join(tensors_list, batch_size, capacity,
min_after_dequeue, seed=None, enqueue_many=False,
shapes=None, allow_smaller_final_batch=False,
shared_name=None, name=None)

4) maybe_shuffle_batch_join(tensors_list, batch_size, capacity,
min_after_dequeue, keep_input, seed=None,
enqueue_many=False, shapes=None,
allow_smaller_final_batch=False, shared_name=None,
name=None)

例：

import tensorflow as tf 
 
 
filenames = ['A.csv', 'B.csv', 'C.csv'] 
# 生成一個(gè)先入先出隊(duì)列和一個(gè) QueueRunner，生成文件名隊(duì)列
filename_queue = tf.train.string_input_producer(filenames, shuffle=False) 
 
 
# 定義 Reader
reader = tf.TextLineReader() 
key, value = reader.read(filename_queue) 
 
 
#定義了多個(gè) Decoder, 每個(gè) Decoder 跟一個(gè) Reader 相連, 即有多個(gè) Reader
example_list = [tf.decode_csv(value, record_defaults=[['null'], ['null']])
         for _ in range(2)] # Decoder 和 Reader 為 2
   
# 使用tf.train.batch_join() 會(huì)為 graph 添加一個(gè)樣本隊(duì)列和一個(gè) QueueRunner。 
# 經(jīng)過多個(gè) Reader 讀取文件和 Decoder 解碼后數(shù)據(jù)會(huì)進(jìn)入這個(gè)隊(duì)列，再批量出隊(duì)。   
# 使用 tf.train.batch_join(), 可以使用多個(gè) Reader 并行讀取數(shù)據(jù)。每個(gè) Reader 使用一個(gè)線程
example_batch, label_batch = tf.train.batch_join(example_list, batch_size=5) 
with tf.Session() as sess: 
  coord = tf.train.Coordinator() 
  threads = tf.train.start_queue_runners(coord=coord) 
  for i in range(10): 
    e_val,l_val = sess.run([example_batch,label_batch]) 
    print e_val,l_val 
  coord.request_stop() 
  coord.join(threads) 

以上這篇Tensorflow 多線程設(shè)置方式就是小編分享給大家的全部內(nèi)容了，希望能給大家一個(gè)參考，也希望大家多多支持腳本之家。

最新評(píng)論