fix: guard connection-state callbacks against no running event loop
The previous hotfix's disconnect()-on-comm-error path can now fire during server/brewpi.py's own final best-effort hardware-release cleanup (finally: block's heater.close()/stirrer.activate(False)), which runs after the event loop has already stopped - previously nothing about hardware teardown touched an AttributeChange-tracked field, so this path was dormant. asyncio.create_task() there raises RuntimeError: no running event loop, crashing the shutdown sequence. Adds fire_and_forget() (tasks/task.py) - asyncio.create_task() that silently drops the message instead of raising when there's no loop to schedule on - and uses it for the Connected/FirmwareVersion broadcasts. SudTask.check_connections() also now no-ops entirely without a running loop, since self.sud.stop() cascades into further unguarded create_task() calls (on_state_changed, heater_task.shutdown(), ...) that a stopped loop can't help with anyway. Reproduced and verified fixed directly against HeaterTask/HeaterHendi with no event loop running (mirrors the exact server/brewpi.py shutdown-time call path that crashed on the brewpi Pi). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01YaPLuRPpyjWcwhMvCvpHCL
This commit is contained in:
+1
-1
@@ -1,4 +1,4 @@
|
||||
from tasks.task import ATask, TaskManager
|
||||
from tasks.task import ATask, TaskManager, fire_and_forget
|
||||
from tasks.tempsensor import TempSensorTask
|
||||
from tasks.heater import HeaterTask
|
||||
from tasks.stirrer import StirrerTask
|
||||
|
||||
+3
-3
@@ -1,5 +1,5 @@
|
||||
import asyncio
|
||||
from tasks import ATask
|
||||
from tasks import ATask, fire_and_forget
|
||||
from components import AHeater
|
||||
from ws.message import MsgIo
|
||||
from utils.value import ChangedInteger
|
||||
@@ -34,12 +34,12 @@ class HeaterTask(ATask):
|
||||
self._on_connected_changed = callback
|
||||
|
||||
def on_connected_changed(self, value):
|
||||
asyncio.create_task(self.send({'Connected': value}))
|
||||
fire_and_forget(self.send({'Connected': value}))
|
||||
if self._on_connected_changed:
|
||||
self._on_connected_changed(value)
|
||||
|
||||
def on_firmware_version_changed(self, value):
|
||||
asyncio.create_task(self.send({'FirmwareVersion': value}))
|
||||
fire_and_forget(self.send({'FirmwareVersion': value}))
|
||||
|
||||
def shutdown(self):
|
||||
"""Called when the brew ends (DONE/IDLE). Switches to open-loop at
|
||||
|
||||
+3
-3
@@ -1,5 +1,5 @@
|
||||
import asyncio
|
||||
from tasks import ATask
|
||||
from tasks import ATask, fire_and_forget
|
||||
from ws.message import MsgIo
|
||||
from components import AStirrer
|
||||
from utils.value import ChangedInteger
|
||||
@@ -34,12 +34,12 @@ class StirrerTask(ATask):
|
||||
self._on_connected_changed = callback
|
||||
|
||||
def on_connected_changed(self, value):
|
||||
asyncio.create_task(self.send({'Connected': value}))
|
||||
fire_and_forget(self.send({'Connected': value}))
|
||||
if self._on_connected_changed:
|
||||
self._on_connected_changed(value)
|
||||
|
||||
def on_firmware_version_changed(self, value):
|
||||
asyncio.create_task(self.send({'FirmwareVersion': value}))
|
||||
fire_and_forget(self.send({'FirmwareVersion': value}))
|
||||
|
||||
async def recv(self, data):
|
||||
for pair in data.items():
|
||||
|
||||
+18
-4
@@ -1,6 +1,6 @@
|
||||
import asyncio
|
||||
import bisect
|
||||
from tasks import ATask
|
||||
from tasks import ATask, fire_and_forget
|
||||
from ws.message import MsgIo
|
||||
from utils.value import ChangedFloat
|
||||
from components import APid, AStirrer, AHeater
|
||||
@@ -265,14 +265,28 @@ class SudTask(ATask):
|
||||
outside RAMPING/HOLDING/WAIT_USER/PAUSED), the same path a manual
|
||||
Stop press takes, so all the usual shutdown plumbing (heater
|
||||
shutdown, sud log stop_run - see SudTask.set_on_end()'s wiring in
|
||||
server/brewpi.py) fires exactly as it would for Stop."""
|
||||
server/brewpi.py) fires exactly as it would for Stop.
|
||||
|
||||
No-op if there's no running event loop - happens when a
|
||||
Connectable's connected attribute changes as a side effect of the
|
||||
server's own final best-effort hardware-release cleanup
|
||||
(server/brewpi.py's `finally:` block), which runs after the loop
|
||||
has already stopped. self.sud.stop() below cascades into Sud's own
|
||||
'state' observable (on_state_changed -> heater_task.shutdown()/
|
||||
sud_log_task.stop_run()), which isn't itself guarded against a
|
||||
stopped loop - the whole server is already tearing down at that
|
||||
point regardless, so there's nothing useful to stop/notify."""
|
||||
try:
|
||||
asyncio.get_running_loop()
|
||||
except RuntimeError:
|
||||
return
|
||||
if self.sud.state in (SudState.IDLE, SudState.DONE):
|
||||
return
|
||||
if self.heater.connected and self.stirrer.connected:
|
||||
return
|
||||
self.sud.stop()
|
||||
asyncio.create_task(self.send({'Error': 'Heater/Stirrer disconnected - brew stopped.'}))
|
||||
asyncio.create_task(self.send({'Error': None}))
|
||||
fire_and_forget(self.send({'Error': 'Heater/Stirrer disconnected - brew stopped.'}))
|
||||
fire_and_forget(self.send({'Error': None}))
|
||||
|
||||
def on_state_changed(self, value):
|
||||
asyncio.create_task(self.send({'State': str(value)}))
|
||||
|
||||
@@ -3,6 +3,21 @@ import asyncio
|
||||
from utils.value import AttributeChange
|
||||
|
||||
|
||||
def fire_and_forget(coro):
|
||||
"""asyncio.create_task(), tolerant of there being no running event loop.
|
||||
An actor's observable attribute (e.g. Connectable.connected) can change
|
||||
as a side effect of the server's own final best-effort hardware-release
|
||||
cleanup (server/brewpi.py's `finally:` block - heater.close()/
|
||||
stirrer.activate(False)), which runs after the loop has already
|
||||
stopped. There's nobody left to notify at that point anyway, so this
|
||||
just drops the message instead of crashing the shutdown sequence."""
|
||||
try:
|
||||
return asyncio.create_task(coro)
|
||||
except RuntimeError:
|
||||
coro.close()
|
||||
return None
|
||||
|
||||
|
||||
class ATask(AttributeChange):
|
||||
def __init__(self, interval):
|
||||
AttributeChange.__init__(self)
|
||||
|
||||
Reference in New Issue
Block a user