Merge pull request #56 from sezanzeb/events-taskqueue

Using GLib events instead of callbacks, fixed recursion problems

soundconverter/gstreamer/converter.py

@@ -221,7 +221,6 @@ def __init__(self, sound_file, name_generator):
         self.newname = None
         self.existing_behaviour = Converter.INCREMENT
         self.name_generator = name_generator
-        self.callback = lambda: None
 
         # All relevant gio settings have to be copied and remembered, so that
         # they don't suddenly change during the conversion
@@ -236,10 +235,12 @@ def __init__(self, sound_file, name_generator):
         # State
         self.command = None
         self.pipeline = None
-        self.done = False
+        self._done = False
         self.error = None
         self.output_uri = None
 
+        super().__init__()
+
     def _query_position(self):
         """Ask for the stream position of the current pipeline."""
         if self.pipeline:
@@ -253,7 +254,7 @@ def get_progress(self):
         """Fraction of how much of the task is completed."""
         duration = self.sound_file.duration
 
-        if self.done:
+        if self._done:
             return 1, duration
 
         if self.pipeline is None or duration is None:
@@ -267,7 +268,7 @@ def get_progress(self):
     def cancel(self):
         """Cancel execution of the task."""
         self._stop_pipeline()
-        self.callback()
+        self.done()
 
     def pause(self):
         """Pause execution of the task."""
@@ -352,14 +353,14 @@ def _conversion_done(self):
                 beautify_uri(input_uri),
                 self.error
             ))
-            self.callback()
+            self.done()
             return
 
         if not vfs_exists(self.temporary_filename):
             self.error = 'Expected {} to exist after conversion.'.format(
                 self.temporary_filename
             )
-            self.callback()
+            self.done()
             return
 
         # rename temporary file
@@ -398,7 +399,7 @@ def _conversion_done(self):
                 beautify_uri(newname),
                 str(error)
             ))
-            self.callback()
+            self.done()
             return
 
         assert vfs_exists(newname)
@@ -448,9 +449,12 @@ def _conversion_done(self):
                 ))
 
         self.output_uri = newname
-        self.done = True
-        self.callback()
+        self.done()
+
+    def done(self):
+        self._done = True
         self._cleanup()
+        super().done()
 
     def run(self):
         """Call this in order to run the whole Converter task."""
@@ -469,7 +473,7 @@ def run(self):
             logger.info('output file already exists, skipping \'{}\''.format(
                 beautify_uri(self.newname)
             ))
-            self._conversion_done()
+            self.done()
             return
 
         # construct a pipeline for conversion
@@ -530,7 +534,7 @@ def _on_error(self, error):
             beautify_uri(self.sound_file.uri)
         )
         self._stop_pipeline()
-        self.callback()
+        self.done()
 
     def _on_message(self, _, message):
         """Handle message events sent by gstreamer.

soundconverter/gstreamer/discoverer.py

@@ -185,14 +185,16 @@ def __init__(self, sound_files):
         self.bus = None
         self.thread = None
 
+        super().__init__()
+
     def get_progress(self):
         """Fraction of how much of the task is completed."""
         return self.discovered / len(self.sound_files), 1
 
     def cancel(self):
         """Cancel execution of the task."""
         # fast task, use case doesn't exist
-        self.callback()
+        self.done()
 
     def pause(self):
         """Pause execution of the task."""
@@ -216,6 +218,6 @@ def _on_message(self, _, message):
         """Write down that it is finished and call the callback."""
         if message.type == Gst.MessageType.EOS:
             self.running = False
-            self.callback()
+            self.done()
         else:
             self.discovered += 1

soundconverter/interface/batch.py

@@ -37,6 +37,7 @@
 from soundconverter.util.fileoperations import filename_to_uri, beautify_uri
 from soundconverter.util.logger import logger
 from soundconverter.util.formatting import format_time
+from soundconverter.interface.mainloop import gtk_iteration
 
 cli_convert = [None]
 
@@ -364,19 +365,24 @@ def __init__(self, input_files):
             len(sound_files)
         ))
         self.conversions = conversions
+
+        def finished(_=None):
+            total_time = conversions.get_duration()
+            logger.info('converted {} files in {}'.format(
+                len(sound_files),
+                format_time(total_time)
+            ))
+
+        conversions.connect('done', finished)
+
         conversions.run()
-        while conversions.running:
+
+        while not conversions.finished:
             # make the eventloop of glibs async stuff run until finished:
-            context.iteration(True)
+            context.iteration(may_block=True)
 
         # do another one to print the queue done message
-        context.iteration(True)
-
-        total_time = conversions.get_duration()
-        logger.info('converted {} files in {}'.format(
-            len(sound_files),
-            format_time(total_time)
-        ))
+        context.iteration(may_block=False)
 
 
 class CLICheck:

soundconverter/interface/filelist.py

@@ -214,7 +214,7 @@ def add_uris(self, uris, base=None, extensions=None):
 
         add_discoverers(self.discoverers, sound_files)
 
-        self.discoverers.set_on_queue_finished(self.discoverer_queue_ended)
+        self.discoverers.connect('done', self.discoverer_queue_ended)
         self.discoverers.run()
 
         self.window.set_status('{}'.format(_('Adding Files…')))

soundconverter/interface/ui.py

@@ -257,7 +257,7 @@ def do_convert(self):
         name_generator = TargetNameGenerator()
         files = self.filelist.get_files()
         self.converter_queue = TaskQueue()
-        self.converter_queue.set_on_queue_finished(self.on_queue_finished)
+        self.converter_queue.connect('done', self.on_queue_finished)
         for sound_file in files:
             gtk_iteration()
             self.converter_queue.add(Converter(
@@ -408,12 +408,12 @@ def on_aboutdialog_response(self, *args):
     def selection_changed(self, *args):
         self.set_sensitive()
 
-    def on_queue_finished(self, queue):
+    def on_queue_finished(self, __=None):
         """Should be called when all conversions are completed."""
-        total_time = queue.get_duration()
+        total_time = self.converter_queue.get_duration()
         msg = _('Conversion done in %s') % format_time(total_time)
         error_count = len([
-            task for task in queue.done
+            task for task in self.converter_queue.done
             if task.error
         ])
         if error_count > 0:

soundconverter/util/task.py

@@ -20,8 +20,14 @@
 # USA
 
 
-class Task:
+from gi.repository import GObject
+
+
+class Task(GObject.Object):
     """Abstract class of a single task."""
+    def __init__(self):
+        super().__init__()
+
     # avoid storing a variable called timer in your inheriting class
     def get_progress(self):
         """Fraction of how much of the task is completed.
@@ -49,18 +55,15 @@ def run(self):
         """Run the task."""
         raise NotImplementedError()
 
-    # don't overwrite
+    def done(self):
+        """Emit a "done" event."""
+        self.emit('done')
 
-    def set_callback(self, callback):
-        """For the Taskqueue to get notified when the Task is done.
 
-        Don't overwrite this function.
-
-        Make sure to call self.callback() when your task that inherits from
-        Task is finished.
-        """
-        def callback_wrapped():
-            # automatically provide self as argument, so that
-            # it's only required to call callback() without any argument
-            return callback(self)
-        self.callback = callback_wrapped
+GObject.signal_new(
+    'done',
+    Task,
+    GObject.SignalFlags.RUN_FIRST,
+    None,
+    []
+)

soundconverter/util/taskqueue.py

@@ -22,11 +22,13 @@
 import time
 from queue import Queue
 
+from gi.repository import GObject, GLib
+
 from soundconverter.util.settings import get_num_jobs
-from soundconverter.util.logger import logger
+from soundconverter.interface.mainloop import gtk_iteration
 
 
-class TaskQueue:
+class TaskQueue(GObject.Object):
     """Executes multiple tasks in parallel."""
     def __init__(self):
         self._on_queue_finished = None
@@ -40,6 +42,8 @@ def __init__(self):
         self.paused = False
         self._timer = Timer()
 
+        super().__init__()
+
     def add(self, task):
         """Add a task to the queue that will be executed later.
 
@@ -48,7 +52,6 @@ def add(self, task):
         task : Task
             Any object inheriting from Task
         """
-        task.set_callback(self.task_done)
         task.timer = Timer()
         self.all_tasks.append(task)
         self.pending.put(task)
@@ -116,46 +119,60 @@ def task_done(self, task):
         task : Task
             A completed task
         """
+        # avoid adding duplicate signal handlers if the queue is restarted
+        task.disconnect_by_func(self.task_done)
+
+        if task in self.done:
+            raise Exception('Duplicate task_done call')
+
         if self.finished:
             return
 
-        self.done.append(task)
         task.timer.stop()
-        if task not in self.running:
-            logger.warning('tried to remove task that was already removed')
-        else:
-            self.running.remove(task)
+
+        self.done.append(task)
+        self.running.remove(task)
+
         if self.pending.qsize() > 0:
             self.start_next()
         elif len(self.running) == 0:
             self.finished = True
             self._timer.stop()
-            if self._on_queue_finished is not None:
-                self._on_queue_finished(self)
+            self.emit('done')
 
-    def start_next(self):
+    def start_next(self, _=None):
         """Start the next task if available."""
         if self.pending.qsize() > 0:
             task = self.pending.get()
+
+            task.connect('done', self.task_done)
+
             self.running.append(task)
+
+            # - Just looping over self.pending causes too many tasks to be running in
+            # parallel
+            # - There is no semaphore mechanism for glib to do it in a simple while
+            # loop, in order to limit it to num_jobs
+            # - I don't think GLib.Mutex works with main-loop single-thread
+            # parallelization and the whole thread will stop here
+            # - Telling tasks to start the next task via a python callback or event
+            # causes the runtime to crash due to recursion with too many tasks.
+            # However, idle_add prevents those recursion depth problems
+            GLib.idle_add(task.run)
             task.timer.start()
-            task.run()
 
     def run(self):
-        """Run as many tasks as the configured number of jobs.
-        
-        Finished tasks will trigger running the next task over the task_done
-        callback.
-        """
+        """Run all tasks."""
         self.finished = False
         self._timer.start()
         num_jobs = get_num_jobs()
+
         while self.pending.qsize() > 0 and len(self.running) < num_jobs:
+            # Run as many tasks as the configured number of jobs. Finished tasks will
+            # trigger running the next task via a event
             self.start_next()
 
-    def set_on_queue_finished(self, on_queue_finished):
-        """Add a custom function to be used when the queue finishes."""
-        self._on_queue_finished = on_queue_finished
+        gtk_iteration()
 
     def get_duration(self):
         """Get for how many seconds the queue has been actively running.
@@ -218,6 +235,15 @@ def get_remaining(self):
         return remaining
 
 
+GObject.signal_new(
+    'done',
+    TaskQueue,
+    GObject.SignalFlags.RUN_FIRST,
+    None,
+    []
+)
+
+
 class Timer:
     """Time how long the TaskQueue took."""
     # separate class because I would like to not pollute the TaskQueue