Skip to content

Commit 5b4e5b4

Browse files
CamDavidsonPilonCamDavidsonPilon
committed
add priorities to huey, which I think should prevent jams
1 parent 186c59a commit 5b4e5b4

File tree

8 files changed

+43
-33
lines changed

8 files changed

+43
-33
lines changed

‎asset-manifest.json‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
{
22
"files": {
33
"main.css": "/static/css/main.02152627.css",
4-
"main.js": "/static/js/main.7aa869dd.js",
4+
"main.js": "/static/js/main.c1b772c6.js",
55
"static/media/roboto-all-500-normal.woff": "/static/media/roboto-all-500-normal.0ab669b7a0d19b178f57.woff",
66
"static/media/roboto-all-700-normal.woff": "/static/media/roboto-all-700-normal.a457fde362a540fcadff.woff",
77
"static/media/roboto-all-400-normal.woff": "/static/media/roboto-all-400-normal.c5d001fa922fa66a147f.woff",
@@ -36,10 +36,10 @@
3636
"static/media/roboto-greek-ext-700-normal.woff2": "/static/media/roboto-greek-ext-700-normal.bd9854c751441ccc1a70.woff2",
3737
"index.html": "/index.html",
3838
"main.02152627.css.map": "/static/css/main.02152627.css.map",
39-
"main.7aa869dd.js.map": "/static/js/main.7aa869dd.js.map"
39+
"main.c1b772c6.js.map": "/static/js/main.c1b772c6.js.map"
4040
},
4141
"entrypoints": [
4242
"static/css/main.02152627.css",
43-
"static/js/main.7aa869dd.js"
43+
"static/js/main.c1b772c6.js"
4444
]
4545
}

‎index.html‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1 +1 @@
1-
<!doctype html><html lang="en"><head><meta charset="utf-8"/><link rel="icon" href="/favicon.ico"/><meta name="viewport" content="width=device-width,initial-scale=1"/><meta name="theme-color" content="#000000"/><meta name="description" content="Pioreactor"/><link rel="apple-touch-icon" href="/logo192.png"/><link rel="manifest" href="/manifest.json"/><script defer="defer" src="/static/js/main.7aa869dd.js"></script><link href="/static/css/main.02152627.css" rel="stylesheet"></head><body><div id="root"></div></body></html>
1+
<!doctype html><html lang="en"><head><meta charset="utf-8"/><link rel="icon" href="/favicon.ico"/><meta name="viewport" content="width=device-width,initial-scale=1"/><meta name="theme-color" content="#000000"/><meta name="description" content="Pioreactor"/><link rel="apple-touch-icon" href="/logo192.png"/><link rel="manifest" href="/manifest.json"/><script defer="defer" src="/static/js/main.c1b772c6.js"></script><link href="/static/css/main.02152627.css" rel="stylesheet"></head><body><div id="root"></div></body></html>

‎pioreactorui/tasks.py‎

Lines changed: 10 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -75,7 +75,7 @@ def initialized():
7575
logger.info(f"Cache directory = {CACHE_DIR}")
7676

7777

78-
@huey.task()
78+
@huey.task(priority=10)
7979
def pio_run(*args: str, env: dict[str, str] = {}) -> bool:
8080
# for long running pio run jobs where we don't care about the output / status
8181
command = ("nohup", PIO_EXECUTABLE, "run") + args
@@ -178,7 +178,7 @@ def pio_run_export_experiment_data(*args: str, env: dict[str, str] = {}) -> bool
178178
return result.returncode == 0
179179

180180

181-
@huey.task()
181+
@huey.task(priority=100)
182182
def pio_kill(*args: str, env: dict[str, str] = {}) -> bool:
183183
logger.info(f'Executing `{join(("pio", "kill") + args)}`, {env=}')
184184
result = run((PIO_EXECUTABLE, "kill") + args, env=dict(os.environ) | env)
@@ -306,7 +306,7 @@ def write_config_and_sync(
306306
return (False, "Could not sync configs to all Pioreactors.")
307307

308308

309-
@huey.task()
309+
@huey.task(priority=10)
310310
def post_to_worker(worker: str, endpoint: str, json: dict | None = None) -> tuple[str, Any]:
311311
try:
312312
r = post_into(resolve_to_address(worker), endpoint, json=json, timeout=1)
@@ -324,7 +324,7 @@ def post_to_worker(worker: str, endpoint: str, json: dict | None = None) -> tupl
324324
return worker, None
325325

326326

327-
@huey.task()
327+
@huey.task(priority=5)
328328
def multicast_post_across_cluster(
329329
endpoint: str, workers: list[str], json: dict | None = None
330330
) -> dict[str, Any]:
@@ -338,7 +338,7 @@ def multicast_post_across_cluster(
338338
} # add a timeout so that we don't hold up a thread forever.
339339

340340

341-
@huey.task()
341+
@huey.task(priority=10)
342342
def get_from_worker(
343343
worker: str, endpoint: str, json: dict | None = None, timeout=1.0, return_raw=False
344344
) -> tuple[str, Any]:
@@ -361,7 +361,7 @@ def get_from_worker(
361361
return worker, None
362362

363363

364-
@huey.task()
364+
@huey.task(priority=5)
365365
def multicast_get_across_cluster(
366366
endpoint: str,
367367
workers: list[str],
@@ -379,7 +379,7 @@ def multicast_get_across_cluster(
379379
} # add a timeout so that we don't hold up a thread forever.
380380

381381

382-
@huey.task()
382+
@huey.task(priority=10)
383383
def patch_to_worker(worker: str, endpoint: str, json: dict | None = None) -> tuple[str, Any]:
384384
try:
385385
r = patch_into(resolve_to_address(worker), endpoint, json=json, timeout=1)
@@ -397,7 +397,7 @@ def patch_to_worker(worker: str, endpoint: str, json: dict | None = None) -> tup
397397
return worker, None
398398

399399

400-
@huey.task()
400+
@huey.task(priority=5)
401401
def multicast_patch_across_cluster(
402402
endpoint: str, workers: list[str], json: dict | None = None
403403
) -> dict[str, Any]:
@@ -411,7 +411,7 @@ def multicast_patch_across_cluster(
411411
} # add a timeout so that we don't hold up a thread forever.
412412

413413

414-
@huey.task()
414+
@huey.task(priority=10)
415415
def delete_from_worker(worker: str, endpoint: str, json: dict | None = None) -> tuple[str, Any]:
416416
try:
417417
r = delete_from(resolve_to_address(worker), endpoint, json=json, timeout=1)
@@ -429,7 +429,7 @@ def delete_from_worker(worker: str, endpoint: str, json: dict | None = None) ->
429429
return worker, None
430430

431431

432-
@huey.task()
432+
@huey.task(priority=5)
433433
def multicast_delete_across_cluster(
434434
endpoint: str, workers: list[str], json: dict | None = None
435435
) -> dict[str, Any]:

‎pioreactorui/unit_api.py‎

Lines changed: 10 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,6 @@
77
from pathlib import Path
88
from subprocess import run
99
from time import sleep
10-
from time import time
1110

1211
from flask import abort
1312
from flask import Blueprint
@@ -26,7 +25,6 @@
2625
from pioreactor.config import get_leader_hostname
2726
from pioreactor.structs import CalibrationBase
2827
from pioreactor.structs import subclass_union
29-
from pioreactor.utils import local_intermittent_storage
3028
from pioreactor.utils import local_persistent_storage
3129
from pioreactor.utils.timing import current_utc_timestamp
3230
from pioreactor.utils.timing import to_datetime
@@ -41,6 +39,7 @@
4139
from .config import huey
4240
from .utils import attach_cache_control
4341
from .utils import create_task_response
42+
from .utils import is_rate_limited
4443
from pioreactorui import structs
4544

4645

@@ -242,18 +241,6 @@ def dir_listing(req_path: str):
242241
## RUNNING JOBS CONTROL
243242

244243

245-
def is_rate_limited(job: str, expire_time_seconds=1.0) -> bool:
246-
"""
247-
Check if the user has made a request within the debounce duration.
248-
"""
249-
with local_intermittent_storage("debounce") as cache:
250-
if cache.get(job) and (time() - cache.get(job)) < expire_time_seconds:
251-
return True
252-
else:
253-
cache.set(job, time())
254-
return False
255-
256-
257244
@unit_api.route("/jobs/run/job_name/<job>", methods=["PATCH", "POST"])
258245
def run_job(job: str) -> ResponseReturnValue:
259246
"""
@@ -336,6 +323,15 @@ def get_all_running_jobs() -> ResponseReturnValue:
336323
return jsonify(jobs)
337324

338325

326+
@unit_api.route("/jobs/running/<job>", methods=["GET"])
327+
def get_running_job(job) -> ResponseReturnValue:
328+
jobs = query_temp_local_metadata_db(
329+
"SELECT * FROM pio_job_metadata where is_running=1 and job_name=?", (job,)
330+
)
331+
332+
return jsonify(jobs)
333+
334+
339335
@unit_api.route("/long_running_jobs/running", methods=["GET"])
340336
def get_all_long_running_jobs() -> ResponseReturnValue:
341337
jobs = query_temp_local_metadata_db(

‎pioreactorui/utils.py‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,10 +3,12 @@
33
from __future__ import annotations
44

55
import re
6+
from time import time
67

78
from flask import jsonify
89
from flask import Response
910
from flask.typing import ResponseReturnValue
11+
from pioreactor.utils import local_intermittent_storage
1012
from pioreactor.whoami import get_unit_name
1113

1214

@@ -46,3 +48,15 @@ def is_valid_unix_filename(filename: str) -> bool:
4648
and "/" not in filename
4749
and "\0" not in filename
4850
)
51+
52+
53+
def is_rate_limited(job: str, expire_time_seconds=1.0) -> bool:
54+
"""
55+
Check if the user has made a request within the debounce duration.
56+
"""
57+
with local_intermittent_storage("debounce") as cache:
58+
if cache.get(job) and (time() - cache.get(job)) < expire_time_seconds:
59+
return True
60+
else:
61+
cache.set(job, time())
62+
return False

static/js/main.7aa869dd.js renamed to static/js/main.c1b772c6.js

Lines changed: 4 additions & 4 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

static/js/main.7aa869dd.js.LICENSE.txt renamed to static/js/main.c1b772c6.js.LICENSE.txt

File renamed without changes.

static/js/main.7aa869dd.js.map renamed to static/js/main.c1b772c6.js.map

Lines changed: 1 addition & 1 deletion
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

0 commit comments

Comments
 (0)