Dedicated hardware poller + read-only consumers #4
15
build.sh
15
build.sh
@@ -13,14 +13,15 @@ REG="docker.adamksmith.xyz"
|
||||
|
||||
cd "$(dirname "$0")"
|
||||
|
||||
TARGET="${1:-all}"
|
||||
case "$TARGET" in
|
||||
poller) TARGETS="poller" ;;
|
||||
web) TARGETS="web" ;;
|
||||
all) TARGETS="poller web" ;;
|
||||
*) echo "usage: $0 [poller|web|all] [message]" >&2; exit 1 ;;
|
||||
# First arg is the target if it's a known one; otherwise it's treated as the
|
||||
# commit message (backward-compatible with `./build.sh "message"`).
|
||||
case "${1:-}" in
|
||||
poller) TARGETS="poller"; MSG="${2:-Build and push images}" ;;
|
||||
web) TARGETS="web"; MSG="${2:-Build and push images}" ;;
|
||||
all|"") TARGETS="poller web"; MSG="${2:-Build and push images}" ;;
|
||||
*) TARGETS="poller web"; MSG="${1}" ;;
|
||||
esac
|
||||
MSG="${2:-Build and push images}"
|
||||
TARGET="${TARGETS// /+}"
|
||||
|
||||
# Stage and commit all changes
|
||||
git add -A
|
||||
|
||||
@@ -165,6 +165,13 @@ async def main() -> None:
|
||||
await asyncio.sleep(LOCK_TTL // 2)
|
||||
logger.info("poller lock acquired (%s)", TOKEN)
|
||||
|
||||
# Start the refresher IMMEDIATELY — before any pre-sweep sleep. The jitter
|
||||
# and restart-storm deferral below can together exceed LOCK_TTL, so without
|
||||
# an active refresher the lock would expire and a second poller could begin
|
||||
# sweeping concurrently (a hardware-storm risk).
|
||||
refresher = asyncio.create_task(lock_refresher())
|
||||
refresher.add_done_callback(_critical_task_done)
|
||||
|
||||
if not smartctl_available():
|
||||
logger.warning("smartctl not found — install smartmontools")
|
||||
if not sg_ses_available():
|
||||
@@ -188,8 +195,6 @@ async def main() -> None:
|
||||
logger.info("last sweep %.0fs ago — deferring first sweep %.0fs", since, wait)
|
||||
await asyncio.sleep(wait)
|
||||
|
||||
refresher = asyncio.create_task(lock_refresher())
|
||||
refresher.add_done_callback(_critical_task_done)
|
||||
leds = asyncio.create_task(led_consumer())
|
||||
try:
|
||||
while True:
|
||||
|
||||
@@ -27,11 +27,23 @@ def _semaphore() -> asyncio.Semaphore:
|
||||
global _sem
|
||||
if _sem is None:
|
||||
n = max(1, int(os.environ.get("POLL_CONCURRENCY", "1")))
|
||||
# Serialized (1) is the safe default. Concurrent hardware I/O can drop
|
||||
# fragile SAS expanders, so >1 is clamped unless explicitly overridden.
|
||||
if n > 1:
|
||||
logger.warning(
|
||||
"POLL_CONCURRENCY=%d (>1) — hardware ops will run concurrently; "
|
||||
"only do this on backplanes known to tolerate it", n,
|
||||
override = os.environ.get("POLL_ALLOW_UNSAFE_CONCURRENCY", "").lower() in (
|
||||
"1", "true", "yes",
|
||||
)
|
||||
if override:
|
||||
logger.warning(
|
||||
"POLL_CONCURRENCY=%d with unsafe override — concurrent "
|
||||
"hardware I/O enabled; backplane must tolerate it", n,
|
||||
)
|
||||
else:
|
||||
logger.warning(
|
||||
"POLL_CONCURRENCY=%d ignored (clamped to 1 for hardware "
|
||||
"safety); set POLL_ALLOW_UNSAFE_CONCURRENCY=1 to override", n,
|
||||
)
|
||||
n = 1
|
||||
_sem = asyncio.Semaphore(n)
|
||||
return _sem
|
||||
|
||||
|
||||
Reference in New Issue
Block a user