vincewin commited on
Commit
f2ef87a
·
verified ·
1 Parent(s): 21bf9a1

queue pressure > ZERO_PRESSURE_N(4) promotes backup to active drain (claim without deferral)

Browse files
Files changed (1) hide show
  1. app.py +14 -1
app.py CHANGED
@@ -73,6 +73,11 @@ TMIN = "%Y-%m-%dT%H:%MZ"
73
  QPREFIX = "events/queue"
74
  REPO = os.environ.get("CREST_FEEDBACK_REPO", "vincewin/CREST_data")
75
  ZERO_DELAY_S = float(os.environ.get("ZERO_DELAY_S", "420")) # HPC first shot
 
 
 
 
 
76
  STALE_S = float(os.environ.get("ZERO_STALE_S", "600"))
77
  HB_S = float(os.environ.get("ZERO_HB_S", "240"))
78
  POLL_S = float(os.environ.get("ZERO_POLL_S", "60"))
@@ -159,6 +164,14 @@ def _scan():
159
  ids = sorted({os.path.basename(f)[:-5] for f in qf
160
  if f.endswith(".json") and not f.endswith(".failed.json")})
161
  now = datetime.datetime.utcnow()
 
 
 
 
 
 
 
 
162
  for ev in ids:
163
  if f"{QPREFIX}/{ev}.forcing.tar.gz" not in qf:
164
  continue
@@ -172,7 +185,7 @@ def _scan():
172
  ).total_seconds()
173
  except Exception:
174
  age = 0.0
175
- if age < ZERO_DELAY_S:
176
  continue
177
  w, s, e, n = spec["bbox_basin"]
178
  res = DEM_RES_DEG["1"]
 
73
  QPREFIX = "events/queue"
74
  REPO = os.environ.get("CREST_FEEDBACK_REPO", "vincewin/CREST_data")
75
  ZERO_DELAY_S = float(os.environ.get("ZERO_DELAY_S", "420")) # HPC first shot
76
+ # Backup -> active drain under pressure (user directive 2026-08-14): with
77
+ # more than this many jobs pending, skip the polite deferral and claim the
78
+ # next unclaimed event immediately — one solver per event, so this Space
79
+ # always works a DIFFERENT event than the HPC.
80
+ PRESSURE_N = int(os.environ.get("ZERO_PRESSURE_N", "4"))
81
  STALE_S = float(os.environ.get("ZERO_STALE_S", "600"))
82
  HB_S = float(os.environ.get("ZERO_HB_S", "240"))
83
  POLL_S = float(os.environ.get("ZERO_POLL_S", "60"))
 
164
  ids = sorted({os.path.basename(f)[:-5] for f in qf
165
  if f.endswith(".json") and not f.endswith(".failed.json")})
166
  now = datetime.datetime.utcnow()
167
+ # queue pressure: backlog > PRESSURE_N promotes this Space from deferred
168
+ # backup to active drain (claim immediately; claims keep events disjoint)
169
+ backlog = sum(1 for ev in ids if f"{QPREFIX}/{ev}.forcing.tar.gz" in qf)
170
+ delay = 0.0 if backlog > PRESSURE_N else ZERO_DELAY_S
171
+ if backlog > PRESSURE_N and W.get("pressure") != backlog:
172
+ W["pressure"] = backlog
173
+ log(f"queue pressure: {backlog} pending jobs > {PRESSURE_N} — "
174
+ f"claiming without deferral")
175
  for ev in ids:
176
  if f"{QPREFIX}/{ev}.forcing.tar.gz" not in qf:
177
  continue
 
185
  ).total_seconds()
186
  except Exception:
187
  age = 0.0
188
+ if age < delay:
189
  continue
190
  w, s, e, n = spec["bbox_basin"]
191
  res = DEM_RES_DEG["1"]