Spaces:
Running
Running
| """单工作线程同步调度；持久待同步记录由调用方管理。""" | |
| import math | |
| import threading | |
| import time | |
| class SyncScheduler: | |
| MAX_PENDING_NAMES = 128 | |
| def __init__(self, sync, interval, on_error, cooldown=None): | |
| self._sync = sync | |
| self._interval = interval | |
| self._on_error = on_error | |
| self._cooldown = cooldown or (lambda: 0) | |
| self._condition = threading.Condition() | |
| self._state = 'new' | |
| self._thread = None | |
| self._pending = set() | |
| self._all_pending = False | |
| def _delay(self): | |
| delay = float(self._interval()) | |
| if not math.isfinite(delay) or delay <= 0: | |
| raise ValueError('invalid sync interval') | |
| return delay | |
| def _start_locked(self, restart): | |
| if self._state == 'running': | |
| return False | |
| if self._state == 'stopping' or (self._state == 'stopped' and not restart): | |
| return False | |
| self._delay() | |
| self._state = 'running' | |
| self._thread = threading.Thread(target=self._run, name='ranking-hf-sync', daemon=True) | |
| try: | |
| self._thread.start() | |
| except Exception: | |
| self._state = 'new' | |
| self._thread = None | |
| raise | |
| return True | |
| def start(self, restart=False): | |
| with self._condition: | |
| return self._start_locked(restart) | |
| def request(self, file_name): | |
| with self._condition: | |
| if self._state in {'stopping', 'stopped'}: | |
| return False | |
| if not self._all_pending: | |
| self._pending.add(file_name) | |
| if len(self._pending) > self.MAX_PENDING_NAMES: | |
| self._pending.clear() | |
| self._all_pending = True | |
| self._start_locked(False) | |
| self._condition.notify_all() | |
| return True | |
| def close(self): | |
| with self._condition: | |
| thread = self._thread | |
| if thread is threading.current_thread(): | |
| raise RuntimeError('sync worker cannot join itself') | |
| self._state = 'stopping' | |
| self._pending.clear() | |
| self._all_pending = False | |
| self._condition.notify_all() | |
| if thread is not None: | |
| thread.join() | |
| with self._condition: | |
| if self._thread is thread: | |
| self._state = 'stopped' | |
| self._thread = None | |
| def _run(self): | |
| try: | |
| deadline = time.monotonic() + self._delay() | |
| while True: | |
| with self._condition: | |
| while self._state == 'running': | |
| periodic = time.monotonic() >= deadline | |
| if periodic or self._pending or self._all_pending: | |
| cooldown = float(self._cooldown()) | |
| if not math.isfinite(cooldown) or cooldown < 0: | |
| raise ValueError('invalid sync cooldown') | |
| if cooldown > 0: | |
| # 远期服务端日期也不能超过平台锁等待上限。 | |
| self._condition.wait(min(cooldown, 60)) | |
| continue | |
| names = None if periodic or self._all_pending else sorted(self._pending) | |
| self._pending.clear() | |
| self._all_pending = False | |
| break | |
| self._condition.wait(min(60, max(0, deadline - time.monotonic()))) | |
| else: | |
| return | |
| try: | |
| self._sync(names) | |
| except Exception as error: | |
| self._on_error(error) | |
| # 即时请求不能无限推迟正常批量窗口。 | |
| if periodic: | |
| deadline = time.monotonic() + self._delay() | |
| finally: | |
| with self._condition: | |
| if self._state == 'running': | |
| self._state = 'stopped' | |
| self._condition.notify_all() | |