TensorFlow Multiple GPU and Reading data using Queue Runner Example

Posted on | In

#!/usr/local/bin/python
# -- coding: utf-8 --
‘’’3D convolutional neural network trained
to reduce the False Positive Rate for the LUNA datasets.
The LUNA datasets are organized in the CIFAR architecture.

Author: Kong Haiyang
‘’’
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import os
import sys
import time
import math
import numpy as np
from six.moves import xrange
import tensorflow as tf
import csv
import cv2

FLAGS = tf.app.flags.FLAGS
tf.app.flags.DEFINE_integer(‘IMAGE_SIZE’, 40, “Size of input Image.”)
tf.app.flags.DEFINE_integer(‘PIXEL_DATA_SIZE’, 4, “Size of Image pixel.”)
tf.app.flags.DEFINE_integer(‘CHANNEL_NUMBER’, 1, “Size of input Image.”)
tf.app.flags.DEFINE_integer(‘LABEL_NUMBER’, 2, “Label number.”)
tf.app.flags.DEFINE_integer(‘BATCH_SIZE’, 128, “Size of a Batch.”)
tf.app.flags.DEFINE_integer(‘NUM_EPOCHS’, 10, “Number of epochs.”)
tf.app.flags.DEFINE_integer(‘EVAL_BATCH_SIZE’, 64, “Size of an Evalution Batch.”)
tf.app.flags.DEFINE_integer(‘SEED’, 66478, “Seed of Shuffle.”)
tf.app.flags.DEFINE_string(‘TOWER_NAME’, ‘JP’, “Name of tower.”)
tf.app.flags.DEFINE_integer(‘NUM_GPU’, 2, “How many GPUs to use.”)
tf.app.flags.DEFINE_integer(‘NUM_PREPROCESS_THREADS’, 8,
“Number of preprocessing threads.”)
tf.app.flags.DEFINE_integer(‘NUM_LABEL’, 1, “How many Label bytes in a unit of Bin file.”)
tf.app.flags.DEFINE_integer(
‘NUM_IMAGE’, 40 ** 3, “How many Image bytes in a unit of Bin file.”)
tf.app.flags.DEFINE_integer(‘BYTE_LENGTH’, 4, “Byte length of label or image bytes.”)
tf.app.flags.DEFINE_string(
‘CSV_FILE’, ‘/home/kong/4T/official3D_110W/Shuffle.csv’, “Csv file path and name.”)
tf.app.flags.DEFINE_string(
‘BIN_FILE’, ‘/home/kong/4T/official3D_110W/shuffle3D64.bin’, “Bin file path and name.”)
tf.app.flags.DEFINE_string(‘XAVIER_INIT’,
‘tf.contrib.layers.xavier_initializer(seed=SEED)’,
“Initialize with XAVIER_INIT.”)

IMAGE_SIZE = FLAGS.IMAGE_SIZE
PIXEL_DATA_SIZE = FLAGS.PIXEL_DATA_SIZE
NUM_CHANNELS = FLAGS.CHANNEL_NUMBER
NUM_LABELS = FLAGS.LABEL_NUMBER
SEED = FLAGS.SEED
BATCH_SIZE = FLAGS.BATCH_SIZE
NUM_EPOCHS = FLAGS.NUM_EPOCHS
EVAL_BATCH_SIZE = FLAGS.EVAL_BATCH_SIZE
XAVIER_INIT = FLAGS.XAVIER_INIT
TOWER_NAME = FLAGS.TOWER_NAME
NUM_GPU = FLAGS.NUM_GPU
NUM_PREPROCESS_THREADS = FLAGS.NUM_PREPROCESS_THREADS
CSV_FILE = FLAGS.CSV_FILE
BIN_FILE = FLAGS.BIN_FILE
NUM_LABEL = FLAGS.NUM_LABEL
NUM_IMAGE = FLAGS.NUM_IMAGE
BYTE_LENGTH = FLAGS.BYTE_LENGTH
DTYPE = tf.float32

def readCSV(filename):
‘’’read lines from a csv file.
‘’’
lines = []
with open(filename, “rb”) as f:
csvreader = csv.reader(f)
for line in csvreader:
lines.append(line)
return lines

def get_noaug_first(CSV_FILE, BIN_FILE, no, count):
lines = readCSV(CSV_FILE)[1:]
data = np.empty([count, IMAGE_SIZE, IMAGE_SIZE, IMAGE_SIZE, 1], dtype=float)
labels = np.empty(count, dtype=int)
i = count_ = 0
length = (NUM_LABEL + NUM_IMAGE) 4
with open(BIN_FILE, ‘rb’) as f:
for line in lines:
if line[1] == str(no) and line[-1] == ‘0’ and line[-2] == ‘1’:
f.seek(i length)
buf = f.read(length)
data[count_, …] = (np.frombuffer(buf[4:], dtype=np.float32)).reshape(
IMAGE_SIZE, IMAGE_SIZE, IMAGE_SIZE, 1)
labels[count_] = np.frombuffer(buf[0:4], dtype=np.float32).astype(np.int64)
count_ += 1
i += 1
return data, labels

def init_bin_file(BIN_FILE):
bin_file_name = [BIN_FILE]
for f in bin_file_name:
if not tf.gfile.Exists(f):
raise ValueError(‘Failed to find file: ‘ + f)
fqb = tf.train.string_input_producer(bin_file_name, num_epochs=1)
record_bytes = (NUM_LABEL + NUM_IMAGE) * BYTE_LENGTH
rb = tf.FixedLengthRecordReader(record_bytes=record_bytes)
return fqb, rb

def init_csv_file(CSV_FILE):
csv_file_name = [CSV_FILE]
for f in csv_file_name:
if not tf.gfile.Exists(f):
raise ValueError(‘Failed to find file: ‘ + f)
fqc = tf.train.string_input_producer(csv_file_name, num_epochs=1)
rc = tf.TextLineReader(skip_header_lines=True)
return fqc, rc

def get_data_without_no(fqb, rb, fqc, rc, val_no, test_no):
def getBIN():
def getID():
key_raw, value = rc.read(fqc)
value_raw = tf.reshape(value, [1])
split_values = tf.string_split(value_raw, delimiter=’,’)
subsetid = tf.string_to_number(split_values.values[1], out_type=tf.int32)
return subsetid
key, value = rb.read(fqb)
record_bytes = tf.decode_raw(value, tf.float32)
label = tf.cast(tf.slice(record_bytes, [0], [NUM_LABEL]), tf.int64)
image = tf.reshape(tf.slice(record_bytes, [NUM_LABEL], [NUM_IMAGE]),
shape=[40, 40, 40, 1])
return getID(), label, image
subsetid, label, image = getBIN()
cond = lambda subsetid, label, image: tf.logical_or(tf.equal(subsetid, tf.constant(
val_no, dtype=tf.int32)), tf.equal(subsetid, tf.constant(test_no, dtype=tf.int32)))
doRead = lambda subsetid, label, image: getBIN()
result = tf.while_loop(cond, doRead, [subsetid, label, image])
return result

def get_data_with_no(fqb, rb, fqc, rc, no):
def getBIN():
def getID():
key_raw, value = rc.read(fqc)
value_raw = tf.reshape(value, [1])
split_values = tf.string_split(value_raw, delimiter=’,’)
subsetid = tf.string_to_number(split_values.values[1], out_type=tf.int32)
return subsetid
key, value = rb.read(fqb)
record_bytes = tf.decode_raw(value, tf.float32)
label = tf.cast(tf.slice(record_bytes, [0], [NUM_LABEL]), tf.int64)
image = tf.reshape(tf.slice(record_bytes, [NUM_LABEL], [NUM_IMAGE]),
shape=[40, 40, 40, 1])
return getID(), label, image
subsetid, label, image = getBIN()
cond = lambda subsetid, label, image: tf.not_equal(
subsetid, tf.constant(no, dtype=tf.int32))
doRead = lambda subsetid, label, image: getBIN()
result = tf.while_loop(cond, doRead, [subsetid, label, image])
return result

def get_noaug_with_no(fqb, rb, fqc, rc, no):
def getBIN():
def getID():
key_raw, value = rc.read(fqc)
value_raw = tf.reshape(value, [1])
split_values = tf.string_split(value_raw, delimiter=’,’)
subsetid = tf.string_to_number(split_values.values[1], out_type=tf.int32)
class_flag = tf.string_to_number(split_values.values[-2], out_type=tf.int32)
noaug = tf.string_to_number(split_values.values[-1], out_type=tf.int32)
return subsetid, class_flag, noaug
key, value = rb.read(fqb)
record_bytes = tf.decode_raw(value, tf.float32)
label = tf.cast(tf.slice(record_bytes, [0], [NUM_LABEL]), tf.int64)
image = tf.reshape(tf.slice(record_bytes, [NUM_LABEL], [NUM_IMAGE]),
shape=[40, 40, 40, 1])
subsetid, class_flag, noaug = getID()
return subsetid, class_flag, noaug, label, image
subsetid, class_flag, noaug, label, image = getBIN()
cond = lambda subsetid, class_flag, noaug, label, image: tf.logical_or(tf.not_equal(subsetid, tf.constant(no, dtype=tf.int32)),
tf.logical_or(tf.not_equal(class_flag, tf.constant(1, dtype=tf.int32)), tf.not_equal(noaug, tf.constant(0, dtype=tf.int32))))
doRead = lambda subsetid, class_flag, noaug, label, image: getBIN()
result = tf.while_loop(cond, doRead, [subsetid, class_flag, noaug, label, image])
return result

def get_train_data(fqb, rb, fqc, rc, val_no, test_no):
subsetid, label, image = get_data_without_no(fqb, rb, fqc, rc, val_no, test_no)
min_queue_examples = BATCH_SIZE 20
sis, labels, images = tf.train.batch(
[subsetid, label, image],
batch_size=BATCH_SIZE,
num_threads=NUM_PREPROCESS_THREADS,
capacity=min_queue_examples + 3 BATCH_SIZE)
labels = tf.reshape(labels, [-1])
return labels, images

def get_test_data(fqb, rb, fqc, rc, no):
subsetid, label, image = get_data_with_no(fqb, rb, fqc, rc, no)
min_queue_examples = BATCH_SIZE 20
sis, labels, images = tf.train.batch(
[subsetid, label, image],
batch_size=BATCH_SIZE,
num_threads=NUM_PREPROCESS_THREADS,
capacity=min_queue_examples + 3 BATCH_SIZE)
labels = tf.reshape(labels, [-1])
return labels, images

def get_noaug_data(fqb, rb, fqc, rc, no):
subsetid, class_flag, noaug, label, image = get_noaug_with_no(fqb, rb, fqc, rc, no)
min_queue_examples = BATCH_SIZE 20
sis, cfs, noaugs, labels, images = tf.train.batch(
[subsetid, class_flag, noaug, label, image],
batch_size=BATCH_SIZE // 10,
num_threads=NUM_PREPROCESS_THREADS,
capacity=min_queue_examples + 3 BATCH_SIZE)
labels = tf.reshape(labels, [-1])
return sis, cfs, noaugs, labels, images

def get_size(CSV_FILE):
ss = [0] 10
noaug = [0] 10
with open(CSV_FILE) as f:
csvreader = csv.reader(f)
for line in csvreader:
if line[0] != ‘candidateID’:
ss[int(line[1])] += 1
if line[-1] == ‘0’ and line[-2] == ‘1’:
noaug[int(line[1])] += 1
return ss, noaug

def _weight_on_cpu(name, shape):
with tf.device(‘/cpu:0’):
var = tf.get_variable(name, shape, DTYPE, tf.truncated_normal_initializer(stddev=0.1))
return var

def _bias_on_cpu(name, shape):
with tf.device(‘/cpu:0’):
var = tf.get_variable(name, shape, DTYPE, tf.constant_initializer(0.0))
return var

def model(data, isTraining, keep_prob):
with tf.variable_scope(‘conv1’) as scope:
W1 = _weight_on_cpu(‘W1’, [3, 3, 3, NUM_CHANNELS, 16])
conv = tf.nn.conv3d(data, W1, strides=[1, 1, 1, 1, 1], padding=’SAME’)
b1 = _bias_on_cpu(‘b1’, [16])
relu = tf.nn.relu(tf.nn.bias_add(conv, b1))
with tf.variable_scope(‘conv2’) as scope:
W2 = _weight_on_cpu(‘W2’, [3, 3, 3, 16, 24])
conv = tf.nn.conv3d(relu, W2, strides=[1, 1, 1, 1, 1], padding=’SAME’)
b2 = _bias_on_cpu(‘b2’, [24])
relu = tf.nn.relu(tf.nn.bias_add(conv, b2))
pool = tf.nn.max_pool3d(relu, ksize=[1, 2, 2, 2, 1],
strides=[1, 2, 2, 2, 1], padding=’VALID’)
with tf.variable_scope(‘conv3’) as scope:
W3 = _weight_on_cpu(‘W3’, [3, 3, 3, 24, 32])
conv = tf.nn.conv3d(pool, W3, strides=[1, 1, 1, 1, 1], padding=’SAME’)
b3 = _bias_on_cpu(‘b3’, [32])
relu = tf.nn.relu(tf.nn.bias_add(conv, b3))
pool = tf.nn.max_pool3d(relu, ksize=[1, 2, 2, 2, 1],
strides=[1, 2, 2, 2, 1], padding=’VALID’)
with tf.variable_scope(‘conv4’) as scope:
W4 = _weight_on_cpu(‘W4’, [3, 3, 3, 32, 48])
conv = tf.nn.conv3d(pool, W4, strides=[1, 1, 1, 1, 1], padding=’SAME’)
b4 = _bias_on_cpu(‘b4’, [48])
relu = tf.nn.relu(tf.nn.bias_add(conv, b4))
pool = tf.nn.max_pool3d(relu, ksize=[1, 2, 2, 2, 1],
strides=[1, 2, 2, 2, 1], padding=’VALID’)
with tf.variable_scope(‘conv5’) as scope:
W5 = _weight_on_cpu(‘W5’, [3, 3, 3, 48, 64])
conv = tf.nn.conv3d(pool, W5, strides=[1, 1, 1, 1, 1], padding=’SAME’)
b5 = _bias_on_cpu(‘b5’, [64])
relu = tf.nn.relu(tf.nn.bias_add(conv, b5))
pool = tf.nn.max_pool3d(relu, ksize=[1, 2, 2, 2, 1],
strides=[1, 2, 2, 2, 1], padding=’VALID’)
with tf.variable_scope(‘reshape’) as scope:
ps = pool.get_shape().as_list()
reshape = tf.reshape(pool, [-1, ps[1] ps[2] ps[3] ps[4]])
with tf.variable_scope(‘fc1’) as scope:
fcw1 = _weight_on_cpu(‘fcw1’, [2\*3 * 64, 32])
fcb1 = _bias_on_cpu(‘fcb1’, [32])
hidden = tf.nn.relu(tf.matmul(reshape, fcw1) + fcb1)

hidden = tf.cond(isTraining,
                 lambda: tf.nn.dropout(hidden, keep_prob),
                 lambda: hidden)

with tf.variable_scope(‘fc2’) as scope:
fcw2 = _weight_on_cpu(‘fcw2’, [32, NUM_LABELS])
fcb2 = _bias_on_cpu(‘fcb2’, [NUM_LABELS])
out = tf.add(tf.matmul(hidden, fcw2), fcb2)
return out

def loss(logits, labels):
“””Add L2Loss to all the trainable variables.
“””
cross_entropy = tf.nn.sparse_softmax_cross_entropy_with_logits(
logits, labels, name=’cross_entropy_per_example’)
cross_entropy_mean = tf.reduce_mean(cross_entropy, name=’cross_entropy’)
tf.add_to_collection(‘losses’, cross_entropy_mean)

return tf.add_n(tf.get_collection(‘losses’), name=’total_loss’)

def tower_loss(images, labels, isTraining, scope):
logits = model(images, isTraining, 0.5)
_ = loss(logits, labels)
losses = tf.get_collection(‘losses’, scope)
total_loss = tf.add_n(losses, name=’total_loss’)

loss_averages = tf.train.ExponentialMovingAverage(0.9, name=’avg’)
loss_averages_op = loss_averages.apply(losses + [total_loss])

with tf.control_dependencies([loss_averages_op]):
total_loss = tf.identity(total_loss)
return total_loss, logits

def average_gradients(tower_grads):
“””Calculate the average gradient for each shared variable across all towers.
Note that this function provides a synchronization point across all towers.
“””
average_grads = []
for grad_and_vars in zip(*tower_grads):
grads = []
for g, _ in grad_and_vars:
expanded_g = tf.expand_dims(g, 0)
grads.append(expanded_g)

grad = tf.concat(0, grads)
grad = tf.reduce_mean(grad, 0)
v = grad\_and\_vars\[0\]\[1\]
grad\_and\_var = (grad, v)
average\_grads.append(grad\_and_var)

return average_grads

def eval_in_batches(data, sess, eval_prediction, eval_data, isTraining):
size = data.shape[0]
if size < EVAL_BATCH_SIZE:
raise ValueError(“batch size for evals larger than dataset: %d” % size)

size -= size % NUM_GPU # comment this will cause tf exception?

predictions = np.ndarray(shape=(size, NUM_LABELS), dtype=np.float32)
for begin in xrange(0, size, EVAL_BATCH_SIZE):
end = begin + EVAL_BATCH_SIZE
if end <= size:
predict = sess.run(eval_prediction, feed_dict={
eval_data: data[begin:end, …], isTraining: False})
predictions[begin:end, :] = np.vstack(predict)
else:
batch_predictions = sess.run(eval_prediction, feed_dict={
eval_data: data[-EVAL_BATCH_SIZE:, …], isTraining: False})
predictions[begin:, :] = batch_predictions[begin - size:, :]
return predictions

def error_rate(predictions, labels):
“””Return the error rate based on dense predictions and sparse labels.”””
return 100.0 - (100.0 * np.sum(np.argmax(predictions, 1) == labels) /
predictions.shape[0])

def lunaTrain(VIEW_DIRECTORY, imgName, csvName, ss_list, noaug_list):
with tf.Graph().as_default(), tf.device(‘/cpu:0’):
for cross in range(10):
sssstttt = time.time()
print(‘Cross {}…’.format(cross))
WORK_DIRECTORY = os.path.join(VIEW_DIRECTORY, ‘Cross{}’.format(cross))
testNo = cross
valNo = (cross + 1) % 10
st = time.time()
train_size = sum(ss_list) - ss_list[testNo] - ss_list[valNo]
test_no_aug_data, test_no_aug_label = get_noaug_first(
CSV_FILE, BIN_FILE, testNo, noaug_list[testNo])
val_no_aug_data, val_no_aug_label = get_noaug_first(
CSV_FILE, BIN_FILE, valNo, noaug_list[valNo])
print(‘Reading no aug data finished in {:.2f} seconds…’.format(time.time() - st))
st = time.time()
fqbt, rbt = init_bin_file(BIN_FILE)
fqct, rct = init_csv_file(CSV_FILE)
fqbe, rbe = init_bin_file(BIN_FILE)
fqce, rce = init_csv_file(CSV_FILE)
data_node = tf.placeholder(tf.float32, shape=(
None, IMAGE_SIZE, IMAGE_SIZE, IMAGE_SIZE, NUM_CHANNELS))
labels_node = tf.placeholder(tf.int64, shape=(None,))
isTraining = tf.placeholder(tf.bool, name=’isTrain’)

batch = tf.Variable(0, trainable=False)
learning\_rate = tf.train.exponential\_decay(0.01, batch * BATCH\_SIZE, train\_size / 5,
                                           0.95, staircase=True)
opt = tf.train.MomentumOptimizer(learning_rate, 0.9)

split\_images = tf.split(0, NUM\_GPU, data_node)
split\_labels = tf.split(0, NUM\_GPU, labels_node)

eval_predictions = \[\]
tower_grads = \[\]
for i in xrange(NUM_GPU):
  with tf.device('/gpu:%d' % i):
    with tf.name\_scope('%s\_%d' % (TOWER_NAME, i)) as scope:
      loss, logits = tower\_loss(split\_images\[i\], split_labels\[i\], isTraining, scope)
      predictions = tf.nn.softmax(logits)
      eval_predictions.append(predictions)
      tf.get\_variable\_scope().reuse_variables()

      grads = opt.compute_gradients(loss)
      tower_grads.append(grads)

grads = average\_gradients(tower\_grads)
apply\_gradient\_op = opt.apply_gradients(grads)

variable_averages = tf.train.ExponentialMovingAverage(0.9999)
variables\_averages\_op = variable\_averages.apply(tf.trainable\_variables())

train\_op = tf.group(apply\_gradient\_op, variables\_averages_op)

train\_label\_node, train\_data\_node = get\_train\_data(
    fqbt, rbt, fqct, rct, valNo, testNo)
val\_label\_node, val\_data\_node = get\_test\_data(fqbe, rbe, fqce, rce, valNo)
test\_label\_node, test\_data\_node = get\_test\_data(fqbe, rbe, fqce, rce, testNo)

saver = tf.train.Saver(tf.all_variables())

TRAIN\_FREQUENCY = train\_size // BATCH_SIZE // 20
VAL\_FREQUENCY = train\_size // BATCH_SIZE
TEST\_FREQUENCY = train\_size // BATCH_SIZE * 5

config = tf.ConfigProto(allow\_soft\_placement=True, log\_device\_placement=False)
with tf.Session(config=config) as sess:
  summary\_writer = tf.train.SummaryWriter(WORK\_DIRECTORY, sess.graph)
  sess.run(tf.initialize\_local\_variables())
  sess.run(tf.initialize\_all\_variables())
  coord = tf.train.Coordinator()
  threads = tf.train.start\_queue\_runners(sess=sess, coord=coord)
  try:
    while not coord.should_stop():
      start_time = time.time()
      for step in xrange(int(NUM\_EPOCHS * train\_size) // BATCH_SIZE):
        train\_data, train\_label = sess.run(\[train\_data\_node, train\_label\_node\])
        feed\_dict = {data\_node: train_data,
                     labels\_node: train\_label, isTraining: True}
        _, l, lr = sess.run(\[train\_op, loss, learning\_rate\], feed\_dict=feed\_dict)
        if step != 0 and step % TRAIN_FREQUENCY == 0:
          et = time.time() - start_time
          print('Step %d (epoch %.2f), %.1f ms' %
                (step, float(step) * BATCH\_SIZE / train\_size, 1000 * et / TRAIN_FREQUENCY))
          print('Minibatch loss: %.3f, learning rate: %.6f' % (l, lr))
          start_time = time.time()
        if step != 0 and VAL\_FREQUENCY != 0 and step % VAL\_FREQUENCY == 0:
          val\_data, val\_label = sess.run(\[val\_data\_node, val\_label\_node\])
          valE = error\_rate(eval\_in_batches(
              val\_data, sess, eval\_predictions, data\_node, isTraining), val\_label)
          print('Validation error: %.3f%%' % valE)
          valPE = error\_rate(eval\_in_batches(
              val\_no\_aug\_data, sess, eval\_predictions, data\_node, isTraining), val\_no\_aug\_label)
          print('Validation error of no aug Positive: %.3f%%' % valPE)
          val_data = 0
          start_time = time.time()
        if step != 0 and TEST\_FREQUENCY != 0 and step % TEST\_FREQUENCY == 0:
          test\_data, test\_label = sess.run(\[test\_data\_node, test\_label\_node\])
          test\_error = error\_rate(eval\_in\_batches(
              test\_data, sess, eval\_predictions, data\_node, isTraining), test\_label)
          print('Test error: %.3f%%' % test_error)
          test\_errorP = error\_rate(eval\_in\_batches(
              test\_no\_aug\_data, sess, eval\_predictions, data\_node, isTraining), test\_no\_aug\_label)
          print('Test error of no aug Positive: %.3f%%' % test_errorP)
          test_data = 0
          start_time = time.time()
      else:
        checkpoint\_path = os.path.join(WORK\_DIRECTORY, 'model.ckpt')
        saver.save(sess, checkpoint\_path, global\_step=step)
  except tf.errors.OutOfRangeError:
    print('Done training -- epoch limit reached')
  finally:
    coord.request_stop()
  coord.join(threads)
print('All costs {:.2f} seconds...'.format(time.time() - sssstttt))
train\_data = val\_data = test_data = 0
train\_labels = val\_labels = test_labels = 0

def main(_):
viewPath = ‘/home/kong/4T/official3D_110W’
csvName = ‘/home/kong/4T/official3D_110W/Shuffle.csv’
imgName = ‘/home/kong/4T/official3D_110W/shuffle3D.bin’

ss_list, noaug_list = get_size(CSV_FILE)
lunaTrain(viewPath, imgName, csvName, ss_list, noaug_list)

if __name__ == ‘__main__‘:
tf.app.run()