Robin on Linux

Yesterday I wrote a Tensorflow program to train CIFAR100 dataset with Resnet-50 model. But when the training begin, I saw the ‘loss’ of classification is abnormally big and didn’t reduce at all:

loss[2.6032338e+25]
loss[2.5617402e+25]
loss[3.3851871e+25]
loss[3.092054e+25]
...

Firstly, I thought the code for processing dataset may be wrong. But after print out the data in console, the loading input data seems all right. Then I print all the value of tensors right after initialization of model. And these value seems correct either.
Without other choices, I began to check the initializer in Tensorflow code:

    with slim.arg_scope([slim.conv2d],
                        weights_initializer = tf.truncated_normal_initializer(mean = 0, stddev = 0.1)):
      img_inf, _ = resnet_v2.resnet_v2_50(image, NUM_CLASSES)

If the loss is too big, maybe I could decrease the initial value of tensors in model? Then I change ‘mean’ from ‘0’ to ‘0.1’ for ‘slim.conv2d’:

                        weights_initializer = tf.truncated_normal_initializer(mean = 0.001, stddev = 1)):

But the loss seems more crazy:

loss[1.1468245e+29]
loss[1.6610325e+29]
loss[1.1840615e+29]
...

I have to change ‘mean’ and ‘stddev’ again:

                        weights_initializer = tf.truncated_normal_initializer(mean = 0.001, stddev = 1)):

This time, the loss seems correct now.

loss[1215.8724]
loss[1023.67676]
loss[583.6274]
...

This is the first time I saw that initialized value could make the training accuracy so different.

In tensorflow, what should we do if we want run something before fetching data (such as, using queue in tensorflow)? Here is an example tested by myself:

import tensorflow as tf
from tensorflow.python.framework import ops
def my_func(x):
    print("hello")
    return x
queue = tf.FIFOQueue(10, "float32")
init  = queue.enqueue_many(([1, 2, 3],))
my_op = tf.py_func(my_func, [1.0], tf.float32)
with ops.control_dependencies([my_op]):
    inc = queue.enqueue([1],)
with tf.Session() as sess:
    sess.run(init)
    print("Before Enqueue")
    sess.run(inc)
    print("After Enqueue")

It will print

Before Enqueue
hello
After Enqueue

Successfully, we add an operation before enqueue a item into queue.

In previous article, I have found out the reason. But how to resolve it on Multi-GPU-Training is still a question. As the suggestion of this issue in github, I tried two way to fix the problem:
First, rewrite my Averaging-Gradients-Training to learn tf.slim.create_train_op():

...
def create_train_grads(total_loss, optimizer):
  update_ops = set(ops.get_collection(ops.GraphKeys.UPDATE_OPS))
  with ops.control_dependencies(update_ops):
    barrier = control_flow_ops.no_op(name='update_barrier')
  total_loss = control_flow_ops.with_dependencies([barrier], total_loss)
  variables_to_train = tf_variables.trainable_variables()
  grads = optimizer.compute_gradients(total_loss, variables_to_train)
  return grads
...
          cross_entropy = tf.reduce_mean(cross_entropy)
          tf.get_variable_scope().reuse_variables()
          grads = create_train_grads(cross_entropy, opt)
          tower_grads.append(grads)
...
  grads = average_gradients(tower_grads)
  grad_updates = opt.apply_gradients(grads)
  with ops.name_scope('train_op'):
    # Ensure the train_tensor computes grad_updates.
    train_op = control_flow_ops.with_dependencies([grad_updates], cross_entropy)
  # Add the operation used for training to the 'train_op' collection
  train_ops = ops.get_collection_ref(ops.GraphKeys.TRAIN_OP)
  if train_op not in train_ops:
    train_ops.append(train_op)

But unfortunately, this didn’t work at all. The inference result was still a mess.
Then, another way, I use Asynchronous-Gradient-Training and tf.slim.create_train_op():

...
          cross_entropy = tf.reduce_mean(cross_entropy)
          train_op = tf.contrib.slim.learning.create_train_op(cross_entropy, opt)
          tower_ops.append(train_op)
...
  train_step = tf.group(*tower_ops)

Now the inference works very well! And the training speed become a little bit faster than Averaging-Gradients-Training, for the Averaging Operation needs to wait multi gradients from multi GPUs.

After using resnet_v2_50 in tensorflow/models, I found that the inference result is totally incorrect, though the training accuracy looks very well.
Firstly, I suspected the regularization of samples:

  image = tf.image.resize_image_with_crop_or_pad(image, IMAGE_HEIGHT + 66, IMAGE_WIDTH + 66)
  image = tf.random_crop(image, [IMAGE_HEIGHT, IMAGE_WIDTH, IMAGE_CHANNELS])
  image = tf.image.random_flip_left_right(image)

Indeed I had extended the image to a too big size. But after I changing padding size to ’10’, the inference accuracy was still incorrect.
Then I checked the code about importing data:

# To avoid various formats of picture, I encode all image to 'jpeg' and write them as TFRecord
img = cv2.imread(file_name)
raw_image = cv2.imencode('.jpeg', img)[1].tostring()
....
# When importing data from TFRecord
image = tf.image.decode_image(image)

and changed my inference code as the data importing routines. But the problem still existed.
About one week past. Finally, I found this issue in Github. It explains all my questions: the cause is the slim.batch_norm(). After I adding these code to my program (learning from slim.create_train_op()):

update_ops = set(ops.get_collection(ops.GraphKeys.UPDATE_OPS))
with ops.control_dependencies(update_ops):
  barrier = control_flow_ops.no_op(name='update_barrier')
total_loss = control_flow_ops.with_dependencies([barrier], total_loss)
grads = optimizer.compute_gradients(total_loss)
...

The inference accuracy is — still low. Without other choice, I removed all slim.batch_norm() in resnet_v2.py, and at this time inference accuracy becomes the same with training accuracy.
Looks problem partly been solved, but I still need to find out why sli.batch_norm() doesn’t work well in inference …

Here is my experimental code for distributed Tensorflow, which is learned from the example.

import tensorflow as tf
import argparse
import time
FLAGS = None
def main():
    print(tf.__version__)
    cluster_spec = tf.train.ClusterSpec({
        'worker': ['localhost:1829'],
        'ps': ['localhost:1057'],
        })
    if FLAGS.ps:
        server = tf.train.Server(cluster_spec, job_name = 'ps', task_index = 0)
        server.join()
    else:
        server = tf.train.Server(cluster_spec, job_name = 'worker', task_index = FLAGS.worker)
        print(server.target)
        with tf.device('/job:ps/task:0'):
            init = tf.constant_initializer([0])
            c = tf.get_variable('myc', shape = [], initializer = init)
        res = tf.add(c, 1)
        train_op = tf.assign(c, res)
        with tf.Session(target = server.target) as sess:
            c.initializer.run()
            while True:
                res = sess.run(train_op)
                print(res)
                time.sleep(1)
...

The important thing is that we need to use tf.assign() to push Variable back to Parameter Server. The operation ‘tf.add’ was about to run on the task0 of worker in this example. But if we deploy more complicated application by many tasks, things became weird: a pipeline operation sometimes even runs on ‘ps’ role! The official solution to this problem is using ‘tf.train.replica_device_setter()’， which will automatically deploy Variables to parameter servers and Operations (many replicas) to many workers. What did ‘tf.train.replica_device_setter()’ do? Let’s see the backbone code of its implementation:

def replica_device_setter(ps_tasks=0, ps_device="/job:ps",
                          worker_device="/job:worker", merge_devices=True,
                          cluster=None, ps_ops=None, ps_strategy=None):
...
  if ps_ops is None:
    # TODO(sherrym): Variables in the LOCAL_VARIABLES collection should not be
    # placed in the parameter server.
    ps_ops = ["Variable", "VariableV2", "VarHandleOp"]
  if not merge_devices:
    logging.warning(
        "DEPRECATION: It is recommended to set merge_devices=true in "
        "replica_device_setter")
  if ps_strategy is None:
    ps_strategy = _RoundRobinStrategy(ps_tasks)
  if not six.callable(ps_strategy):
    raise TypeError("ps_strategy must be callable")
  chooser = _ReplicaDeviceChooser(
      ps_tasks, ps_device, worker_device, merge_devices, ps_ops, ps_strategy)
  return chooser.device_function

All the Variables will be counted as ‘ps_ops’, and the deploy strategy for Operations will be replication, for it’s called ‘_ReplicaDeviceChooser’.

def device_function(self, op):
...
    node_def = op if isinstance(op, node_def_pb2.NodeDef) else op.node_def
    if self._ps_tasks and self._ps_device and node_def.op in self._ps_ops:
      ps_device = pydev.DeviceSpec.from_string(self._ps_device)
      current_job, ps_job = current_device.job, ps_device.job
      if ps_job and (not current_job or current_job == ps_job):
        ps_device.task = self._ps_strategy(op)
      ps_device.merge_from(current_device)
      return ps_device.to_string()
    worker_device = pydev.DeviceSpec.from_string(self._worker_device or "")
    worker_device.merge_from(current_device)
    return worker_device.to_string()

All the ‘op’ in ‘self._ps_ops’ will be put into ‘ps_device’.