<h3 dir="auto">Environment info</h3>
<p dir="auto">Operating System: Kubernetes/Ubuntu 14.04</p>
<h3 dir="auto">Steps to reproduce</h3>
<ol dir="auto">
<li>start tensorflow server on kubernetes using yaml generated by <a href="https://github.com/tensorflow/tensorflow/blob/master/tensorflow/tools/dist_test/scripts/k8s_tensorflow.py">the python code</a></li>
<li>Start tensorflow code:</li>
</ol>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="python mnist_dnn.py --worker_grpc_url=grpc://180.101.191.78:30001 --worker_index=0 --workers=180.101.191.78:30001,180.101.191.78:30002,180.101.191.78:30003 --parameter_servers=tf-ps0:2222,tf-ps1:2222"><pre class="notranslate"><code class="notranslate">python mnist_dnn.py --worker_grpc_url=grpc://180.101.191.78:30001 --worker_index=0 --workers=180.101.191.78:30001,180.101.191.78:30002,180.101.191.78:30003 --parameter_servers=tf-ps0:2222,tf-ps1:2222
</code></pre></div>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="python mnist_dnn.py --worker_grpc_url=grpc://180.101.191.78:30002 --worker_index=1 --workers=180.101.191.78:30001,180.101.191.78:30002,180.101.191.78:30003 --parameter_servers=tf-ps0:2222,tf-ps1:2222"><pre class="notranslate"><code class="notranslate">python mnist_dnn.py --worker_grpc_url=grpc://180.101.191.78:30002 --worker_index=1 --workers=180.101.191.78:30001,180.101.191.78:30002,180.101.191.78:30003 --parameter_servers=tf-ps0:2222,tf-ps1:2222
</code></pre></div>
<h3 dir="auto">What have you tried?</h3>
<ol dir="auto">
<li>The worker with index 0 (chief) can execute normally.</li>
<li>It was able to execute well (using the same yaml and code)</li>
<li>I tried to restart the servers, but it didn't work.</li>
<li>All other workers stuck at <code class="notranslate">prepare_or_wait_for_session</code>. However, it seems that logs suggest other workers are actually executing.</li>
</ol>
<p dir="auto">Log is <a href="https://github.com/tensorflow/tensorflow/files/332895/log.txt">here</a> and the code is here:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="import sys
import time

import tensorflow as tf
from tensorflow.examples.tutorials.mnist import input_data
import datetime

flags = tf.app.flags
flags.DEFINE_string(&quot;data_dir&quot;, &quot;/tmp/data&quot;,
                    &quot;Directory for storing mnist data&quot;)

flags.DEFINE_boolean(&quot;download_only&quot;, False,
                     &quot;Only perform downloading of data; Do not proceed to &quot;
                     &quot;session preparation, model definition or training&quot;)

flags.DEFINE_integer(&quot;worker_index&quot;, 0,
                     &quot;Worker task index, should be &gt;= 0. worker_index=0 is &quot;
                     &quot;the master worker task the performs the variable &quot;
                     &quot;initialization &quot;)

flags.DEFINE_string(&quot;workers&quot;, None,
                    &quot;The worker url list, separated by comma (e.g. tf-worker1:2222,1.2.3.4:2222)&quot;)

flags.DEFINE_string(&quot;parameter_servers&quot;, None,
                    &quot;The ps url list, separated by comma (e.g. tf-ps2:2222,1.2.3.5:2222)&quot;)

flags.DEFINE_integer(&quot;grpc_port&quot;, 2222,
                     &quot;TensorFlow GRPC port&quot;)

flags.DEFINE_integer(&quot;train_steps&quot;, 200000,
                     &quot;Number of (global) training steps to perform&quot;)

flags.DEFINE_string(&quot;worker_grpc_url&quot;, None,
                    &quot;Worker GRPC URL (e.g., grpc://1.2.3.4:2222, or &quot;
                    &quot;grpc://tf-worker0:2222)&quot;)
FLAGS = flags.FLAGS

cur_time = datetime.datetime.now().strftime('%Y%m%d%H%M%S')

def nn_layer(input_tensor, input_dim, output_dim, act=tf.nn.relu):
    with tf.name_scope(cur_time):
        weights = tf.Variable(tf.truncated_normal([input_dim, output_dim], stddev=0.1))
        biases = tf.Variable(tf.constant(0.1, shape=[output_dim]))
    activations = act(tf.matmul(input_tensor, weights) + biases)
    return activations

def model(x, y_, global_step):
    hidden_nodes = 500
    hidden1 = nn_layer(x, 784, hidden_nodes)
    y = nn_layer(hidden1, hidden_nodes, 10, act=tf.nn.softmax)

    cross_entropy = -tf.reduce_mean(y_ * tf.log(tf.clip_by_value(y, 1e-10, 1.0)))
    train_step = tf.train.AdamOptimizer(0.001).minimize(cross_entropy, global_step=global_step)

    correct_prediction = tf.equal(tf.argmax(y, 1), tf.argmax(y_, 1))
    accuracy = tf.reduce_mean(tf.cast(correct_prediction, tf.float32))

    return train_step, accuracy

print(&quot;Loading data from worker index = %d&quot; % FLAGS.worker_index)

mnist = input_data.read_data_sets(FLAGS.data_dir, one_hot=True)
print(&quot;Testing set size: %d&quot; % len(mnist.test.images))
print(&quot;Training set size: %d&quot; % len(mnist.train.images))
if FLAGS.download_only: sys.exit(0)

print(&quot;Worker GRPC URL: %s&quot; % FLAGS.worker_grpc_url)
print(&quot;Workers = %s&quot; % FLAGS.workers)
print(&quot;Using time = %s&quot; % cur_time)

is_chief = (FLAGS.worker_index == 0)
cluster = tf.train.ClusterSpec({&quot;ps&quot;: FLAGS.parameter_servers.split(&quot;,&quot;), &quot;worker&quot;: FLAGS.workers.split(&quot;,&quot;)})
# Construct device setter object
device_setter = tf.train.replica_device_setter(cluster=cluster)

# The device setter will automatically place Variables ops on separate
# parameter servers (ps). The non-Variable ops will be placed on the workers.
with tf.device(device_setter):
    with tf.name_scope(cur_time):
        global_step = tf.Variable(0, trainable=False)

    x = tf.placeholder(tf.float32, [None, 784])
    y_ = tf.placeholder(tf.float32, [None, 10])
    val_feed = {x: mnist.test.images, y_: mnist.test.labels}

    train_step, accuracy = model(x, y_, global_step)

    sv = tf.train.Supervisor(is_chief=is_chief,
                             logdir=&quot;/tmp/dist-mnist-log/train&quot;,
                             saver=tf.train.Saver(),
                             init_op=tf.initialize_all_variables(),
                             recovery_wait_secs=1,
                             global_step=global_step)
    sess_config = tf.ConfigProto(allow_soft_placement=True, log_device_placement=True,
                                 device_filters=[&quot;/job:ps&quot;, &quot;/job:worker/task:%d&quot; % FLAGS.worker_index])

    # The chief worker (worker_index==0) session will prepare the session,
    # while the remaining workers will wait for the preparation to complete.
    if is_chief:
        print(&quot;Worker %d: Initializing session...&quot; % FLAGS.worker_index)
    else:
        print(&quot;Worker %d: Waiting for session to be initialized...&quot; % FLAGS.worker_index)

    with sv.prepare_or_wait_for_session(FLAGS.worker_grpc_url, config=sess_config) as sess:
        print(&quot;Worker %d: Session initialization complete.&quot; % FLAGS.worker_index)

        # Perform training
        time_begin = time.time()
        print(&quot;Training begins @ %f&quot; % time_begin)

        local_step = 0
        while True:
            # Training feed
            batch_xs, batch_ys = mnist.train.next_batch(100)
            train_feed = {x: batch_xs, y_: batch_ys}

            _, step = sess.run([train_step, global_step], feed_dict=train_feed)
            local_step += 1
            if local_step % 100 == 0:
                print(&quot;Worker %d: training step %d done (global step: %d); Accuracy: %g&quot; % 
                      (FLAGS.worker_index, local_step, step, sess.run(accuracy, feed_dict=val_feed)))
            if step &gt;= FLAGS.train_steps: break

        time_end = time.time()
        print(&quot;Training ends @ %f&quot; % time_end)
        training_time = time_end - time_begin
        print(&quot;Training elapsed time: %f s&quot; % training_time)

        # Accuracy on test data
        print(&quot;Final test accuracy = %g&quot; % (sess.run(accuracy, feed_dict=val_feed)))"><pre class="notranslate"><code class="notranslate">import sys
import time

import tensorflow as tf
from tensorflow.examples.tutorials.mnist import input_data
import datetime

flags = tf.app.flags
flags.DEFINE_string("data_dir", "/tmp/data",
                    "Directory for storing mnist data")

flags.DEFINE_boolean("download_only", False,
                     "Only perform downloading of data; Do not proceed to "
                     "session preparation, model definition or training")

flags.DEFINE_integer("worker_index", 0,
                     "Worker task index, should be &gt;= 0. worker_index=0 is "
                     "the master worker task the performs the variable "
                     "initialization ")

flags.DEFINE_string("workers", None,
                    "The worker url list, separated by comma (e.g. tf-worker1:2222,1.2.3.4:2222)")

flags.DEFINE_string("parameter_servers", None,
                    "The ps url list, separated by comma (e.g. tf-ps2:2222,1.2.3.5:2222)")

flags.DEFINE_integer("grpc_port", 2222,
                     "TensorFlow GRPC port")

flags.DEFINE_integer("train_steps", 200000,
                     "Number of (global) training steps to perform")

flags.DEFINE_string("worker_grpc_url", None,
                    "Worker GRPC URL (e.g., grpc://1.2.3.4:2222, or "
                    "grpc://tf-worker0:2222)")
FLAGS = flags.FLAGS

cur_time = datetime.datetime.now().strftime('%Y%m%d%H%M%S')

def nn_layer(input_tensor, input_dim, output_dim, act=tf.nn.relu):
    with tf.name_scope(cur_time):
        weights = tf.Variable(tf.truncated_normal([input_dim, output_dim], stddev=0.1))
        biases = tf.Variable(tf.constant(0.1, shape=[output_dim]))
    activations = act(tf.matmul(input_tensor, weights) + biases)
    return activations

def model(x, y_, global_step):
    hidden_nodes = 500
    hidden1 = nn_layer(x, 784, hidden_nodes)
    y = nn_layer(hidden1, hidden_nodes, 10, act=tf.nn.softmax)

    cross_entropy = -tf.reduce_mean(y_ * tf.log(tf.clip_by_value(y, 1e-10, 1.0)))
    train_step = tf.train.AdamOptimizer(0.001).minimize(cross_entropy, global_step=global_step)

    correct_prediction = tf.equal(tf.argmax(y, 1), tf.argmax(y_, 1))
    accuracy = tf.reduce_mean(tf.cast(correct_prediction, tf.float32))

    return train_step, accuracy

print("Loading data from worker index = %d" % FLAGS.worker_index)

mnist = input_data.read_data_sets(FLAGS.data_dir, one_hot=True)
print("Testing set size: %d" % len(mnist.test.images))
print("Training set size: %d" % len(mnist.train.images))
if FLAGS.download_only: sys.exit(0)

print("Worker GRPC URL: %s" % FLAGS.worker_grpc_url)
print("Workers = %s" % FLAGS.workers)
print("Using time = %s" % cur_time)

is_chief = (FLAGS.worker_index == 0)
cluster = tf.train.ClusterSpec({"ps": FLAGS.parameter_servers.split(","), "worker": FLAGS.workers.split(",")})
# Construct device setter object
device_setter = tf.train.replica_device_setter(cluster=cluster)

# The device setter will automatically place Variables ops on separate
# parameter servers (ps). The non-Variable ops will be placed on the workers.
with tf.device(device_setter):
    with tf.name_scope(cur_time):
        global_step = tf.Variable(0, trainable=False)

    x = tf.placeholder(tf.float32, [None, 784])
    y_ = tf.placeholder(tf.float32, [None, 10])
    val_feed = {x: mnist.test.images, y_: mnist.test.labels}

    train_step, accuracy = model(x, y_, global_step)

    sv = tf.train.Supervisor(is_chief=is_chief,
                             logdir="/tmp/dist-mnist-log/train",
                             saver=tf.train.Saver(),
                             init_op=tf.initialize_all_variables(),
                             recovery_wait_secs=1,
                             global_step=global_step)
    sess_config = tf.ConfigProto(allow_soft_placement=True, log_device_placement=True,
                                 device_filters=["/job:ps", "/job:worker/task:%d" % FLAGS.worker_index])

    # The chief worker (worker_index==0) session will prepare the session,
    # while the remaining workers will wait for the preparation to complete.
    if is_chief:
        print("Worker %d: Initializing session..." % FLAGS.worker_index)
    else:
        print("Worker %d: Waiting for session to be initialized..." % FLAGS.worker_index)

    with sv.prepare_or_wait_for_session(FLAGS.worker_grpc_url, config=sess_config) as sess:
        print("Worker %d: Session initialization complete." % FLAGS.worker_index)

        # Perform training
        time_begin = time.time()
        print("Training begins @ %f" % time_begin)

        local_step = 0
        while True:
            # Training feed
            batch_xs, batch_ys = mnist.train.next_batch(100)
            train_feed = {x: batch_xs, y_: batch_ys}

            _, step = sess.run([train_step, global_step], feed_dict=train_feed)
            local_step += 1
            if local_step % 100 == 0:
                print("Worker %d: training step %d done (global step: %d); Accuracy: %g" % 
                      (FLAGS.worker_index, local_step, step, sess.run(accuracy, feed_dict=val_feed)))
            if step &gt;= FLAGS.train_steps: break

        time_end = time.time()
        print("Training ends @ %f" % time_end)
        training_time = time_end - time_begin
        print("Training elapsed time: %f s" % training_time)

        # Accuracy on test data
        print("Final test accuracy = %g" % (sess.run(accuracy, feed_dict=val_feed)))
</code></pre></div>
<p dir="auto">I tried both Distributed <a href="https://www.tensorflow.org/versions/r0.8/how_tos/distributed/index.html" rel="nofollow">tensorflow</a> and <a href="https://github.com/tensorflow/tensorflow/files/271770/mnist_softmax.py.txt">minist_softmax.py.txt</a>. They both hang at the prepare_or_wait_for_session() function with non-chief worker tasks.</p>
<p dir="auto">task_index == 0(chief worker ) can run as expected.<br>
I can not find out the reasion why it behaves like this.<br>
Can anyone please give some advice?</p>