Fix config guard and parallel worker probing
This commit is contained in:
@@ -226,6 +226,8 @@ async def update_setting_endpoint(request):
|
||||
|
||||
if not key or value is None:
|
||||
return await handle_api_error(request, "Missing 'key' or 'value' in request", 400)
|
||||
if key not in _SETTINGS_FIELDS:
|
||||
return await handle_api_error(request, f"Unknown setting: {key}", 400)
|
||||
|
||||
config = load_config()
|
||||
if 'settings' not in config:
|
||||
@@ -254,7 +256,10 @@ async def update_master_endpoint(request):
|
||||
if "name" in data:
|
||||
config['master']['name'] = data['name']
|
||||
if "host" in data:
|
||||
config['master']['host'] = data['host']
|
||||
if data["host"] is None:
|
||||
config['master'].pop('host', None)
|
||||
else:
|
||||
config['master']['host'] = normalize_host(data["host"])
|
||||
if "port" in data:
|
||||
config['master']['port'] = data['port']
|
||||
if "cuda_device" in data:
|
||||
|
||||
+1
-1
@@ -52,7 +52,7 @@ class DistributedCollectorNode:
|
||||
|
||||
if not multi_job_id or pass_through:
|
||||
if pass_through:
|
||||
print(f"[Distributed Collector] Pass-through mode enabled, returning images unchanged")
|
||||
debug_log("Collector: pass-through mode enabled, returning images unchanged")
|
||||
return (images, audio if audio is not None else empty_audio)
|
||||
|
||||
# Use async helper to run in server loop
|
||||
|
||||
+9
-8
@@ -94,8 +94,8 @@ class DistributedQueueNode:
|
||||
|
||||
async def _fetch_worker_statuses(self, workers):
|
||||
session = await get_client_session()
|
||||
statuses = []
|
||||
for worker in workers:
|
||||
|
||||
async def _probe(worker):
|
||||
url = build_worker_url(worker, "/prompt")
|
||||
try:
|
||||
async with session.get(
|
||||
@@ -103,17 +103,18 @@ class DistributedQueueNode:
|
||||
timeout=aiohttp.ClientTimeout(total=3),
|
||||
) as resp:
|
||||
if resp.status != 200:
|
||||
continue
|
||||
return None
|
||||
data = await resp.json()
|
||||
queue_remaining = int(data.get("exec_info", {}).get("queue_remaining", 0))
|
||||
statuses.append({
|
||||
return {
|
||||
"worker": worker,
|
||||
"queue_remaining": queue_remaining,
|
||||
})
|
||||
"queue_remaining": int(data.get("exec_info", {}).get("queue_remaining", 0)),
|
||||
}
|
||||
except Exception as e:
|
||||
debug_log(f"DistributedQueue: Worker status check failed ({url}): {e}")
|
||||
return None
|
||||
|
||||
return statuses
|
||||
results = await asyncio.gather(*[_probe(worker) for worker in workers])
|
||||
return [result for result in results if result is not None]
|
||||
|
||||
def _select_worker(self, statuses):
|
||||
idle_workers = [status for status in statuses if status["queue_remaining"] == 0]
|
||||
|
||||
Reference in New Issue
Block a user