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’.

Recently, we create a new custom operation in Tensorflow:

REGISTER_OP("GetImageID")
    .Input("count: int32")
    .Output("image_id: string");
using namespace tensorflow;
using namespace std;
class GetImageIDOp : public OpKernel {
 public:
  explicit GetImageIDOp(OpKernelConstruction* ctx) : OpKernel(ctx) {
  }
  void Compute(OpKernelContext* ctx) override {
    const Tensor* cnt;
    OP_REQUIRES_OK(ctx, ctx->input("count", &cnt));  // This is how we should get 'input' of Op
    OP_REQUIRES(ctx, TensorShapeUtils::IsScalar(cnt->shape()),
                errors::InvalidArgument("cnt is not a scalar: ", cnt->shape().DebugString()));
    int32 count = cnt->scalar()();
    vector image_ids;
    int num = get_image_ids(&image_ids);  // Get image_ids from somewhere, such as network, or disk
    Tensor* image_id;
    OP_REQUIRES_OK(ctx, ctx->allocate_output("image_id", TensorShape({static_cast(num)}), &image_id));
    auto imageid_flat = image_id->flat();
    for (int i = 0; i < num; i++) {
      imageid_flat(i) = image_ids[i];
    }
  }
 private:
};
REGISTER_KERNEL_BUILDER(Name("GetImageID").Device(DEVICE_CPU), GetImageIDOp);

It's as simple as the example in Tensorflow's document. But when we run this Op in session:

get_image_id_op = get_image_id(32)
with tf.Session() as sess:
  while (True):
    sess.run(get_image_id_op)

It only get image_ids from network once, and then use the result of first 'run' forever, without even call 'Compute()' function in cpp code again!
Seems Tensorflow optimized the new Op and never run it twice. My colleague give a suggestion to solve this problem by using tf.placeholder:

counter = tf.placeholder(tf.int32)
get_image_id_op = get_image_id(counter)
with tf.Session() as sess:
  while (True):
    sess.run(get_image_id_op, feed_dict = {counter: count})

Looks a little tricky. The final solution is add flag in cpp code to let new Op to avoid CSE (Common Subexpression Elimination):

REGISTER_OP("GetImageID")
    .SetIsStateful()
    .Input("count: int32")
    .Output("image_id: string");
......

Attachment of the 'CMakeLists.txt':

cmake_minimum_required(VERSION 2.8)
project(my_proj)
set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fPIC -std=c++11 -O2 -g2")
#set(CMAKE_MACOSX_RPATH 0)
set(CMAKE_SKIP_RPATH TRUE)
if (APPLE)
  set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -undefined dynamic_lookup")
elseif (UNIX)
  set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS}")
endif()
set(CMAKE_SHARED_LIBRARY_SUFFIX ".so")
execute_process(COMMAND python -c "import tensorflow as tf; print(tf.sysconfig.get_include())" OUTPUT_VARIABLE tf_inc)
include_directories(${tf_inc} "my/")
link_directories("/usr/lib64/" "${CMAKE_CURRENT_SOURCE_DIR}/tensorflow-core/third_party/erpc_lib/lib/")
set(CMAKE_LIBRARY_OUTPUT_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/lib/)
add_library(my_op SHARED my_op.cc)
target_link_libraries(my_op tbb protobuf)

In Tensorflow, we could use Optimizer to train model:

train_op = optimizer.minimize(loss, global_step=global_step)
....
sess.run([train_op], feed_dict={x: batch[0], y_: batch[1]})

But sometimes, model need to be split to two parts and trained separately, so we need to compute gradients and apply them by two steps:

first_part_vars = []
second_part_vars = []
for var in tf.trainable_variables():
  if var.name.find("first_part/") >= 0:
    first_part_vars.append(var)
  if var.name.find("second_part/") >= 0:
    second_part_vars.append(var)
....
first_grads_and_vars = opt.compute_gradients(loss, var_list = first_part_vars)
train_first_part_op = opt.apply_gradients(first_grads_and_vars)
second_grads_and_vars = opt.compute_gradients(loss, var_list = second_part_vars)
train_second_part_op = opt.apply_gradients(second_grads_and_vars)
....

Then how could we delivery gradients from first part to second part? Here is the equation to answer:

The means ‘intermediate vector’, which is the interface vector between first-part and second-part and it is belong to both first-part and second-part. The is the weights of second part of model. Therefore we could use tf.gradients() to connect gradients of two parts:

first_grads_and_vars = opt.compute_gradients(loss, var_list=[first_part_vars.extend(iv)])
train_first_part_op = opt.apply_gradients(first_grads_and_vars)
iv_ph = tf.placeholder(tf.float32, iv.shape)
second_grads = tf.gradients(loss, second_part_vars, iv_ph)
second_grads_and_vars = list(zip(second_grads, second_part_vars))
train_second_part_op = opt.apply_gradients(second_grads_and_vars)
....
_, ivg = sess.run([train_first_part_op, iv_gradients], feed_dict = {x: batch[0], y_: batch[1]})
sess.run([train_second_part_op], feed_dict = {x: batch[0], iv_ph: ivg})

This article shows how to use sharing variables in Tensroflow. But I still have a question: dose sharing variables have the same value? To answer this question, I write these code below:

import tensorflow as tf
#initial = tf.constant(0.1, shape=[1])
initial = tf.truncated_normal(shape=[3], stddev=1, mean=1)
a = [None] * 3
b = [None] * 3
c = [None] * 3
with tf.variable_scope(tf.get_variable_scope()):
  for i in xrange(3):
    with tf.name_scope("my_%d" % i):
      a[i] = tf.Variable(initial, [3])
      b[i] = tf.Variable(initial, [3])
      c[i] = a[i] + b[i]
      tf.get_variable_scope().reuse_variables()
sess = tf.Session()
sess.run(tf.global_variables_initializer())
curr_a, curr_b, curr_a1, curr_b1, curr_a2, curr_b2, curr_c = sess.run([a[0], b[0], a[1], b[1], a[2], b[2], c[0]], feed_dict={a[0]:[0.1, 0.1, 0.1], b[0]:[0.2,0.2,0.2]})
print(curr_a, curr_b, curr_a1, curr_b1, curr_a2, curr_b2, curr_c)
The result of running these python code is:
(array([ 0.1,  0.1,  0.1], dtype=float32), array([ 0.2,  0.2,  0.2], dtype=float32), array([ 0.90568691,  1.30992699,  1.49500561], dtype=float32), array([ 0.90568691,  1.30992699,  1.49500561], dtype=float32), array([ 0.90568691,  1.30992699,  1.49500561], dtype=float32), array([ 0.90568691,  1.30992699,  1.49500561], dtype=float32), array([ 0.30000001,  0.30000001,  0.30000001], dtype=float32))

Therefore, the "sharing variables" mechanism is made only for convenience of writing short code to create multi-models. For sharing same value for different variables, we still need 'assign' operation.
   
    
    	

Q: How to fix error report like

tensorflow.python.framework.errors_impl.InvalidArgumentError: Input 0 of node
Adam/update_embeddings/AssignSub was passed float from _recv_embeddings_0:0
incompatible with expected float_ref

A: We can’t feed a value into a variable and optimize it in the same time (So the problem only occurs when using Optimizers). Should using ‘tf.assign()’ in graph to give value to tf.Variable
Q: How to get a tensor by name?
A: like this:

tensor = tf.get_default_graph().get_tensor_by_name("example:0")

Q: How to get variable by name?
A:

my_var = [v for v in tf.global_variables() if v.name == "myvar_23:0"][0]

Sometimes, we need to average an array of gradients in deep learning model. Fortunately, Tensorflow divided models into fine-grained tensors and operations, therefore it’s not difficult to implement gradients average by using it.
Let’s see the code from github：

with tf.variable_scope(tf.get_variable_scope()):
      for i in xrange(FLAGS.num_gpus):
        with tf.device('/gpu:%d' % i):
          with tf.name_scope('%s_%d' % (cifar10.TOWER_NAME, i)) as scope:
            # Dequeues one batch for the GPU
            image_batch, label_batch = batch_queue.dequeue()
            # Calculate the loss for one tower of the CIFAR model. This function
            # constructs the entire CIFAR model but shares the variables across
            # all towers.
            loss = tower_loss(scope, image_batch, label_batch)
            # Reuse variables for the next tower.
            tf.get_variable_scope().reuse_variables()
            # Retain the summaries from the final tower.
            summaries = tf.get_collection(tf.GraphKeys.SUMMARIES, scope)
            # Calculate the gradients for the batch of data on this CIFAR tower.
            grads = opt.compute_gradients(loss)
            # Keep track of the gradients across all towers.
            tower_grads.append(grads)
    # We must calculate the mean of each gradient. Note that this is the
    # synchronization point across all towers.
    grads = average_gradients(tower_grads)
......
    apply_gradient_op = opt.apply_gradients(grads, global_step=global_step)

We should keep in mind that these codes will only build a static graph (the ‘grads; are references rather than values).

def average_gradients(tower_grads):
  average_grads = []
  for grad_and_vars in zip(*tower_grads):
    # Note that each grad_and_vars looks like the following:
    #   ((grad0_gpu0, var0_gpu0), ... , (grad0_gpuN, var0_gpuN))
    grads = []
    for g, _ in grad_and_vars:
      # Add 0 dimension to the gradients to represent the tower.
      expanded_g = tf.expand_dims(g, 0)
      # Append on a 'tower' dimension which we will average over below.
      grads.append(expanded_g)
    # Average over the 'tower' dimension.
    grad = tf.concat(axis=0, values=grads)
    grad = tf.reduce_mean(grad, 0)
    # Keep in mind that the Variables are redundant because they are shared
    # across towers. So .. we will just return the first tower's pointer to
    # the Variable.
    v = grad_and_vars[0][1]
    grad_and_var = (grad, v)
    average_grads.append(grad_and_var)
  return average_grads

First, we need to expand dimensions of tensor(gradient) and concatenate them. Then use reduce_mean() to do actually average operation (seems not intuitive).