关于 fit_generator() / fit() 和线程安全

Question

上下文

为了在 Keras 中使用 fit_generator()，我使用了像这样的生成器函数 pseudocode-one:

def generator(data: np.array) -> (np.array, np.array):
    """Simple generator yielding some samples and targets"""

    while True:
        for batch in range(number_of_batches):
            yield data[batch * length_sequence], data[(batch + 1) * length_sequence]

在 Keras 的 fit_generator() 函数中，我想使用 workers=4 和 use_multiprocessing=True - 因此，我需要一个线程安全生成器。

在here 或here 或Keras docs 等stackoverflow 上的答案中，我读到了关于创建一个继承自Keras.utils.Sequence() 的类，如下所示：

class generatorClass(Sequence):

    def __init__(self, x_set, y_set, batch_size):
        self.x, self.y = x_set, y_set
        self.batch_size = batch_size

    def __len__(self):
        return int(np.ceil(len(self.x) / float(self.batch_size)))

    def __getitem__(self, idx):
        batch_x = self.x[idx * self.batch_size:(idx + 1) * self.batch_size]
        batch_y = self.y[idx * self.batch_size:(idx + 1) * self.batch_size]

        return ...

通过使用Sequences Keras 不会在使用多个工作和多处理时抛出任何警告；生成器应该是线程安全的。

无论如何，由于我使用的是自定义函数，我偶然发现了 github 上提供的 Omer Zohars 代码，它允许通过添加装饰器使我的 generator() 线程安全。 代码如下：

import threading

class threadsafe_iter:
    """
    Takes an iterator/generator and makes it thread-safe by
    serializing call to the `next` method of given iterator/generator.
    """
    def __init__(self, it):
        self.it = it
        self.lock = threading.Lock()

    def __iter__(self):
        return self

    def __next__(self):
        with self.lock:
            return self.it.__next__()


def threadsafe_generator(f):
    """A decorator that takes a generator function and makes it thread-safe."""
    def g(*a, **kw):
        return threadsafe_iter(f(*a, **kw))

    return g

现在我可以做：

@threadsafe_generator
def generator(data):
    ...

问题是：使用此版本的线程安全生成器 Keras 仍然会发出警告，即在使用 workers > 1 和 use_multiprocessing=True 时生成器必须是线程安全的，并且可以通过使用 Sequences 来避免这种情况。

我现在的问题是：

1. Keras 是否仅因为生成器未继承 Sequences 而发出此警告，或者 Keras 是否也会检查生成器是否通常是线程安全的？
2. 是否使用我选择作为线程安全的方法作为使用来自Keras-docs 的generatorClass(Sequence)-version？
3. Keras 可以处理哪些与这两个示例不同的导致线程安全生成器的方法？
   
   

---

编辑： 在较新的 tensorflow/keras-versions (tf > 2) 中，fit_generator() 已弃用。相反，建议将fit() 与生成器一起使用。但是，这个问题仍然适用于使用生成器的fit()。

Answer 1

在我对此进行研究期间，我发现了一些回答我的问题的信息。

注意：如更新的tensorflow/keras-versions (tf > 2) fit_generator() 中的问题所更新，已弃用。相反，建议将fit() 与生成器一起使用。但是，答案仍然适用于使用生成器的fit()。

---

1. Keras 是否仅因为生成器未继承序列而发出此警告，或者 Keras 是否也检查生成器是否一般是线程安全的？

取自 Keras 的 gitRepo (training_generators.py) 我在 46-52 行中找到以下内容：

use_sequence_api = is_sequence(generator)
if not use_sequence_api and use_multiprocessing and workers > 1:
    warnings.warn(
        UserWarning('Using a generator with `use_multiprocessing=True`'
                    ' and multiple workers may duplicate your data.'
                    ' Please consider using the `keras.utils.Sequence'
                    ' class.'))

is_sequence() 的定义取自 training_utils.py 行 624-635 是：

def is_sequence(seq):
    """Determine if an object follows the Sequence API.
    # Arguments
        seq: a possible Sequence object
    # Returns
        boolean, whether the object follows the Sequence API.
    """
    # TODO Dref360: Decide which pattern to follow. First needs a new TF Version.
    return (getattr(seq, 'use_sequence_api', False)
            or set(dir(Sequence())).issubset(set(dir(seq) + ['use_sequence_api'])))

关于这段代码，Keras 只检查传递的生成器是否是 Keras 序列（或者更确切地说是使用 Keras 的序列 API），一般不检查生成器是否是线程安全的。

---

2. 是否使用我选择作为线程安全的方法作为使用来自Keras-docs 的 generatorClass(Sequence)-version？

正如 Omer Zohar 在 gitHub 上所展示的那样，他的装饰器是线程安全的 - 我看不出它不应该成为 Keras 的线程安全的任何理由（即使 Keras 会发出警告，如 1. 所示）。 根据docs，thread.Lock()的实现可以认为是线程安全的：

返回一个新的原始锁对象的工厂函数。 一旦一个线程获取了它，后续的获取它的尝试就会阻塞，直到它被释放;任何线程都可以释放它。

生成器也是可腌制的，可以像这样进行测试（有关更多信息，请参阅此 SO-Q&A here）：

#Dump yielded data in order to check if picklable
with open("test.pickle", "wb") as outfile:
    for yielded_data in generator(data):
        pickle.dump(yielded_data, outfile, protocol=pickle.HIGHEST_PROTOCOL)

恢复这一点，我什至建议在扩展 Keras 的 Sequence() 时实现 thread.Lock()，例如：

import threading

class generatorClass(Sequence):

    def __init__(self, x_set, y_set, batch_size):
        self.x, self.y = x_set, y_set
        self.batch_size = batch_size
        self.lock = threading.Lock()   #Set self.lock

    def __len__(self):
        return int(np.ceil(len(self.x) / float(self.batch_size)))

    def __getitem__(self, idx):
        with self.lock:                #Use self.lock
            batch_x = self.x[idx * self.batch_size:(idx + 1) * self.batch_size]
            batch_y = self.y[idx * self.batch_size:(idx + 1) * self.batch_size]

            return ...

2020 年 4 月 24 日编辑：

使用self.lock = threading.Lock() 可能会遇到以下错误：

TypeError: can't pickle _thread.lock objects

如果发生这种情况，请尝试将 __getitem__ 内的 with self.lock: 替换为 with threading.Lock(): 并注释掉/删除 __init__ 内的 self.lock = threading.Lock()。

在类中存储lock-object 时似乎存在一些问题（参见例如this Q&A）。

---

3. 是否有任何其他方法导致 Keras 可以处理与这两个示例不同的线程安全生成器？

在我的研究过程中，我没有遇到任何其他方法。 当然，我不能百分百肯定地说。