GithubFollow/One-KVM
1
0
Fork 0
mirror of https://github.com/mofeng-git/One-KVM.git synced 2025-12-12 01:00:29 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

simplified AioNotifier()

Browse Source
This commit is contained in:
Maxim Devaev 2022-08-07 19:04:32 +03:00
parent aa630988cc
commit ec9785b4be
18 changed files with 43 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
kvmd/aiogp.py
View File

@ -107,7 +107,7 @@ class AioReader: # pylint: disable=too-many-instance-attributes
) )
for (pin, value) in zip(pins, lines.get_values()) for (pin, value) in zip(pins, lines.get_values())
} }
self.__loop.call_soon_threadsafe(self.__notifier.notify_sync) self.__loop.call_soon_threadsafe(self.__notifier.notify)
while not self.__stop_event.is_set(): while not self.__stop_event.is_set():
ev_lines = lines.event_wait(1) ev_lines = lines.event_wait(1)
@ -170,5 +170,5 @@ class _DebouncedValue:
value = await self.__queue.get() value = await self.__queue.get()
if self.__value != value: if self.__value != value:
self.__value = value self.__value = value
await self.__notifier.notify() self.__notifier.notify()
await asyncio.sleep(self.__debounce) await asyncio.sleep(self.__debounce)

9
kvmd/aiotools.py
View File

@ -232,10 +232,7 @@ class AioNotifier:
def __init__(self) -> None: def __init__(self) -> None:
self.__queue: "asyncio.Queue[None]" = asyncio.Queue() self.__queue: "asyncio.Queue[None]" = asyncio.Queue()
async def notify(self) -> None: def notify(self) -> None:
await self.__queue.put(None)
def notify_sync(self) -> None:
self.__queue.put_nowait(None) self.__queue.put_nowait(None)
async def wait(self, timeout: Optional[float]=None) -> None: async def wait(self, timeout: Optional[float]=None) -> None:
@ -299,7 +296,7 @@ class AioExclusiveRegion:
self.__busy = True self.__busy = True
try: try:
if self.__notifier: if self.__notifier:
await self.__notifier.notify() self.__notifier.notify()
except: # noqa: E722 except: # noqa: E722
self.__busy = False self.__busy = False
raise raise
@ -309,7 +306,7 @@ class AioExclusiveRegion:
async def exit(self) -> None: async def exit(self) -> None:
self.__busy = False self.__busy = False
if self.__notifier: if self.__notifier:
await self.__notifier.notify() self.__notifier.notify()
async def __aenter__(self) -> None: async def __aenter__(self) -> None:
await self.enter() await self.enter()

8
kvmd/apps/kvmd/server.py
View File

@ -213,13 +213,13 @@ class KvmdServer(HttpServer): # pylint: disable=too-many-arguments,too-many-ins
value = validator(value) # type: ignore value = validator(value) # type: ignore
if current_params[name] != value: if current_params[name] != value:
self.__new_streamer_params[name] = value self.__new_streamer_params[name] = value
await self.__streamer_notifier.notify() self.__streamer_notifier.notify()
return make_json_response() return make_json_response()
@exposed_http("POST", "/streamer/reset") @exposed_http("POST", "/streamer/reset")
async def __streamer_reset_handler(self, _: Request) -> Response: async def __streamer_reset_handler(self, _: Request) -> Response:
self.__reset_streamer = True self.__reset_streamer = True
await self.__streamer_notifier.notify() self.__streamer_notifier.notify()
return make_json_response() return make_json_response()
# ===== WEBSOCKET # ===== WEBSOCKET
@ -299,11 +299,11 @@ class KvmdServer(HttpServer): # pylint: disable=too-many-arguments,too-many-ins
logger.info("On-Cleanup complete") logger.info("On-Cleanup complete")
async def _on_ws_opened(self) -> None: async def _on_ws_opened(self) -> None:
await self.__streamer_notifier.notify() self.__streamer_notifier.notify()
async def _on_ws_closed(self) -> None: async def _on_ws_closed(self) -> None:
self.__hid.clear_events() self.__hid.clear_events()
await self.__streamer_notifier.notify() self.__streamer_notifier.notify()
def __has_stream_clients(self) -> bool: def __has_stream_clients(self) -> bool:
return bool(sum(map( return bool(sum(map(

4
kvmd/apps/kvmd/snapshoter.py
View File

@ -96,7 +96,7 @@ class Snapshoter: # pylint: disable=too-many-instance-attributes
logger.info("Time to take the new idle snapshot") logger.info("Time to take the new idle snapshot")
try: try:
self.__snapshoting = True self.__snapshoting = True
await notifier.notify() notifier.notify()
if not live: if not live:
await self.__wakeup() await self.__wakeup()
@ -116,7 +116,7 @@ class Snapshoter: # pylint: disable=too-many-instance-attributes
logger.exception("Unhandled exception while taking snapshot") logger.exception("Unhandled exception while taking snapshot")
finally: finally:
self.__snapshoting = False self.__snapshoting = False
await asyncio.shield(notifier.notify()) notifier.notify()
async def __wakeup(self) -> None: async def __wakeup(self) -> None:
logger = get_logger(0) logger = get_logger(0)

4
kvmd/apps/kvmd/streamer.py
View File

@ -319,7 +319,7 @@ class Streamer: # pylint: disable=too-many-instance-attributes
async def poll_state(self) -> AsyncGenerator[Dict, None]: async def poll_state(self) -> AsyncGenerator[Dict, None]:
def signal_handler(*_: Any) -> None: def signal_handler(*_: Any) -> None:
get_logger(0).info("Got SIGUSR2, checking the stream state ...") get_logger(0).info("Got SIGUSR2, checking the stream state ...")
asyncio.ensure_future(self.__notifier.notify()) self.__notifier.notify()
get_logger(0).info("Installing SIGUSR2 streamer handler ...") get_logger(0).info("Installing SIGUSR2 streamer handler ...")
asyncio.get_event_loop().add_signal_handler(signal.SIGUSR2, signal_handler) asyncio.get_event_loop().add_signal_handler(signal.SIGUSR2, signal_handler)
@ -370,7 +370,7 @@ class Streamer: # pylint: disable=too-many-instance-attributes
) )
if save: if save:
self.__snapshot = snapshot self.__snapshot = snapshot
await self.__notifier.notify() self.__notifier.notify()
return snapshot return snapshot
logger.error("Stream is offline, no signal or so") logger.error("Stream is offline, no signal or so")
except (aiohttp.ClientConnectionError, aiohttp.ServerConnectionError) as err: except (aiohttp.ClientConnectionError, aiohttp.ServerConnectionError) as err:

4
kvmd/apps/pst/server.py
View File

@ -96,10 +96,10 @@ class PstServer(HttpServer): # pylint: disable=too-many-arguments,too-many-inst
logger.info("On-Cleanup complete") logger.info("On-Cleanup complete")
async def _on_ws_opened(self) -> None: async def _on_ws_opened(self) -> None:
await self.__notifier.notify() self.__notifier.notify()
async def _on_ws_closed(self) -> None: async def _on_ws_closed(self) -> None:
await self.__notifier.notify() self.__notifier.notify()
# ===== SYSTEM TASKS # ===== SYSTEM TASKS

6
kvmd/apps/vnc/server.py
View File

@ -294,7 +294,7 @@ class _Client(RfbClient): # pylint: disable=too-many-instance-attributes
if len(last["data"]) == 0: if len(last["data"]) == 0:
# Вдруг какой-то баг # Вдруг какой-то баг
await self.__fb_notifier.notify() self.__fb_notifier.notify()
continue continue
if last["format"] == StreamFormats.JPEG: if last["format"] == StreamFormats.JPEG:
@ -305,7 +305,7 @@ class _Client(RfbClient): # pylint: disable=too-many-instance-attributes
if has_h264_key: if has_h264_key:
await self._send_fb_h264(last["data"]) await self._send_fb_h264(last["data"])
else: else:
await self.__fb_notifier.notify() self.__fb_notifier.notify()
else: else:
raise RuntimeError(f"Unknown format: {last['format']}") raise RuntimeError(f"Unknown format: {last['format']}")
last["data"] = b"" last["data"] = b""
@ -406,7 +406,7 @@ class _Client(RfbClient): # pylint: disable=too-many-instance-attributes
await self.__kvmd_session.streamer.set_params(quality, self.__desired_fps) await self.__kvmd_session.streamer.set_params(quality, self.__desired_fps)
async def _on_fb_update_request(self) -> None: async def _on_fb_update_request(self) -> None:
await self.__fb_notifier.notify() self.__fb_notifier.notify()
# ===== # =====

4
kvmd/plugins/msd/__init__.py
View File

@ -218,7 +218,7 @@ class MsdFileReader(BaseMsdReader): # pylint: disable=too-many-instance-attribu
now = time.monotonic() now = time.monotonic()
if self.__tick + 1 < now or self.__readed == self.__file_size: if self.__tick + 1 < now or self.__readed == self.__file_size:
self.__tick = now self.__tick = now
await self.__notifier.notify() self.__notifier.notify()
yield chunk yield chunk
@ -277,7 +277,7 @@ class MsdFileWriter(BaseMsdWriter): # pylint: disable=too-many-instance-attribu
now = time.monotonic() now = time.monotonic()
if self.__tick + 1 < now: if self.__tick + 1 < now:
self.__tick = now self.__tick = now
await self.__notifier.notify() self.__notifier.notify()
return self.__written return self.__written

28
kvmd/plugins/msd/otg/__init__.py
View File

@ -121,13 +121,13 @@ class _State:
async def busy(self, check_online: bool=True) -> AsyncGenerator[None, None]: async def busy(self, check_online: bool=True) -> AsyncGenerator[None, None]:
async with self._region: async with self._region:
async with self._lock: async with self._lock:
await self.__notifier.notify() self.__notifier.notify()
if check_online: if check_online:
if self.vd is None: if self.vd is None:
raise MsdOfflineError() raise MsdOfflineError()
assert self.storage assert self.storage
yield yield
await self.__notifier.notify() self.__notifier.notify()
def is_busy(self) -> bool: def is_busy(self) -> bool:
return self._region.is_busy() return self._region.is_busy()
@ -173,7 +173,7 @@ class Plugin(BaseMsd): # pylint: disable=too-many-instance-attributes
logger = get_logger(0) logger = get_logger(0)
logger.info("Using OTG gadget %r as MSD", gadget) logger.info("Using OTG gadget %r as MSD", gadget)
aiotools.run_sync(self.__reload_state()) aiotools.run_sync(self.__reload_state(notify=False))
@classmethod @classmethod
def get_plugin_options(cls) -> Dict: def get_plugin_options(cls) -> Dict:
@ -334,7 +334,7 @@ class Plugin(BaseMsd): # pylint: disable=too-many-instance-attributes
async with self.__state._region: # pylint: disable=protected-access async with self.__state._region: # pylint: disable=protected-access
try: try:
async with self.__state._lock: # pylint: disable=protected-access async with self.__state._lock: # pylint: disable=protected-access
await self.__notifier.notify() self.__notifier.notify()
assert self.__state.storage assert self.__state.storage
assert self.__state.vd assert self.__state.vd
@ -361,10 +361,7 @@ class Plugin(BaseMsd): # pylint: disable=too-many-instance-attributes
except asyncio.CancelledError: except asyncio.CancelledError:
pass pass
finally: finally:
try: self.__notifier.notify()
await asyncio.shield(self.__notifier.notify())
except asyncio.CancelledError:
pass
@contextlib.asynccontextmanager @contextlib.asynccontextmanager
async def write_image(self, name: str, size: int, remove_incomplete: Optional[bool]) -> AsyncGenerator[MsdFileWriter, None]: async def write_image(self, name: str, size: int, remove_incomplete: Optional[bool]) -> AsyncGenerator[MsdFileWriter, None]:
@ -373,7 +370,7 @@ class Plugin(BaseMsd): # pylint: disable=too-many-instance-attributes
path: str = "" path: str = ""
try: try:
async with self.__state._lock: # pylint: disable=protected-access async with self.__state._lock: # pylint: disable=protected-access
await self.__notifier.notify() self.__notifier.notify()
assert self.__state.storage assert self.__state.storage
assert self.__state.vd assert self.__state.vd
@ -395,7 +392,7 @@ class Plugin(BaseMsd): # pylint: disable=too-many-instance-attributes
chunk_size=self.__write_chunk_size, chunk_size=self.__write_chunk_size,
).open() ).open()
await self.__notifier.notify() self.__notifier.notify()
yield self.__writer yield self.__writer
self.__set_image_complete(name, self.__writer.is_complete()) self.__set_image_complete(name, self.__writer.is_complete())
@ -421,10 +418,6 @@ class Plugin(BaseMsd): # pylint: disable=too-many-instance-attributes
await asyncio.shield(self.__reload_state()) await asyncio.shield(self.__reload_state())
except asyncio.CancelledError: except asyncio.CancelledError:
pass pass
try:
await asyncio.shield(self.__notifier.notify())
except asyncio.CancelledError:
pass
@aiotools.atomic @aiotools.atomic
async def remove(self, name: str) -> None: async def remove(self, name: str) -> None:
@ -470,7 +463,6 @@ class Plugin(BaseMsd): # pylint: disable=too-many-instance-attributes
while True: while True:
# Активно ждем, пока не будут на месте все каталоги. # Активно ждем, пока не будут на месте все каталоги.
await self.__reload_state() await self.__reload_state()
await self.__notifier.notify()
if self.__state.vd: if self.__state.vd:
break break
await asyncio.sleep(5) await asyncio.sleep(5)
@ -483,7 +475,6 @@ class Plugin(BaseMsd): # pylint: disable=too-many-instance-attributes
# После установки вотчеров еще раз проверяем стейт, чтобы ничего не потерять # После установки вотчеров еще раз проверяем стейт, чтобы ничего не потерять
await self.__reload_state() await self.__reload_state()
await self.__notifier.notify()
while self.__state.vd: # Если живы после предыдущей проверки while self.__state.vd: # Если живы после предыдущей проверки
need_restart = False need_restart = False
@ -499,12 +490,11 @@ class Plugin(BaseMsd): # pylint: disable=too-many-instance-attributes
break break
if need_reload_state: if need_reload_state:
await self.__reload_state() await self.__reload_state()
await self.__notifier.notify()
except Exception: except Exception:
logger.exception("Unexpected MSD watcher error") logger.exception("Unexpected MSD watcher error")
time.sleep(1) time.sleep(1)
async def __reload_state(self) -> None: async def __reload_state(self, notify: bool=True) -> None:
logger = get_logger(0) logger = get_logger(0)
async with self.__state._lock: # pylint: disable=protected-access async with self.__state._lock: # pylint: disable=protected-access
try: try:
@ -541,6 +531,8 @@ class Plugin(BaseMsd): # pylint: disable=too-many-instance-attributes
self.__state.vd.image = None self.__state.vd.image = None
self.__state.vd.connected = False self.__state.vd.connected = False
if notify:
self.__notifier.notify()
async def __setup_initial(self) -> None: async def __setup_initial(self) -> None:
if self.__initial_image: if self.__initial_image:

2
kvmd/plugins/msd/relay/__init__.py
View File

@ -245,7 +245,7 @@ class Plugin(BaseMsd): # pylint: disable=too-many-instance-attributes
).open() ).open()
await self.__write_image_info(False) await self.__write_image_info(False)
await self.__notifier.notify() self.__notifier.notify()
yield self.__device_writer yield self.__device_writer
await self.__write_image_info(True) await self.__write_image_info(True)
finally: finally:

2
kvmd/plugins/ugpio/ezcoo.py
View File

@ -101,7 +101,7 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
(got, channel) = await aiomulti.queue_get_last(self.__channel_queue, 1) (got, channel) = await aiomulti.queue_get_last(self.__channel_queue, 1)
if got and self.__channel != channel: if got and self.__channel != channel:
self.__channel = channel self.__channel = channel
await self._notifier.notify() self._notifier.notify()
async def cleanup(self) -> None: async def cleanup(self) -> None:
if self.__proc is not None: if self.__proc is not None:

2
kvmd/plugins/ugpio/hidrelay.py
View File

@ -109,7 +109,7 @@ class Plugin(BaseUserGpioDriver):
except Exception: except Exception:
raw = -1 raw = -1
if raw != prev_raw: if raw != prev_raw:
await self._notifier.notify() self._notifier.notify()
prev_raw = raw prev_raw = raw
await asyncio.sleep(self.__state_poll) await asyncio.sleep(self.__state_poll)

4
kvmd/plugins/ugpio/hue.py
View File

@ -122,7 +122,7 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
get_logger().error("Failed Hue bulk GET request: %s", tools.efmt(err)) get_logger().error("Failed Hue bulk GET request: %s", tools.efmt(err))
self.__state = dict.fromkeys(self.__state, None) self.__state = dict.fromkeys(self.__state, None)
if self.__state != prev_state: if self.__state != prev_state:
await self._notifier.notify() self._notifier.notify()
prev_state = self.__state prev_state = self.__state
await self.__update_notifier.wait(self.__state_poll) await self.__update_notifier.wait(self.__state_poll)
@ -148,7 +148,7 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
get_logger().error("Failed Hue PUT request to pin %s: %s", pin, tools.efmt(err)) get_logger().error("Failed Hue PUT request to pin %s: %s", pin, tools.efmt(err))
raise GpioDriverOfflineError(self) raise GpioDriverOfflineError(self)
else: else:
await self.__update_notifier.notify() self.__update_notifier.notify()
def __ensure_http_session(self) -> aiohttp.ClientSession: def __ensure_http_session(self) -> aiohttp.ClientSession:
if not self.__http_session: if not self.__http_session:

2
kvmd/plugins/ugpio/ipmi.py
View File

@ -135,7 +135,7 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
await self.__update_power() await self.__update_power()
new = (self.__online, self.__power) new = (self.__online, self.__power)
if new != prev: if new != prev:
await self._notifier.notify() self._notifier.notify()
prev = new prev = new
await asyncio.sleep(self.__state_poll) await asyncio.sleep(self.__state_poll)

6
kvmd/plugins/ugpio/otgconf.py
View File

@ -76,7 +76,7 @@ class Plugin(BaseUserGpioDriver):
while True: while True:
try: try:
while True: while True:
await self._notifier.notify() self._notifier.notify()
if os.path.isfile(self.__udc_path): if os.path.isfile(self.__udc_path):
break break
await asyncio.sleep(5) await asyncio.sleep(5)
@ -84,7 +84,7 @@ class Plugin(BaseUserGpioDriver):
with Inotify() as inotify: with Inotify() as inotify:
inotify.watch(os.path.dirname(self.__udc_path), InotifyMask.ALL_MODIFY_EVENTS) inotify.watch(os.path.dirname(self.__udc_path), InotifyMask.ALL_MODIFY_EVENTS)
inotify.watch(self.__profile_path, InotifyMask.ALL_MODIFY_EVENTS) inotify.watch(self.__profile_path, InotifyMask.ALL_MODIFY_EVENTS)
await self._notifier.notify() self._notifier.notify()
while True: while True:
need_restart = False need_restart = False
need_notify = False need_notify = False
@ -97,7 +97,7 @@ class Plugin(BaseUserGpioDriver):
if need_restart: if need_restart:
break break
if need_notify: if need_notify:
await self._notifier.notify() self._notifier.notify()
except Exception: except Exception:
logger.exception("Unexpected OTG-bind watcher error") logger.exception("Unexpected OTG-bind watcher error")
await asyncio.sleep(1) await asyncio.sleep(1)

2
kvmd/plugins/ugpio/pway.py
View File

@ -103,7 +103,7 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
(got, channel) = await aiomulti.queue_get_last(self.__channel_queue, 1) (got, channel) = await aiomulti.queue_get_last(self.__channel_queue, 1)
if got and self.__channel != channel: if got and self.__channel != channel:
self.__channel = channel self.__channel = channel
await self._notifier.notify() self._notifier.notify()
async def cleanup(self) -> None: async def cleanup(self) -> None:
if self.__proc is not None: if self.__proc is not None:

4
kvmd/plugins/ugpio/tesmart.py
View File

@ -111,7 +111,7 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
except Exception: except Exception:
pass pass
if self.__active != prev_active: if self.__active != prev_active:
await self._notifier.notify() self._notifier.notify()
prev_active = self.__active prev_active = self.__active
await self.__update_notifier.wait(self.__state_poll) await self.__update_notifier.wait(self.__state_poll)
@ -127,7 +127,7 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
assert 1 <= channel <= 16 assert 1 <= channel <= 16
if state: if state:
await self.__send_command("{:c}{:c}".format(1, channel).encode()) await self.__send_command("{:c}{:c}".format(1, channel).encode())
await self.__update_notifier.notify() self.__update_notifier.notify()
await asyncio.sleep(self.__switch_delay) # Slowdown await asyncio.sleep(self.__switch_delay) # Slowdown
# ===== # =====

2
kvmd/plugins/ugpio/xh_hk4401.py
View File

@ -99,7 +99,7 @@ class Plugin(BaseUserGpioDriver): # pylint: disable=too-many-instance-attribute
(got, channel) = await aiomulti.queue_get_last(self.__channel_queue, 1) (got, channel) = await aiomulti.queue_get_last(self.__channel_queue, 1)
if got and self.__channel != channel: if got and self.__channel != channel:
self.__channel = channel self.__channel = channel
await self._notifier.notify() self._notifier.notify()
async def cleanup(self) -> None: async def cleanup(self) -> None:
if self.__proc is not None: if self.__proc is not None: