VIDRAFT Claude Opus 5 commited on
Commit
271057e
·
1 Parent(s): e430e5d

Serve the queue round-robin by entrant; requeue abandoned claims

Browse files

There is no cap on submissions, so FIFO would let the first enthusiastic entrant own the
GPU: at ~2.5 min per ligand, ten thousand molecules is seventeen days of nobody else
being scored. Interleaving by entrant costs nothing and removes the need for a cap -
everyone's first entry is scored before anyone's second.

A worker that dies mid-job used to strand its claim forever; claims now carry a timestamp
and stale ones return to the queue.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Files changed (1) hide show
  1. app.py +42 -3
app.py CHANGED
@@ -330,6 +330,22 @@ def models():
330
  "totals": {"models": len(out), "submissions": sum(x["submissions"] for x in out)}}
331
 
332
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
333
  @app.get("/api/queue")
334
  def queue():
335
  rs = _latest()
@@ -343,14 +359,37 @@ def _auth(request: Request):
343
  raise HTTPException(401, "worker key required")
344
 
345
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
346
  @app.get("/api/pending")
347
  def pending(request: Request, limit: int = 5):
348
  _auth(request)
349
- out = [r for r in _latest() if r.get("status") == "queued"][:limit]
350
  for r in out:
351
- _append({"id": r["id"], "status": "scoring"})
352
  return {"items": [{"id": r["id"], "smiles": r["smiles"],
353
- "candidate_id": r["candidate_id"]} for r in out]}
 
354
 
355
 
356
  @app.post("/api/result")
 
330
  "totals": {"models": len(out), "submissions": sum(x["submissions"] for x in out)}}
331
 
332
 
333
+ # If a worker dies mid-job the entry is left marked "scoring" and would never be picked
334
+ # up again. Anything claimed longer ago than this is treated as abandoned and requeued.
335
+ CLAIM_TTL = int(os.environ.get("ODC_CLAIM_TTL", "3600"))
336
+
337
+
338
+ @app.get("/api/requeue_stale")
339
+ def requeue_stale(request: Request):
340
+ _auth(request)
341
+ now, n = int(time.time()), 0
342
+ for r in _latest():
343
+ if r.get("status") == "scoring" and now - (r.get("claimed_at") or 0) > CLAIM_TTL:
344
+ _append({"id": r["id"], "status": "queued"})
345
+ n += 1
346
+ return {"requeued": n}
347
+
348
+
349
  @app.get("/api/queue")
350
  def queue():
351
  rs = _latest()
 
359
  raise HTTPException(401, "worker key required")
360
 
361
 
362
+ def _round_robin(rows):
363
+ """Order queued work so that entrants take turns.
364
+
365
+ Each entrant's submissions keep their own arrival order; the queues are then read
366
+ across rather than down. Someone who submits a hundred molecules still gets every one
367
+ of them scored - just not before everyone else's first.
368
+ """
369
+ by_user = {}
370
+ for r in sorted(rows, key=lambda x: x.get("submitted_at") or 0):
371
+ who = r.get("hf_user") or r.get("display_name") or "anonymous"
372
+ by_user.setdefault(who, []).append(r)
373
+ # entrants who have been waiting longest go first within each pass
374
+ order = sorted(by_user, key=lambda u: by_user[u][0].get("submitted_at") or 0)
375
+ out, depth = [], 0
376
+ while any(len(by_user[u]) > depth for u in order):
377
+ for u in order:
378
+ if len(by_user[u]) > depth:
379
+ out.append(by_user[u][depth])
380
+ depth += 1
381
+ return out
382
+
383
+
384
  @app.get("/api/pending")
385
  def pending(request: Request, limit: int = 5):
386
  _auth(request)
387
+ out = _round_robin([r for r in _latest() if r.get("status") == "queued"])[:limit]
388
  for r in out:
389
+ _append({"id": r["id"], "status": "scoring", "claimed_at": int(time.time())})
390
  return {"items": [{"id": r["id"], "smiles": r["smiles"],
391
+ "candidate_id": r["candidate_id"],
392
+ "entrant": r.get("hf_user") or r.get("display_name")} for r in out]}
393
 
394
 
395
  @app.post("/api/result")