40 lines
1.4 KiB
Python
40 lines
1.4 KiB
Python
"""One writer process for the cross-process create_tables race test.
|
|
|
|
Usage: python create_tables_race_worker.py <db url> <trigger file> <events>
|
|
|
|
Builds a DatabaseSink with create_tables=True, waits until the trigger file
|
|
exists so every worker hits the empty database at the same moment, records
|
|
events, flushes, and prints {"sink_errors": n} as JSON. Prints "ready" once
|
|
imports and setup are done, so the parent can release all workers together.
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import sys
|
|
import time
|
|
|
|
from fastmcp_feedback.instrumentation import DatabaseSink, InstrumentationMiddleware
|
|
|
|
|
|
async def main(url: str, trigger: str, count: int) -> dict:
|
|
sink = DatabaseSink(url, create_tables=True)
|
|
mw = InstrumentationMiddleware([sink])
|
|
print("ready", flush=True)
|
|
deadline = time.monotonic() + 30
|
|
while not os.path.exists(trigger):
|
|
if time.monotonic() > deadline:
|
|
raise SystemExit("trigger file never appeared")
|
|
await asyncio.sleep(0.001)
|
|
for i in range(count):
|
|
mw.record_event("race.write", key=f"{os.getpid()}-{i}")
|
|
await mw.flush()
|
|
errors = mw.dispatcher.sink_errors
|
|
await mw.aclose() # also disposes the sink's engine
|
|
return {"sink_errors": errors}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
url, trigger, count = sys.argv[1], sys.argv[2], int(sys.argv[3])
|
|
print(json.dumps(asyncio.run(main(url, trigger, count))), flush=True)
|