python - tf.Data : what are stragglers in parallel interleaving?

Question

interleave是一种 tf.Data.Dataset 方法，可用于将来自多个数据集的元素交织在一起。 tf.contrib.data.parallel_interleave在 apply 的帮助下提供相同功能的并行版本.

我可以看到并行读取多个数据集并为它们设置并行版本允许的缓冲区将提高吞吐量。但是 documentation关于 parallel_interleave 如何提高数据吞吐量，也有这样的说法:

Unlike tf.data.Dataset.interleave, it gets elements from cycle_length nested datasets in parallel, which increases the throughput, especially in the presence of stragglers.

散乱者到底是什么，为什么 parallel_interleave 在他们存在的情况下在吞吐量方面表现得特别好？

Answer 1

落后者是一种需要比正常时间更长的时间才能产生输出的函数。这可能是由于网络拥塞或随机性的奇怪组合。

interleave在单个线程上以顺序方式执行所有处理。在以下架构中，设 ___表示 等待 IO/计算，<waiting>表示 等待轮到它吐出一个元素 和 111表示 生成第一个元素 ( 1)。

假设我们有一个目录数据集 ds = [A, B, C, D]我们生成文件 1,2,3...从他们每个人。然后使用 r = ds.interleave(cycle_length=3, block_length=2)将像这样工作:

A: ___111___222
B:   <waiting> ___111___________222
C:   <waiting> <waiting> <waiting> ___111___222

R: ____A1____A2____B1____________B2____C1____C2

您会看到，如果从 B 零散地生成元素，则所有后续元素都必须等待处理。

parallel_interleave以两种方式帮助散兵游勇。首先，它并行 启动循环中的每个元素(因此得名)。因此，生产模式变为:

A: ___111___222
B: ___<waiting>111___________222
C: ___<waiting><waiting><waitin>111___222

R: ____A1____A2_B1____________B2_C1____C2|....|

这样做有助于通过并行等待减少无用的等待。零件|....|显示与顺序版本相比我们节省了多少。

它的第二种帮助方式是允许 sloppy争论。如果我们将它设置为 True ，它允许跳过一个不可用的元素直到它可用，但代价是产生一个不确定的顺序。方法如下:

A: ___111___<w>222
B: ___<w>111___________222
C: ___<w><w>111___222

R: ____A1_B1_C1_A2_C2___B2|...................|

看看节省的钱!!但也要注意元素的顺序!

---

我在代码中重现了这些。这是一种丑陋的方式，但它稍微说明了差异。

from time import sleep
DS = tf.data.Dataset

def repeater(val):
    def _slow_gen():
        for i in range(5):
            if i % 2:
                sleep(1)
            yield i
    return DS.from_generator(_slow_gen, tf.int8)

ds = DS.range(5)

slow_ds = ds.interleave(repeater, cycle_length=2, block_length=3)

para_ds = ds.apply(tf.contrib.data.parallel_interleave(
    repeater, cycle_length=2, block_length=3)
)

sloppy_ds = ds.apply(tf.contrib.data.parallel_interleave(
    repeater, cycle_length=2, block_length=3, sloppy=True)
)


%time apply_python_func(slow_ds, print, sess)
# 10 sec, you see it waiting each time

%time apply_python_func(para_ds, print, sess) 
#  3 sec always! you see it burping a lot after the first wait

%time apply_python_func(sloppy_ds, print, sess) 
# sometimes 3, sometimes 4 seconds

这是显示数据集的函数

def apply_python_func(ds, func, sess):
    """Exact values from ds using sess and apply func on them"""
    it = ds.make_one_shot_iterator()
    next_value = it.get_next()
    num_examples = 0
    while True:
        try:
            value = sess.run(next_value)
            num_examples += 1
            func(value)
        except tf.errors.OutOfRangeError:
            break
    print('Evaluated {} examples'.format(num_examples))