| 12345678910111213141516171819202122232425262728293031323334353637383940 |
- """fast_load pacing: never more than the byte budget in flight (8 bytes here, 1 per copy), all complete, sync path untouched."""
- import concurrent.futures, os, sys, threading, time, types
- os.environ["DSV41_FAST_LOAD"] = "1"; os.environ["DSV41_FAST_LOAD_INFLIGHT_GB"] = repr(8 / 2**30)
- sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "adapter"))
- import fast_load
- def orig(*, executor, futures, use_async, func, func_args=(), func_kwargs=None):
- if use_async:
- futures.append(executor.submit(func, *func_args, **(func_kwargs or {})))
- else:
- func(*func_args, **(func_kwargs or {}))
- mod = types.SimpleNamespace(maybe_executor_submit=orig)
- fast_load.install_deepseek_v4(mod)
- assert mod.maybe_executor_submit is not orig
- inflight = [0]; peak = [0]; done = [0]; lock = threading.Lock()
- def work(i):
- with lock:
- inflight[0] += 1; peak[0] = max(peak[0], inflight[0])
- time.sleep(0.001)
- with lock:
- inflight[0] -= 1; done[0] += 1
- futures = []
- with concurrent.futures.ThreadPoolExecutor(24) as ex:
- t = time.time()
- for i in range(2000):
- mod.maybe_executor_submit(executor=ex, futures=futures, use_async=True, func=work, func_args=(i,))
- for f in concurrent.futures.as_completed(futures):
- f.result()
- assert done[0] == 2000 and len(futures) == 2000, (done, len(futures))
- assert peak[0] <= 8, peak
- mod.maybe_executor_submit(executor=None, futures=futures, use_async=False, func=work, func_args=(0,))
- assert done[0] == 2001
- # an exception in the wrapped submit must not leak a permit
- def boom(**kw): raise RuntimeError("x")
- mod2 = types.SimpleNamespace(maybe_executor_submit=boom); fast_load.install_deepseek_v4(mod2)
- for _ in range(20):
- try: mod2.maybe_executor_submit(executor=None, futures=[], use_async=True, func=work)
- except RuntimeError: pass
- print(f"pacing OK: peak in flight {peak[0]} <= 8, 2000 copies in {time.time()-t:.2f}s")
|
