Does horovod handle the problem of each process has different batch num?

[formath]

is_chief = (hvd.rank() == 0)
# model
train_iterator = data_reader.iterator(
                              batch_size=100,
                              file_pattern='hdfs://....../part-*',
                              shard_num=device_num,
                              worker_index=hvd.rank()) # data parallel between device using tf.Dataset
loss = model(train_iterator.get_next())
opt = tf.train.AdamOptimizer(learning_rate=0.0001)
opt = hvd.DistributedOptimizer(opt)
train_op = opt.minimize(train_loss)
broadcast_global_variables_op = hvd.broadcast_global_variables(0)

# train process
sess_config = tf.ConfigProto(allow_soft_placement=True)
sess_config.gpu_options.allow_growth = True
sess_config.gpu_options.visible_device_list = str(hvd.local_rank())
with tf.Session(config=sess_config) as sess:
    sess.run(tf.global_variables_initializer())
    sess.run(tf.local_variables_initializer())
    tf.tables_initializer().run()
    sess.run(broadcast_global_variables_op)
    
    # train epoch by epoch
    epoch_num = 0
    while epoch_num < 6:
        epoch_num += 1
        sess.run(train_iterator.initializer)
        while True:
            try:
                sess.run(train_op)
            except tf.errors.OutOfRangeError:
                if is_chief:
                    saver.save(sess=sess,
                                save_path="hdfs://....../model.checkpoint." + str(epoch_num),
                                latest_filename='checkpoint.' + str(epoch_num))
                break

I start horovod with 8 gpu devices where each device uses 1/8 part of all data. The job trains 6 epochs. The first 5 epochs run ok. But when the last epoch of the chief is done, an error will occur before saver.save.
The error message is:

tensorflow.python.framework.errors_impl.UnknownError: Horovod has been shut down. This was caused by an exception on one of the ranks or an attempt to allreduce, allgather or broadcast a tensor after one of the ranks finished execution. If the shutdown was caused by an exception, you should see the exception in the log before the first shutdown message.
[[node DistributedAdamOptimizer_Allreduce/cond_8/HorovodAllgather_gradients_concat_16_0 (defined at /usr/local/lib64/python3.6/site-packages/tensorflow_core/python/framework/ops.py:1748) ]]

Even each device uses 1/8 part of all data, I can't assure all devices has the same batch num e.g. 1/8 * all_data_num / batch_size. So when some one device needs allreduce, some other devices may have been stopped.

[maxhgerlach]

Is hvd.join() any help?

[formath]

Do I need to run horovod.tensorflow.broadcast_global_variables(last_joined_rank) after all workers have joined as mentioned in #832 (comment)?

is_chief = (hvd.rank() == 0)
# model
train_iterator = data_reader.iterator(
                              batch_size=100,
                              file_pattern='hdfs://....../part-*',
                              shard_num=device_num,
                              worker_index=hvd.rank()) # data parallel between device using tf.Dataset
loss = model(train_iterator.get_next())
opt = tf.train.AdamOptimizer(learning_rate=0.0001)
opt = hvd.DistributedOptimizer(opt)
train_op = opt.minimize(train_loss)
broadcast_global_variables_op = hvd.broadcast_global_variables(0)

# train process
sess_config = tf.ConfigProto(allow_soft_placement=True)
sess_config.gpu_options.allow_growth = True
sess_config.gpu_options.visible_device_list = str(hvd.local_rank())
with tf.Session(config=sess_config) as sess:
    sess.run(tf.global_variables_initializer())
    sess.run(tf.local_variables_initializer())
    tf.tables_initializer().run()
    sess.run(broadcast_global_variables_op)
    
    # train epoch by epoch
    epoch_num = 0
    while epoch_num < 6:
        epoch_num += 1
        sess.run(train_iterator.initializer)
        while True:
            try:
                sess.run(train_op)
            except tf.errors.OutOfRangeError:
                # join and broadcast newest parameter
                last_joined_rank = sess.run(hvd.join())
                sess.run(hvd.broadcast_global_variables(last_joined_rank))
                if is_chief:
                    saver.save(sess=sess,
                                save_path="hdfs://....../model.checkpoint." + str(epoch_num),
                                latest_filename='checkpoint.' + str(epoch_num))
                break

[formath]

After adding join, this error occurs.

tensorflow.python.framework.errors_impl.FailedPreconditionError: Allgather is not supported with Join at this time. Specify sparse_to_dense=True if using DistributedOptimizer
[[{{node DistributedAdamOptimizer_Allreduce/cond_8/HorovodAllgather_gradients_concat_16_0}}]]