Keras 的 OrderedEnqueuer 中是否保证批次的顺序？

Question

我有一个自定义 keras.utils.sequence，它以特定（和关键）顺序生成批次。

但是，我需要跨多个内核并行化批量生成。名称'OrderedEnqueuer'是否暗示结果队列中批次的顺序保证与原始keras.utils.sequence的顺序相同？

我认为这个顺序不能保证的原因：

1. OrderedEnqueuer 在内部使用 python multiprocessing 的 apply_async。
2. Keras 的文档明确表示 OrderedEnqueuer 保证不会重复批次 - 但不能保证顺序。

我认为是这样的原因：

1. 名字！
2. 我知道keras.utils.sequence 对象是可索引的。
3. 我在 Keras 的 github 上找到了似乎旨在验证顺序的测试脚本 - 尽管我找不到任何关于这些是否通过或它们是否真正具有决定性的文档。

如果这里的订单不能保证，我欢迎任何关于如何在保证订单的同时并行化批处理准备的建议，但前提是它必须能够并行化任意 python 代码 - 我相信例如 tf.data.Dataset API 可以不允许这样做（tf.py_function 回调原始 python 进程）。

Answer 1

是的，已订购。

通过以下测试自行检查。

首先，让我们创建一个虚拟Sequence，它在等待随机时间后仅返回批次索引（随机时间是为了确保批次不会按顺序完成）：

import time, random, datetime
import numpy as np
import tensorflow as tf

class DataLoader(tf.keras.utils.Sequence):
    def __len__(self):
        return 10
    def __getitem__(self, i):
        time.sleep(random.randint(1,2))
        #you could add a print here to see that it's out of order
        return i

现在让我们创建一个测试函数来创建并使用它。 该函数获取工人的数量并打印所花费的时间以及返回的结果。

def test(workers):
    enq = tf.keras.utils.OrderedEnqueuer(DataLoader())
    enq.start(workers = workers)
    gen = enq.get()

    results = []
    start = datetime.datetime.now()
    for i in range(30):
        results.append(next(gen))
    enq.stop()
    print('test with', workers, 'workers took', datetime.datetime.now() - start)
    print("results:", results)

结果：

test(1)
test(8)

1 名工人的测试时间为 0:00:45.093122
结果：[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3 , 4, 5, 6, 7, 8, 9]
8 名工人的测试时间为 0:00:09.127771
结果：[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3 , 4, 5, 6, 7, 8, 9]

注意：

• 8 个工作人员比 1 个工作人员快得多 -> 可以并行化
• 两种情况的结果都已排序