Обучение ГАН в керасе с помощью .fit_generator ()

Question

Я тренировал условную архитектуру GAN, аналогичную Pix2Pix, со следующим циклом обучения:

for epoch in range(start_epoch, end_epoch):
    for batch_i, (input_batch, target_batch) in enumerate(dataLoader.load_batch(batch_size)):
                fake_batch= self.generator.predict(input_batch)

                d_loss_real = self.discriminator.train_on_batch(target_batch, valid)
                d_loss_fake = self.discriminator.train_on_batch(fake_batch, invalid)
                d_loss = np.add(d_loss_fake, d_loss_real) * 0.5

                g_loss = self.combined.train_on_batch([target_batch, input_batch], [valid, target_batch])

Теперь это работает хорошо, но не очень эффективно, поскольку загрузчик данных быстро становится узким местом. мудрый. Я изучил функцию .fit_generator (), которую предоставляет keras, которая позволяет генератору работать в рабочем потоке и работает намного быстрее.

self.combined.fit_generator(generator=trainLoader,
                                    validation_data=evalLoader
                                    callbacks=[checkpointCallback, historyCallback],
                                    workers=1,
                                    use_multiprocessing=True)

Мне потребовалось некоторое время, чтобы понять, что это неправильно, я больше не тренировал свой генератор и дискриминатор отдельно, а дискриминатор не тренировался вообще, так как он установлен на trainable = Falseв комбинированной модели, по сути, разрушающей любой вид состязательной потери, и я мог бы также обучить свой генератор самостоятельно с помощью MSE.

Теперь мой вопрос заключается в том, есть ли какая-то работа вокруг, такая как тренировка моего дискриминаторав пользовательском обратном вызове, который запускается в каждом пакете метода .fit_generator ()? Можно реализовать создание пользовательских обратных вызовов, например, вот так:

class MyCustomCallback(tf.keras.callbacks.Callback):
  def on_train_batch_end(self, batch, logs=None):
    discriminator.train_on_batch()

Другая возможность - распараллелить исходный цикл обучения, но я боюсь, что у меня нет времени, чтобы сделать это прямо сейчас.

Answer 1

Я создал этот параллельный итератор именно для этой цели. Я использую его в своих тренировках;

Вот как вы его используете:

for epoch, batchIndex, originalBatchIndex, xAndY in ParallelIterator(
                                       generator, 
                                       epochs, 
                                       shuffle_bool, 
                                       use_on_epoch_end_from_generator_bool,
                                       workers = 8, 
                                       queue_size=10):
    #loop content
    x_train_batch, y_train_batch = xAndY
    model.train_on_batch(x_train_batch, y_train_batch)

generator должен быть ваш dataloader, но он должен быть keras.utils.Sequence, а не просто генератор урожая.

Но это не очень сложно адаптировать, если вам нужно. (Я просто не знаю, будет ли оно правильно распараллеливаться, хотя я не знаю, можно ли правильно распараллелить циклы yield)
В приведенном ниже определении итератора вы должны заменить:

• len(keras_sequence) с steps_per_epoch
• keras_sequence[i] с next(keras_sequence)
• use_on_epoch_end = False

И это определение итератора:

import multiprocessing.dummy as mp

#A generator that wraps a Keras Sequence and simulates a `fit_generator` behavior for custom training loops
#It will also work with any iterator that has `__len__` and `__getitem__`.    
def ParallelIterator(keras_sequence, epochs, shuffle, use_on_epoch_end, workers = 4, queue_size = 10):

    sourceQueue = mp.Queue()                     #queue for getting batch indices
    batchQueue = mp.Queue(maxsize = queue_size)  #queue for getting actual batches 
    indices = np.arange(len(keras_sequence))     #array of indices to be shuffled

    use_on_epoch_end = 'on_epoch_end' in dir(keras_sequence) if use_on_epoch_end == True else False
    batchesLeft = 0

#     printQueue = mp.Queue()                      #queue for printing messages
#     import threading
#     screenLock = threading.Semaphore(value=1)
#     totalWorkers= 0

#     def printer():
#         nonlocal printQueue, printing
#         while printing:
#             while not printQueue.empty():
#                 text = printQueue.get(block=True)
#                 screenLock.acquire()
#                 print(text)
#                 screenLock.release()

    #fills the batch indices queue (called when sourceQueue is empty -> a few batches before an epoch ends)
    def fillSource():
        nonlocal batchesLeft

#         printQueue.put("Iterator: fill source - source qsize = " + str(sourceQueue.qsize()))
        if shuffle == True:
            np.random.shuffle(indices)

        #puts the indices in the indices queue
        batchesLeft += len(indices)
#         printQueue.put("Iterator: batches left:" + str(batchesLeft))
        for i in indices:
            sourceQueue.put(i)

    #function that will load batches from the Keras Sequence
    def worker():
        nonlocal sourceQueue, batchQueue, keras_sequence, batchesLeft
#         nonlocal printQueue, totalWorkers
#         totalWorkers += 1
#         thisWorker = totalWorkers

        while True:
#             printQueue.put('Worker: ' + str(thisWorker) + ' will try to get item')
            index = sourceQueue.get(block = True) #get index from the queue
#             printQueue.put('Worker: ' + str(thisWorker) + ' got item ' +  str(index) + " - source q size = " + str(sourceQueue.qsize()))

            if index is None:
                break

            item = keras_sequence[index] #get batch from the sequence
            batchesLeft -= 1
#             printQueue.put('Worker: ' + str(thisWorker) + ' batches left ' + str(batchesLeft))

            batchQueue.put((index,item), block=True) #puts batch in the batch queue
#             printQueue.put('Worker: ' + str(thisWorker) + ' added item ' + str(index) + ' - queue: ' + str(batchQueue.qsize()))

#         printQueue.put("hitting end of worker" + str(thisWorker))

#       #printing pool that will print messages from the print queue
#     printing = True
#     printPool = mp.Pool(1, printer)

    #creates the thread pool that will work automatically as we get from the batch queue
    pool = mp.Pool(workers, worker)    
    fillSource()   #at this point, data starts being taken and stored in the batchQueue

    #generation loop
    for epoch in range(epochs):

        #if not waiting for epoch end synchronization, always keeps 1 epoch filled ahead
        if (use_on_epoch_end == False):
            if epoch + 1 < epochs: #only fill if not last epoch
                fillSource()

        for batch in range(len(keras_sequence)):

            #if waiting for epoch end synchronization, wait for workers to have no batches left to get, then call epoch end and fill
            if use_on_epoch_end == True:
                if batchesLeft == 0:
                    keras_sequence.on_epoch_end()
                    if epoch + 1 < epochs:  #only fill if not last epoch
                        fillSource()
                    else:
                        batchesLeft = -1   #in the last epoch, prevents from calling epoch end again and again

            #yields batches for the outside loop that is using this generator
            originalIndex, batchItems = batchQueue.get(block = True)
            yield epoch, batch, originalIndex, batchItems


#         print("iterator epoch end")
#     printQueue.put("closing threads")

    #terminating the pool - add None to the queue so any blocked worker gets released
    for i in range(workers):
        sourceQueue.put(None)
    pool.terminate()
    pool.close()
    pool.join()
#     printQueue.put("terminated")

#     printing = False
#     printPool.terminate()
#     printPool.close()
#     printPool.join()


    del pool,sourceQueue,batchQueue
#     del printPool, printQueue