Count server items per run instead of per process
Browse files
app.py
CHANGED
|
@@ -9,6 +9,7 @@
|
|
| 9 |
# After the fix: every job completes.
|
| 10 |
|
| 11 |
import inspect
|
|
|
|
| 12 |
import time
|
| 13 |
|
| 14 |
import gradio as gr
|
|
@@ -18,12 +19,14 @@ from gradio_client import client as client_module
|
|
| 18 |
|
| 19 |
HAS_HELPER_POOL = "helper_executor" in inspect.getsource(client_module.Client.__init__)
|
| 20 |
|
|
|
|
| 21 |
items_handled = 0
|
| 22 |
|
| 23 |
|
| 24 |
def echo(text: list[str]) -> list[list[str]]:
|
| 25 |
global items_handled
|
| 26 |
-
|
|
|
|
| 27 |
return [text]
|
| 28 |
|
| 29 |
|
|
@@ -35,27 +38,39 @@ def self_url() -> str:
|
|
| 35 |
def run_load_test(count, max_workers):
|
| 36 |
count = int(count)
|
| 37 |
max_workers = int(max_workers)
|
|
|
|
|
|
|
|
|
|
|
|
|
| 38 |
client = Client(self_url(), verbose=False, max_workers=max_workers)
|
| 39 |
|
| 40 |
started = time.time()
|
| 41 |
jobs = [client.submit(str(i), api_name="/echo") for i in range(count)]
|
| 42 |
|
| 43 |
done = 0
|
|
|
|
| 44 |
while time.time() - started < 90:
|
| 45 |
done = sum(1 for job in jobs if job.done())
|
|
|
|
|
|
|
| 46 |
elapsed = time.time() - started
|
| 47 |
if done == count:
|
| 48 |
-
yield
|
|
|
|
|
|
|
|
|
|
| 49 |
return
|
| 50 |
yield (
|
| 51 |
f"{done}/{count} jobs completed after {elapsed:.0f}s\n"
|
| 52 |
-
f"server
|
| 53 |
)
|
| 54 |
time.sleep(1)
|
| 55 |
|
| 56 |
yield (
|
| 57 |
f"STALLED: {done}/{count} jobs completed and then nothing for the rest of "
|
| 58 |
-
f"the 90s window
|
|
|
|
|
|
|
| 59 |
f"The client's worker threads are wedged for good now, so restart the "
|
| 60 |
f"Space before running this again."
|
| 61 |
)
|
|
@@ -70,7 +85,11 @@ with gr.Blocks() as demo:
|
|
| 70 |
"Submitting many jobs in a tight loop through `Client.submit()` used to "
|
| 71 |
"wedge the client permanently: jobs stopped completing well before the "
|
| 72 |
"last one, and `job.done()` never became True for the rest even though "
|
| 73 |
-
"the server had already done the work."
|
|
|
|
|
|
|
|
|
|
|
|
|
| 74 |
)
|
| 75 |
with gr.Row():
|
| 76 |
with gr.Column():
|
|
|
|
| 9 |
# After the fix: every job completes.
|
| 10 |
|
| 11 |
import inspect
|
| 12 |
+
import threading
|
| 13 |
import time
|
| 14 |
|
| 15 |
import gradio as gr
|
|
|
|
| 19 |
|
| 20 |
HAS_HELPER_POOL = "helper_executor" in inspect.getsource(client_module.Client.__init__)
|
| 21 |
|
| 22 |
+
items_lock = threading.Lock()
|
| 23 |
items_handled = 0
|
| 24 |
|
| 25 |
|
| 26 |
def echo(text: list[str]) -> list[list[str]]:
|
| 27 |
global items_handled
|
| 28 |
+
with items_lock:
|
| 29 |
+
items_handled += len(text)
|
| 30 |
return [text]
|
| 31 |
|
| 32 |
|
|
|
|
| 38 |
def run_load_test(count, max_workers):
|
| 39 |
count = int(count)
|
| 40 |
max_workers = int(max_workers)
|
| 41 |
+
with items_lock:
|
| 42 |
+
# Counted per run: the global keeps accumulating for as long as the
|
| 43 |
+
# Space's process lives, and a stalled run leaves items behind.
|
| 44 |
+
baseline = items_handled
|
| 45 |
client = Client(self_url(), verbose=False, max_workers=max_workers)
|
| 46 |
|
| 47 |
started = time.time()
|
| 48 |
jobs = [client.submit(str(i), api_name="/echo") for i in range(count)]
|
| 49 |
|
| 50 |
done = 0
|
| 51 |
+
handled = 0
|
| 52 |
while time.time() - started < 90:
|
| 53 |
done = sum(1 for job in jobs if job.done())
|
| 54 |
+
with items_lock:
|
| 55 |
+
handled = items_handled - baseline
|
| 56 |
elapsed = time.time() - started
|
| 57 |
if done == count:
|
| 58 |
+
yield (
|
| 59 |
+
f"FINISHED: {count}/{count} jobs completed in {elapsed:.0f}s\n"
|
| 60 |
+
f"server handled {handled} items for this run"
|
| 61 |
+
)
|
| 62 |
return
|
| 63 |
yield (
|
| 64 |
f"{done}/{count} jobs completed after {elapsed:.0f}s\n"
|
| 65 |
+
f"server handled {handled} items for this run"
|
| 66 |
)
|
| 67 |
time.sleep(1)
|
| 68 |
|
| 69 |
yield (
|
| 70 |
f"STALLED: {done}/{count} jobs completed and then nothing for the rest of "
|
| 71 |
+
f"the 90s window.\n"
|
| 72 |
+
f"The server handled {handled} items for this run: the {done} completed "
|
| 73 |
+
f"jobs, plus the ones still stuck in flight.\n"
|
| 74 |
f"The client's worker threads are wedged for good now, so restart the "
|
| 75 |
f"Space before running this again."
|
| 76 |
)
|
|
|
|
| 85 |
"Submitting many jobs in a tight loop through `Client.submit()` used to "
|
| 86 |
"wedge the client permanently: jobs stopped completing well before the "
|
| 87 |
"last one, and `job.done()` never became True for the rest even though "
|
| 88 |
+
"the server had already done the work.\n\n"
|
| 89 |
+
"That is why the item count runs ahead of the completion count during a "
|
| 90 |
+
"stalled run. The client has POSTed the jobs it is holding, so the server "
|
| 91 |
+
"handles the completed ones plus up to `max_workers` more whose results "
|
| 92 |
+
"the client can no longer read."
|
| 93 |
)
|
| 94 |
with gr.Row():
|
| 95 |
with gr.Column():
|