summaryrefslogtreecommitdiff
path: root/mpv.py
diff options
context:
space:
mode:
authorjaseg <git@jaseg.de>2023-02-26 19:46:59 +0100
committerjaseg <git@jaseg.de>2023-02-26 19:47:20 +0100
commitf9a655e7ca39a29d71baa177bac94e7be04f6936 (patch)
treead638bf522e19df253e04e66fbfea4ad8e468eaf /mpv.py
parentd7664eb6716cb35a116b402da952dcb5ae0f01bd (diff)
downloadpython-mpv-f9a655e7ca39a29d71baa177bac94e7be04f6936.tar.gz
python-mpv-f9a655e7ca39a29d71baa177bac94e7be04f6936.tar.bz2
python-mpv-f9a655e7ca39a29d71baa177bac94e7be04f6936.zip
Add cross-thread exception handling for event loop and stream callbacks
Diffstat (limited to 'mpv.py')
-rw-r--r--mpv.py122
1 files changed, 89 insertions, 33 deletions
diff --git a/mpv.py b/mpv.py
index 97c73da..6918f97 100644
--- a/mpv.py
+++ b/mpv.py
@@ -20,6 +20,7 @@
from ctypes import *
import ctypes.util
import threading
+import queue
import os
import sys
from warnings import warn
@@ -880,6 +881,7 @@ class MPV(object):
self.register_stream_protocol('python', self._python_stream_open)
self._python_streams = {}
self._python_stream_catchall = None
+ self._exception_futures = set()
self.overlay_ids = set()
self.overlays = {}
if loglevel is not None or log_handler is not None:
@@ -891,6 +893,17 @@ class MPV(object):
else:
self._event_thread = None
+ @contextmanager
+ def _enqueue_exceptions(self):
+ try:
+ yield
+ except Exception as e:
+ try:
+ fut = next(iter(self._exception_futures))
+ fut.set_exception(e)
+ except StopIteration:
+ warn(f'Unhandled exception on python-mpv event loop: {e}\n{traceback.format_exc()}', RuntimeWarning)
+
def _loop(self):
for event in _event_generator(self._event_handle):
try:
@@ -901,45 +914,51 @@ class MPV(object):
self._core_shutdown = True
for callback in self._event_callbacks:
- callback(event)
+ with self._enqueue_exceptions():
+ callback(event)
if eid == MpvEventID.PROPERTY_CHANGE:
pc = event.data
name, value, _fmt = pc.name, pc.value, pc.format
for handler in self._property_handlers[name]:
- handler(name, value)
+ with self._enqueue_exceptions():
+ handler(name, value)
if eid == MpvEventID.LOG_MESSAGE and self._log_handler is not None:
ev = event.data
- self._log_handler(ev.level, ev.prefix, ev.text)
+ with self._enqueue_exceptions():
+ self._log_handler(ev.level, ev.prefix, ev.text)
if eid == MpvEventID.CLIENT_MESSAGE:
# {'event': {'args': ['key-binding', 'foo', 'u-', 'g']}, 'reply_userdata': 0, 'error': 0, 'event_id': 16}
target, *args = event.data.args
target = target.decode("utf-8")
if target in self._message_handlers:
- self._message_handlers[target](*args)
+ with self._enqueue_exceptions():
+ self._message_handlers[target](*args)
if eid == MpvEventID.COMMAND_REPLY:
key = event.reply_userdata
callback = self._command_reply_callbacks.pop(key, None)
if callback:
- callback(ErrorCode.exception_for_ec(event.error), event.data)
+ with self._enqueue_exceptions():
+ 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)
+ with self._enqueue_exceptions():
+ 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)
+ with self._enqueue_exceptions():
+ cb(ShutdownError('libmpv core has been shutdown'), None)
return
except Exception as e:
- print('Exception inside python-mpv event loop:', file=sys.stderr)
- traceback.print_exc()
+ warn(f'Unhandled {e} inside python-mpv event loop!\n{traceback.format_exc()}', RuntimeWarning)
@property
def core_shutdown(self):
@@ -953,35 +972,35 @@ class MPV(object):
if self._core_shutdown:
raise ShutdownError('libmpv core has been shutdown')
- def wait_until_paused(self, timeout=None):
+ def wait_until_paused(self, timeout=None, catch_errors=True):
"""Waits until playback of the current title is paused or done. Raises a ShutdownError if the core is shutdown while
waiting."""
- self.wait_for_property('core-idle', timeout=timeout)
+ self.wait_for_property('core-idle', timeout=timeout, catch_errors=catch_errors)
- def wait_for_playback(self, timeout=None):
+ def wait_for_playback(self, timeout=None, catch_errors=True):
"""Waits until playback of the current title is finished. Raises a ShutdownError if the core is shutdown while
waiting.
"""
- self.wait_for_event('end_file', timeout=timeout)
+ self.wait_for_event('end_file', timeout=timeout, catch_errors=catch_errors)
- def wait_until_playing(self, timeout=None):
+ def wait_until_playing(self, timeout=None, catch_errors=True):
"""Waits until playback of the current title has started. Raises a ShutdownError if the core is shutdown while
waiting."""
- self.wait_for_property('core-idle', lambda idle: not idle, timeout=timeout)
+ self.wait_for_property('core-idle', lambda idle: not idle, timeout=timeout, catch_errors=catch_errors)
- def wait_for_property(self, name, cond=lambda val: val, level_sensitive=True, timeout=None):
+ def wait_for_property(self, name, cond=lambda val: val, level_sensitive=True, timeout=None, catch_errors=True):
"""Waits until ``cond`` evaluates to a truthy value on the named property. This can be used to wait for
properties such as ``idle_active`` indicating the player is done with regular playback and just idling around.
Raises a ShutdownError when the core is shutdown while waiting.
"""
- with self.prepare_and_wait_for_property(name, cond, level_sensitive, timeout=timeout) as result:
+ with self.prepare_and_wait_for_property(name, cond, level_sensitive, timeout=timeout, catch_errors=catch_errors) as result:
pass
return result.result()
- def wait_for_shutdown(self, timeout=None):
+ def wait_for_shutdown(self, timeout=None, catch_errors=True):
'''Wait for core to shutdown (e.g. through quit() or terminate()).'''
try:
- self.wait_for_event(None, timeout=timeout)
+ self.wait_for_event(None, timeout=timeout, catch_errors=catch_errors)
except ShutdownError:
return
@@ -999,7 +1018,7 @@ class MPV(object):
return shutdown_handler.unregister_mpv_events
@contextmanager
- def prepare_and_wait_for_property(self, name, cond=lambda val: val, level_sensitive=True, timeout=None):
+ def prepare_and_wait_for_property(self, name, cond=lambda val: val, level_sensitive=True, timeout=None, catch_errors=True):
"""Context manager that waits until ``cond`` evaluates to a truthy value on the named property. See
prepare_and_wait_for_event for usage.
Raises a ShutdownError when the core is shutdown while waiting. Re-raises any errors inside ``cond``.
@@ -1023,6 +1042,9 @@ class MPV(object):
try:
result.set_running_or_notify_cancel()
+ if catch_errors:
+ self._exception_futures.add(result)
+
yield result
rv = cond(getattr(self, name.replace('-', '_')))
@@ -1035,18 +1057,19 @@ class MPV(object):
finally:
err_unregister()
self.unobserve_property(name, observer)
+ self._exception_futures.discard(result)
- def wait_for_event(self, *event_types, cond=lambda evt: True, timeout=None):
+ def wait_for_event(self, *event_types, cond=lambda evt: True, timeout=None, catch_errors=True):
"""Waits for the indicated event(s). If cond is given, waits until cond(event) is true. Raises a ShutdownError
if the core is shutdown while waiting. This also happens when 'shutdown' is in event_types. Re-raises any error
inside ``cond``.
"""
- with self.prepare_and_wait_for_event(*event_types, cond=cond, timeout=timeout) as result:
+ with self.prepare_and_wait_for_event(*event_types, cond=cond, timeout=timeout, catch_errors=catch_errors) as result:
pass
return result.result()
@contextmanager
- def prepare_and_wait_for_event(self, *event_types, cond=lambda evt: True, timeout=None):
+ def prepare_and_wait_for_event(self, *event_types, cond=lambda evt: True, timeout=None, catch_errors=True):
"""Context manager that waits for the indicated event(s) like wait_for_event after running. If cond is given,
waits until cond(event) is true. Raises a ShutdownError if the core is shutdown while waiting. This also happens
when 'shutdown' is in event_types. Re-raises any error inside ``cond``.
@@ -1081,13 +1104,18 @@ class MPV(object):
try:
result.set_running_or_notify_cancel()
+ if catch_errors:
+ self._exception_futures.add(result)
+
yield result
+
self.check_core_alive()
result.result(timeout)
finally:
err_unregister()
target_handler.unregister_mpv_events()
+ self._exception_futures.discard(result)
def __del__(self):
if self.handle:
@@ -1772,32 +1800,60 @@ class MPV(object):
frontend = open_fn(uri.decode('utf-8'))
except ValueError:
return ErrorCode.LOADING_FAILED
+ except Exception as e:
+ try:
+ fut = next(iter(self._exception_futures))
+ fut.set_exception(e)
+ except StopIteration:
+ warnings.warn(f'Unhandled exception {e} inside stream open callback for URI {uri}\n{traceback.format_exc()}')
- def read_backend(_userdata, buf, bufsize):
- data = frontend.read(bufsize)
- for i in range(len(data)):
- buf[i] = data[i]
- return len(data)
+ return ErrorCode.LOADING_FAILED
cb_info.contents.cookie = None
+
+ def read_backend(_userdata, buf, bufsize):
+ with self._enqueue_exceptions():
+ data = frontend.read(bufsize)
+ for i in range(len(data)):
+ buf[i] = data[i]
+ return len(data)
read = cb_info.contents.read = StreamReadFn(read_backend)
- close = cb_info.contents.close = StreamCloseFn(lambda _userdata: frontend.close())
+
+ def close_backend(_userdata):
+ with self._enqueue_exceptions():
+ del self._stream_protocol_frontends[proto][uri]
+ if hasattr(frontend, 'close'):
+ frontend.close()
+ close = cb_info.contents.close = StreamCloseFn(close_backend)
seek, size, cancel = None, None, None
+
if hasattr(frontend, 'seek'):
- seek = cb_info.contents.seek = StreamSeekFn(lambda _userdata, offx: frontend.seek(offx))
+ def seek_backend(_userdata, offx):
+ with self._enqueue_exceptions():
+ return frontend.seek(offx)
+ seek = cb_info.contents.seek = StreamSeekFn(seek_backend)
+
if hasattr(frontend, 'size') and frontend.size is not None:
- size = cb_info.contents.size = StreamSizeFn(lambda _userdata: frontend.size)
+ def size_backend(_userdata):
+ with self._enqueue_exceptions():
+ return frontend.size
+ size = cb_info.contents.size = StreamSizeFn(size_backend)
+
if hasattr(frontend, 'cancel'):
- cancel = cb_info.contents.cancel = StreamCancelFn(lambda _userdata: frontend.cancel())
+ def cancel_backend(_userdata):
+ with self._enqueue_exceptions():
+ frontend.cancel()
+ cancel = cb_info.contents.cancel = StreamCancelFn(cancel_backend)
- # keep frontend and callbacks in memory forever (TODO)
+ # keep frontend and callbacks in memory until closed
frontend._registered_callbacks = [read, close, seek, size, cancel]
self._stream_protocol_frontends[proto][uri] = frontend
return 0
if proto in self._stream_protocol_cbs:
raise KeyError('Stream protocol already registered')
+ # keep backend in memory forever
self._stream_protocol_cbs[proto] = [open_backend]
_mpv_stream_cb_add_ro(self.handle, proto.encode('utf-8'), c_void_p(), open_backend)