Fix #780, hanging when the estimated frame count is lower

This problem used to be more severe, but has become less frequent with
our fix for #1132

The problem happens because we used to take `frame_count` as an
absolute truth, and we would iterate until we had processed that many
frames.  However, as we've learnt, this is just an estimate: it can
happen that the `Decoder` thread is done before we hit that frame
count.

With this change, we detect that scenario, and gracefully finish the
rendering process, making sure that all pending frames have been read
and processed and get written in the final stream.

video2x/decoder.py

@@ -66,6 +66,7 @@ def __init__(
         frame_rate: float,
         pil_ignore_max_image_pixels: bool = True,
     ) -> None:
+        self.is_done = False
         self.input_path = input_path
         self.input_width = input_width
         self.input_height = input_height
@@ -142,10 +143,11 @@ def __iter__(self):
             buffer = self.decoder.stdout.read(3 * self.input_width * self.input_height)
 
         # automatically self-join and clean up after iterations are done
-        self.join()
+        self.is_done = True
 
     def kill(self):
         self.decoder.send_signal(signal.SIGKILL)
+        self.pipe_printer.stop()
 
     def join(self):
         # close PIPEs to prevent process from getting stuck
@@ -176,18 +178,17 @@ def run(self):
         previous_frame = None
         for frame_index, frame in enumerate(self.decoder):
             while True:
-                # check for the stop signal
-                if self.running is False:
-                    self.decoder.join()
-                    return
-
                 with contextlib.suppress(Full):
                     self.tasks_queue.put(
                         (frame_index, previous_frame, frame, self.processing_settings),
                         timeout=0.1,
                     )
                     break
 
+                # check for the stop signal
+                if self.running is False:
+                    return
+
             previous_frame = frame
 
     def stop(self):

video2x/encoder.py

@@ -121,6 +121,7 @@ def __init__(
 
     def kill(self):
         self.encoder.send_signal(signal.SIGKILL)
+        self.pipe_printer.stop()
 
     def write(self, frame: Image.Image) -> None:
         """

video2x/video2x.py

@@ -375,6 +375,12 @@ def _toggle_pause(_signal_number: int = -1, _frame=None):
                 while frame_index < total_frames:
                     current_frame = processed_frames.get(frame_index)
 
+                    if current_frame is None and decoder.is_done:
+                        # this can happen when we over-estimated how
+                        # many frames there are...
+                        logger.debug("Decoder is done early, finishing...")
+                        break
+
                     if pause_flag.value is True or current_frame is None:
                         time.sleep(0.1)
                         continue
@@ -431,10 +437,14 @@ def _toggle_pause(_signal_number: int = -1, _frame=None):
             decoder_thread.stop()
             decoder_thread.join()
 
-            # clear queue and signal processors to exit
-            # multiprocessing.Queue has no Queue.queue.clear
-            while tasks_queue.empty() is not True:
-                tasks_queue.get()
+            # if we wanna exit early due to exceptions, empty the task
+            # queue by fetching everything from it.  otherwise leave
+            # any pending tasks there so they fully complete and those
+            # frames get into the video
+            if len(exceptions) > 0:
+                logger.debug("Emptying task queue")
+                while tasks_queue.empty() is not True:
+                    tasks_queue.get()
 
             logger.debug("Signaling work is done")
             for _ in range(processes):