Spaces:
Running
Running
File size: 4,118 Bytes
c8e5eaa | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 | """单工作线程同步调度；持久待同步记录由调用方管理。"""
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()
|