From 5f4cf600b5dcbb30c62cd6e5b804655161b32227 Mon Sep 17 00:00:00 2001
From: jaseg <git@jaseg.de>
Date: Sat, 23 Apr 2022 23:05:38 +0200
Subject: Add event queue overflow handling

---
 mpv.py            | 51 ++++++++++++++++++++++++++++++++-------------------
 tests/test_mpv.py | 22 ++++++++++++++++++++++
 2 files changed, 54 insertions(+), 19 deletions(-)

diff --git a/mpv.py b/mpv.py
index 6469ebb..94d99c3 100644
--- a/mpv.py
+++ b/mpv.py
@@ -58,6 +58,9 @@ else:
 class ShutdownError(SystemError):
     pass
 
+class EventOverflowError(SystemError):
+    pass
+
 class MpvHandle(c_void_p):
     pass
 
@@ -886,8 +889,6 @@ class MPV(object):
         for event in _event_generator(self._event_handle):
             try:
                 eid = event.event_id.value
-                if not eid == MpvEventID.LOG_MESSAGE:
-                    print(event)
 
                 with self._event_handler_lock:
                     if eid == MpvEventID.SHUTDOWN:
@@ -919,8 +920,15 @@ class MPV(object):
                     if callback:
                         callback(ErrorCode.exception_for_ec(event.error), event.data)
 
+                if eid == MpvEventID.QUEUE_OVERFLOW:
+                    # cache list, since error handlers will unregister themselves
+                    for cb in list(self._command_reply_callbacks.values()):
+                        cb(EventOverflowError('libmpv event queue has flown over because events have not been processed fast enough'), None)
+
                 if eid == MpvEventID.SHUTDOWN:
                     _mpv_destroy(self._event_handle)
+                    for cb in list(self._command_reply_callbacks.values()):
+                        cb(ShutdownError('libmpv core has been shutdown'), None)
                     return
 
             except Exception as e:
@@ -971,6 +979,19 @@ class MPV(object):
         except ShutdownError:
             return
 
+    def _set_error_handler(self, future):
+        @self.event_callback('shutdown', 'queue-overflow')
+        def shutdown_handler(event):
+            nonlocal future
+            try:
+                if event.event_id.value == MpvEventID.SHUTDOWN:
+                    future.set_exception(ShutdownError('libmpv core has been shutdown'))
+                else:
+                    future.set_exception(EventOverflowError('libmpv event queue has flown over because events have not been processed fast enough'))
+            except InvalidStateError:
+                pass
+        return shutdown_handler.unregister_mpv_events
+
     @contextmanager
     def prepare_and_wait_for_property(self, name, cond=lambda val: val, level_sensitive=True, timeout=None):
         """Context manager that waits until ``cond`` evaluates to a truthy value on the named property. See
@@ -992,13 +1013,7 @@ class MPV(object):
             except InvalidStateError:
                 pass
         self.observe_property(name, observer)
-
-        @self.event_callback('shutdown')
-        def shutdown_handler(event):
-            try:
-                result.set_exception(ShutdownError('libmpv core has been shutdown'))
-            except InvalidStateError:
-                pass
+        err_unregister = self._set_error_handler(result)
 
         try:
             result.set_running_or_notify_cancel()
@@ -1012,7 +1027,7 @@ class MPV(object):
                 self.check_core_alive()
                 result.result(timeout)
         finally:
-            shutdown_handler.unregister_mpv_events()
+            err_unregister()
             self.unobserve_property(name, observer)
 
     def wait_for_event(self, *event_types, cond=lambda evt: True, timeout=None):
@@ -1041,13 +1056,6 @@ class MPV(object):
         """
         result = Future()
 
-        @self.event_callback('shutdown')
-        def shutdown_handler(event):
-            try:
-                result.set_exception(ShutdownError('libmpv core has been shutdown'))
-            except InvalidStateError:
-                pass
-
         @self.event_callback(*event_types)
         def target_handler(evt):
 
@@ -1063,6 +1071,8 @@ class MPV(object):
             except InvalidStateError:
                 pass
 
+        err_unregister = self._set_error_handler(result)
+
         try:
             result.set_running_or_notify_cancel()
             yield result
@@ -1070,7 +1080,7 @@ class MPV(object):
             result.result(timeout)
 
         finally:
-            shutdown_handler.unregister_mpv_events()
+            err_unregister()
             target_handler.unregister_mpv_events()
 
     def __del__(self):
@@ -1138,7 +1148,10 @@ class MPV(object):
                 result = result.unpack(decoder)
                 future.set_result(callback(error, result))
             except Exception as e:
-                future.set_exception(e)
+                try:
+                    future.set_exception(e)
+                except InvalidStateError:
+                    pass
 
         def abort():
             _mpv_abort_async_command(self._event_handle, id(future))
diff --git a/tests/test_mpv.py b/tests/test_mpv.py
index cf5e470..3fecfbd 100755
--- a/tests/test_mpv.py
+++ b/tests/test_mpv.py
@@ -659,6 +659,28 @@ class TestLifecycle(unittest.TestCase):
                 m.terminate()
         self.disp.stop()
 
+    def test_wait_for_prooperty_event_overflow(self):
+        self.disp = Xvfb()
+        self.disp.start()
+        handler = mock.Mock()
+        m = mpv.MPV(vo=testvo)
+        m.play(TESTVID)
+        with self.assertRaises(mpv.EventOverflowError):
+            # level_sensitive=false needed to prevent get_property on dead
+            # handle
+            with m.prepare_and_wait_for_property('mute', cond=lambda val: time.sleep(0.001)):
+                for i in range(10000):
+                    try:
+                        # We really have to try hard to fill up the queue here. Simple async commands will not work,
+                        # since then command_async will throw a memory error first. Property changes also do not work,
+                        # since they are only processsed when the event loop is idle. This here works reliably.
+                        m.command_async('script-message', 'foo', 'bar')
+                    except:
+                        pass
+        self.disp.stop()
+
+
+
     def test_wait_for_event_shutdown(self):
         self.disp = Xvfb()
         self.disp.start()
-- 
cgit