使某种形式的keras多处理/线程在Windows上运行

Question

为了加快用于训练神经网络的数据扩充，我正在尝试某种形式的并行处理，以便为GPU提供数据。目前的局限性是我生成增强数据的速度，而不是GPU训练网络的速度。

如果尝试使用multiprocessing=True生成器，则在Windows 10（v1083）64位下的Python 3.6.6中，keras 2.2.0出现以下错误：

ValueError：use_multiprocessing=TrueWindows不支持将生成器与一起使用（不跨进程边界对生成器进行编组）。而是使用单线程/进程或多线程。

我在GitHub上找到了以下内容，因此这是Windows下keras的预期行为。该链接似乎建议使用序列而不是生成器（即使错误消息似乎建议使用多线程，但我也无法弄清楚如何在keras上使用多线程而不是多处理-我可能忽略了它在文档中，但我只是没有找到它）。因此，我使用了下面的代码（使用序列修改示例），但这也没有实现加速，或者use_multiprocessing=True只是冻结而已。

我是否在这里缺少明显的东西，以了解如何使某种形式的并行生成器运行？

最小（非）工作示例：

from keras.utils import Sequence
from keras.models import Sequential
from keras.layers import Dense
from keras.utils import to_categorical
import numpy as np

class DummySequence(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 np.array(batch_x), np.array(batch_y)



x = np.random.random((100, 3))
y = to_categorical(np.random.random(100) > .5).astype(int)

seq = DummySequence(x, y, 10)

model = Sequential()
model.add(Dense(32, input_dim=3))
model.add(Dense(2, activation='softmax'))
model.compile(optimizer='rmsprop',
              loss='categorical_crossentropy',
              metrics=['accuracy'])

print('single worker')
model.fit_generator(generator=seq, 
                    steps_per_epoch = 100,
                    epochs = 2, 
                    verbose=2,
                    workers=1)
print('achieves no speed-up')
model.fit_generator(generator=seq, 
                    steps_per_epoch = 100,
                    epochs = 2, 
                    verbose=2,
                    workers=6,
                    use_multiprocessing=False)
print('Does not run')
model.fit_generator(generator=seq, 
                    steps_per_epoch = 100,
                    epochs = 2, 
                    verbose=2,
                    workers=6,
                    use_multiprocessing=True)

Answer 1

与序列结合使用 multi_processing=False 和workers=eg 4 确实有效。

我刚刚意识到，在问题的示例代码中，我没有看到加速，因为数据生成得太快。通过插入 time.sleep(2) 这变得很明显。

class DummySequence(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]
    time.sleep(2)
    return np.array(batch_x), np.array(batch_y)

x = np.random.random((100, 3))
y = to_categorical(np.random.random(100) > .5).astype(int)

seq = DummySequence(x, y, 10)

model = Sequential()
model.add(Dense(32, input_dim=3))
model.add(Dense(2, activation='softmax'))
model.compile(optimizer='rmsprop',
              loss='categorical_crossentropy',
              metrics=['accuracy'])

print('single worker')
model.fit_generator(generator=seq, 
                    steps_per_epoch = 10,
                    epochs = 2, 
                    verbose=2,
                    workers=1)

print('achieves speed-up!')
model.fit_generator(generator=seq, 
                    steps_per_epoch = 10,
                    epochs = 2, 
                    verbose=2,
                    workers=4,
                    use_multiprocessing=False)

这在我的笔记本电脑上产生了以下内容：

single worker
>>> model.fit_generator(generator=seq,
...                     steps_per_epoch = 10,
...                     epochs = 2,
...                     verbose=2,
...                     workers=1)
Epoch 1/2
 - 20s - loss: 0.6984 - acc: 0.5000
Epoch 2/2
 - 20s - loss: 0.6955 - acc: 0.5100

和

achieves speed-up!
>>> model.fit_generator(generator=seq,
...                     steps_per_epoch = 10,
...                     epochs = 2,
...                     verbose=2,
...                     workers=4,
...                     use_multiprocessing=False)
Epoch 1/2
 - 6s - loss: 0.6904 - acc: 0.5200
Epoch 2/2
 - 6s - loss: 0.6900 - acc: 0.5000

重要提示： 您可能需要self.lock = threading.Lock()in __init___，然后with self.lock:in __getitem__。尝试在 中执行绝对最低限度的要求with self.lock:，据我了解，这将是对（块运行self.xxxx时阻止多线程）的任何引用。with self.lock:

此外，如果您希望多线程来加速计算（即CPU 操作是极限），则不要期望任何加速。全局解释器锁（GIL）将阻止这种情况发生。仅当 I/O 操作受到限制时，多线程才会对您有所帮助。显然，为了加速 CPU 计算，我们需要真正的多处理，keras目前 Windows 10 不支持。也许可以手工制作一个多处理生成器（我不知道）。