Passed the scheduling argument through the *_generator function.

keras/engine/training.py

@@ -1631,6 +1631,7 @@ def fit_generator(self, generator,
                       max_queue_size=10,
                       workers=1,
                       use_multiprocessing=False,
+                      shuffle=True,
                       initial_epoch=0):
         """Fits the model on data yielded batch-by-batch by a Python generator.
 
@@ -1680,6 +1681,9 @@ def fit_generator(self, generator,
                 non picklable arguments to the generator
                 as they can't be passed
                 easily to children processes.
+            shuffle: whether to shuffle the data at the beginning of each
+                epoch. Only used with instances of `Sequence` (
+                keras.utils.Sequence).
             initial_epoch: epoch at which to start training
                 (useful for resuming a previous training run)
 
@@ -1781,7 +1785,8 @@ def generate_arrays_from_file(path):
         try:
             if is_sequence:
                 enqueuer = OrderedEnqueuer(generator,
-                                           use_multiprocessing=use_multiprocessing)
+                                           use_multiprocessing=use_multiprocessing,
+                                           shuffle=shuffle)
             else:
                 enqueuer = GeneratorEnqueuer(generator,
                                              use_multiprocessing=use_multiprocessing,

keras/utils/data_utils.py

@@ -351,6 +351,12 @@ def __len__(self):
         """
         raise NotImplementedError
 
+    @abstractmethod
+    def on_epoch_end(self):
+        """Method called at the end of every epoch.
+        """
+        raise NotImplementedError
+
 
 def get_index(ds, i):
     """Quick fix for Python2, otherwise, it cannot be pickled.
@@ -434,15 +440,15 @@ class OrderedEnqueuer(SequenceEnqueuer):
     # Arguments
         sequence: A `keras.utils.data_utils.Sequence` object.
         use_multiprocessing: use multiprocessing if True, otherwise threading
-        scheduling: Sequential querying of datas if 'sequential', random otherwise.
+        shuffle: whether to shuffle the data at the beginning of each epoch
     """
 
     def __init__(self, sequence,
                  use_multiprocessing=False,
-                 scheduling='sequential'):
+                 shuffle=False):
         self.sequence = sequence
         self.use_multiprocessing = use_multiprocessing
-        self.scheduling = scheduling
+        self.shuffle = shuffle
         self.workers = 0
         self.executor = None
         self.queue = None
@@ -474,14 +480,16 @@ def _run(self):
         """Function to submit request to the executor and queue the `Future` objects."""
         sequence = list(range(len(self.sequence)))
         while True:
-            if self.scheduling is not 'sequential':
+            if self.shuffle:
                 random.shuffle(sequence)
             for i in sequence:
                 if self.stop_signal.is_set():
                     return
                 self.queue.put(
                     self.executor.apply_async(get_index,
                                               (self.sequence, i)), block=True)
+            # Call the internal on epoch end.
+            self.sequence.on_epoch_end()
 
     def get(self):
         """Creates a generator to extract data from the queue.

tests/keras/engine/test_training.py

@@ -31,6 +31,9 @@ def __getitem__(self, idx):
             np.random.random((self.batch_size, 4)),
             np.random.random((self.batch_size, 3))]
 
+    def on_epoch_end(self):
+        pass
+
 
 @keras_test
 def test_check_array_lengths():

tests/keras/utils/data_utils_test.py

@@ -121,6 +121,9 @@ def __getitem__(self, item):
     def __len__(self):
         return 100
 
+    def on_epoch_end(self):
+        pass
+
 
 class FaultSequence(Sequence):
     def __getitem__(self, item):
@@ -129,6 +132,9 @@ def __getitem__(self, item):
     def __len__(self):
         return 100
 
+    def on_epoch_end(self):
+        pass
+
 
 @threadsafe_generator
 def create_generator_from_sequence_threads(ds):
@@ -199,6 +205,19 @@ def test_ordered_enqueuer_threads():
     enqueuer.stop()
 
 
+def test_ordered_enqueuer_threads_not_ordered():
+    enqueuer = OrderedEnqueuer(TestSequence([3, 200, 200, 3]),
+                               use_multiprocessing=False,
+                               shuffle=True)
+    enqueuer.start(3, 10)
+    gen_output = enqueuer.get()
+    acc = []
+    for i in range(100):
+        acc.append(next(gen_output)[0, 0, 0, 0])
+    assert acc != list(range(100)), "Order was not keep in GeneratorEnqueuer with threads"
+    enqueuer.stop()
+
+
 def test_ordered_enqueuer_processes():
     enqueuer = OrderedEnqueuer(TestSequence([3, 200, 200, 3]), use_multiprocessing=True)
     enqueuer.start(3, 10)