使用不同的输入和相同的占位符运行并行操作

Question

我有必要同时计算多个精度。

correct_prediction = tf.equal(tf.argmax(y,1), tf.argmax(y_,1))
accuracy = tf.reduce_mean(tf.cast(correct_prediction, tf.float32))
print(sess.run(accuracy, feed_dict={x: mnist.test.images, y_: mnist.test.labels}))

这段代码与 TensorFlow 教程中的 mnist 示例相同，但没有：

W = tf.Variable(tf.zeros([784, 10]))
b = tf.Variable(tf.zeros([10]))

我有两个占位符，因为我已经计算并存储了它们。

W = tf.placeholder(tf.float32, [784, 10])
b = tf.placeholder(tf.float32, [10])

我想用我现有的值填充网络，然后计算准确度，这对于我加载的每个网络都必须发生。

因此，如果我加载 20 个网络，我想并行计算每个网络的准确度。会话运行有没有办法用不同的输入执行相同的操作？

Answer 1

您有多种选择可以让事情并行发生：

• 使用多个 python 线程/子进程进行并行化。 （请参阅 Python 的“多处理”库。）
• 将操作分批成单个更大的操作。 （例如，类似于同时对一批图像进行操作的图像操作https://www.tensorflow.org/api_docs/python/image/resizing#resize_bilinear。）
• 制作一个包含 20 个网络精度计算的图表。

我认为最后一个是最简单的，所以我在下面提供了一些示例代码来帮助您入门：

import tensorflow as tf

def construct_accuracy_calculation(i):
  W = tf.placeholder(tf.float32, [784, 10], name=("%d_W" % i))
  b = tf.placeholder(tf.float32, [10], name=("%d_b" % i))
  # ...
  correct_prediction = tf.equal(tf.argmax(y, 1), tf.argmax(y_, 1))
  accuracy = tf.reduce_mean(tf.cast(correct_prediction, tf.float32))
  return (W, b, accuracy)


def main():
  accuracy_computations = []
  feed_dict={}
  for i in xrange(NUM_NETWORKS):
    (W, b) = load_network(i)
    (W_op, b_op, accuracy) = construct_accuracy_calculation(i)
    feed_dict[W_op] = W
    feed_dict[b_op] = b
    accuracy_computations.append(accuracy)

  # sess = ...
  accuracy_values = sess.run(accuracy_computations, feed_dict=feed_dict)

if __name__ == "__main__":
  main()

Answer 2

并行化 TF 计算的一种方法是使用线程并行执行 run 调用（TF 与 multiprocessing 不兼容）。它比其他方法要复杂一些，因为您必须在 Python 端自己处理并行性。

这是一个示例，它在具有不同馈送输入的不同 Python 线程中的同一会话中运行相同的 matmul op，并且与 1 个线程相比，4 个线程的运行速度提高了大约 4 倍

import os, sys, queue, threading, time
import tensorflow as tf
import numpy as np

def p(s):
    # helper function for printing from multiple threads
    # need to append \n or results get intermixed in notebook
    print(s+"\n", flush=True, end="")


num_threads = 4
data_size = 32  # number of data points to enqueue
work_per_thread = data_size/num_threads
timeout = 10  # grace period for dequeing

input_queue = queue.Queue(data_size)
output_queue = queue.Queue(data_size)
dtype = np.float32

# use matrix vector matmul since it's compute intensive and uses single core
# see issue #6752
n = 16*1024
with tf.device("/cpu:0"):
    x = tf.placeholder(dtype)
    matrix = tf.Variable(tf.ones((n, n)))
    vector = tf.Variable(tf.ones((n, 1)))
    y = tf.matmul(matrix, vector)[0, 0] + x

# turn off graph-rewriting optimizations
sess = tf.Session(config=tf.ConfigProto(graph_options=tf.GraphOptions(optimizer_options=tf.OptimizerOptions(opt_level=tf.OptimizerOptions.L0))))
sess.run(tf.global_variables_initializer())

done = False
def runner(runner_id):
    p("Starting runner %s" % (runner_id,))
    count = 0
    while not done:
        try:
            x_val = input_queue.get(timeout=1)
        except queue.Empty:
            # retry on empty queue
            continue
        p("Start computing %d on %d" %(x_val, runner_id))
        out = sess.run(y, {x: x_val})
        count+=1
        output_queue.put(out)
        if count>=work_per_thread:
            break
    else:
        p("Stopping runner "+str(runner_id))

threads = []
print("Creating threads.")
for i in range(num_threads):
    t = threading.Thread(target=runner, args=(i,))
    threads.append(t)

for i in range(data_size):
    input_queue.put(i, timeout=timeout)

# start threads
p("Launching runners.")
start_time = time.time()
for t in threads:
    t.start()

p("Reading results.")
for i in range(data_size):
    try:
        p("Main thread: obtained %.2f" % (output_queue.get(timeout=timeout),))
    except queue.Empty:
        print("No results after %d, terminating computation."%(timeout,))
        break
else:
    p("Computed successfully.")
done = True

p("Waiting for threads to finish.")
for t in threads:
    t.join()
print("Done in %.2f seconds" %(time.time() - start_time))