python – 链接多个队列时的Tensorflow Race条件

2019年7月20日 121次阅读

我想以多线程的方式计算一组图像的每个RGB通道的平均值.

我的想法是让一个string_input_producer填充一个filename_queue,然后让第二个FIFOQueue由num_threads线程填充,这些线程从filename_queue中的文件名加载图像,对它们执行一些操作,然后将结果排入队列.

然后,由一个线程(主线程)访问第二个队列,该线程汇总队列中的所有值.

这是我的代码：

# variables for storing the mean and some intermediate results
mean = tf.Variable([0.0, 0.0, 0.0])
total = tf.Variable(0.0)

# the filename queue and the ops to read from it
filename_queue = tf.train.string_input_producer(filenames, num_epochs=1)
reader = tf.WholeFileReader()
_, value = reader.read(filename_queue)
image = tf.image.decode_jpeg(value, channels=3)
image = tf.cast(image, tf.float32)

sum = tf.reduce_sum(image, [0, 1])
num = tf.mul(tf.shape(image)[0], tf.shape(image)[1])
num = tf.cast(num, tf.float32)

# the second queue and its enqueue op
queue = tf.FIFOQueue(1000, dtypes=[tf.float32, tf.float32], shapes=[[3], []])
enqueue_op = queue.enqueue([sum, num])

# the ops performed by the main thread
img_sum, img_num = queue.dequeue()
mean_op = tf.add(mean, img_sum)
total_op = tf.add(total, img_num)

# adding new queue runner that performs enqueue_op on num_threads threads
qr = tf.train.QueueRunner(queue, [enqueue_op] * num_threads)
tf.train.add_queue_runner(qr)

init_op = tf.initialize_all_variables()

sess = tf.Session()
sess.run(init_op)
coord = tf.train.Coordinator()
threads = tf.train.start_queue_runners(sess=sess, coord=coord)

# the main loop being executed until the OutOfRangeError 
# (when filename_queue does not yield elements anymore)
try:
    while not coord.should_stop():
        mean, total = sess.run([mean_op, total_op])

except tf.errors.OutOfRangeError:
    print 'All images processed.'
finally:
    coord.request_stop()

coord.join(threads)

# some additional computations to get the mean
total_3channel = tf.pack([total, total, total])
mean = tf.div(mean, total_3channel)
mean = sess.run(mean)
print mean

问题是每次我运行这个函数我得到不同的结果,例如：

[ 99.35347748  58.35261154  44.56705856]
[ 95.91153717  92.54192352  87.48269653]
[ 124.991745    121.83417511  121.1891861 ]

我把这归咎于竞争条件.但那些竞争条件来自哪里？有人可以帮我吗？

最佳答案 您的QueueRunner将启动num_threads线程,这些线程将竞争访问您的阅读器并将结果推送到队列中.队列中图像的顺序将根据首先完成的线程而有所不同.

2月12日更新

链接两个队列并从第二个队列中汇总值的简单示例.使用num_threads时> 1,中间值有一些非确定性,但最终值总是30.当num_threads = 1时,一切都是确定的

tf.reset_default_graph()

queue_dtype = np.int32

# values_queue is a queue that will be filled with 0,1,2,3,4
# range_input_producer creates the queue and registers its queue_runner
value_queue = tf.range_input_producer(limit=5, num_epochs=1, shuffle=False)
value = value_queue.dequeue()

# value_squared_queue will be filled with 0,1,4,9,16
value_squared_queue = tf.FIFOQueue(capacity=50, dtypes=queue_dtype)
value_squared_enqueue = value_squared_queue.enqueue(tf.square(value))
value_squared = value_squared_queue.dequeue()

# value_squared_sum keeps running sum of squares of values 
value_squared_sum = tf.Variable(0)
value_squared_sum_update = value_squared_sum.assign_add(value_squared)

# register queue_runner in the global queue runners collection
num_threads = 2
qr = tf.train.QueueRunner(value_squared_queue, [value_squared_enqueue] * num_threads)
tf.train.queue_runner.add_queue_runner(qr)

sess = tf.InteractiveSession()
sess.run(tf.initialize_all_variables())
tf.start_queue_runners()

for i in range(5):
  sess.run([value_squared_sum_update])
  print sess.run([value_squared_sum])

你应该看到：

[0]
[1]
[5]
[14]
[30]

或者有时(当翻转前2个值的顺序时)

[1]
[1]
[5]
[14]
[30]
点赞