moved trigger task cancellation to reaper task

Author: craigbarratt

custom_components/pyscript/function.py

@@ -7,7 +7,7 @@
 
 from .const import LOGGER_PATH
 
-_LOGGER = logging.getLogger(LOGGER_PATH + ".handler")
+_LOGGER = logging.getLogger(LOGGER_PATH + ".function")
 
 
 class Function:
@@ -73,18 +73,22 @@ def init(cls, hass):
         # start a task which is a reaper for canceled tasks, since some # functions
         # like TrigInfo.stop() can't be async (it's called from a __del__ method)
         #
-        async def task_await_reaper(reaper_q):
-            try:
-                while True:
-                    task = await reaper_q.get()
-                    await task
-            except asyncio.CancelledError:
-                raise
-            except Exception:
-                pass
+        async def task_cancel_reaper(reaper_q):
+            while True:
+                try:
+                    try:
+                        task = await reaper_q.get()
+                        task.cancel()
+                        await task
+                    except asyncio.CancelledError:
+                        pass
+                except asyncio.CancelledError:
+                    raise
+                except Exception:
+                    _LOGGER.error("task_cancel_reaper: got exception %s", traceback.format_exc(-1))
 
         cls.task_reaper_q = asyncio.Queue(0)
-        cls.task_await_repeaer = Function.create_task(task_await_reaper(cls.task_reaper_q))
+        cls.task_cancel_repeaer = Function.create_task(task_cancel_reaper(cls.task_reaper_q))
 
     @classmethod
     async def async_sleep(cls, duration):
@@ -101,20 +105,13 @@ async def task_unique(cls, name, kill_me=False):
         """Implement task.unique()."""
         if name in cls.unique_name2task:
             if kill_me:
-                task = asyncio.current_task()
-
-                # it seems we need to use another task to cancel ourselves
-                # I'm sure there is a better way to cancel ourselves...
-                async def cancel_self():
-                    try:
-                        task.cancel()
-                        await task
-                    except asyncio.CancelledError:
-                        pass
-
-                asyncio.create_task(cancel_self())
-                # ugh - wait to be canceled
-                await asyncio.sleep(10000)
+                #
+                # it seems we can't cancel ourselves, so we
+                # tell the repeaer task to cancel us
+                #
+                Function.task_cancel(asyncio.current_task())
+                # wait to be canceled
+                await asyncio.sleep(100000)
             else:
                 task = cls.unique_name2task[name]
                 if task in cls.our_tasks:
@@ -216,14 +213,14 @@ async def run_coro(cls, coro):
         #
         # Add a placeholder for the new task so we know it's one we started
         #
-        task = asyncio.current_task()
-        cls.our_tasks.add(task)
         try:
+            task = asyncio.current_task()
+            cls.our_tasks.add(task)
             await coro
         except asyncio.CancelledError:
             raise
         except Exception:
-            _LOGGER.error("run_coro: %s", traceback.format_exc(-1))
+            _LOGGER.error("run_coro: got exception %s", traceback.format_exc(-1))
         finally:
             if task in cls.unique_task2name:
                 del cls.unique_name2task[cls.unique_task2name[task]]
@@ -236,6 +233,6 @@ def create_task(cls, coro):
         return cls.hass.loop.create_task(cls.run_coro(coro))
 
     @classmethod
-    def task_await_send(cls, task):
-        """Send a task that has been canceled to the reaper task so it can be awaited."""
+    def task_cancel(cls, task):
+        """Send a task to be canceled by the reaper."""
         cls.task_reaper_q.put_nowait(task)

custom_components/pyscript/trigger.py

@@ -221,7 +221,6 @@ async def wait_until(
             Event.notify_del(event_trigger[0], notify_q)
         if exc:
             raise exc
-        _LOGGER.debug("trigger %s wait_until returning %s", ast_ctx.name, ret)
         return ret
 
     @classmethod
@@ -461,8 +460,6 @@ def __init__(
         self.setup_ok = False
         self.run_on_startup = False
 
-        _LOGGER.debug("trigger %s event_trigger = %s", self.name, self.event_trigger)
-
         if self.state_active is not None:
             self.active_expr = AstEval(
                 f"{self.name} @state_active()", self.global_ctx, logger_name=self.name
@@ -519,19 +516,13 @@ def stop(self):
             if self.event_trigger is not None:
                 Event.notify_del(self.event_trigger[0], self.notify_q)
             if self.task:
-                try:
-                    self.task.cancel()
-                    Function.task_await_send(self.task)
-                except asyncio.CancelledError:
-                    pass
-                self.task = None
-                _LOGGER.debug("trigger %s is stopped", self.name)
+                Function.task_cancel(self.task)
 
     def start(self):
         """Start this trigger task."""
         if not self.task and self.setup_ok:
             self.task = Function.create_task(self.trigger_watch())
-            _LOGGER.debug("trigger %s is active", self.name)
+            _LOGGER.debug("trigger %s is active %s", self.name, self.task)
 
     async def trigger_watch(self):
         """Task that runs for each trigger, waiting for the next trigger and calling the function."""
@@ -655,7 +646,7 @@ async def do_func_call(func, ast_ctx, task_unique, kwargs=None):
                     func_args,
                 )
                 Function.create_task(
-                    do_func_call(self.action, self.action_ast_ctx, self.task_unique, kwargs=func_args,)
+                    do_func_call(self.action, self.action_ast_ctx, self.task_unique, kwargs=func_args)
                 )
 
         except asyncio.CancelledError:

docs/reference.rst

@@ -247,21 +247,21 @@ In ``@time_trigger``, each string specification ``time_spec`` can take one of fo
   according to Linux-style crontab. Each of the five entries are separated by spaces and correspond
   to minutes, hours, day-of-month, month, day-of-week (0 = sunday):
 
-============ ==============
-field        allowed values
-============ ==============
-minute       0-59
-hour         0-23
-day of month 1-31
-month        1-12
-day of week  0-6 (0 is Sun)
-============ ==============
-
-Each field can be a ``*`` (which means “all”), a single number, a range or comma-separated list of
-numbers or ranges (no spaces). Ranges are inclusive. For example, if you specify hours as
-``6,10-13`` that means hours of 6,10,11,12,13. The trigger happens on the next minute, hour, day
-that matches the specification. See any Linux documentation for examples and more details (note:
-names for days of week and months are not supported; only their integer values are).
+  ============ ==============
+  field        allowed values
+  ============ ==============
+  minute       0-59
+  hour         0-23
+  day of month 1-31
+  month        1-12
+  day of week  0-6 (0 is Sun)
+  ============ ==============
+
+  Each field can be a ``*`` (which means “all”), a single number, a range or comma-separated list of
+  numbers or ranges (no spaces). Ranges are inclusive. For example, if you specify hours as
+  ``6,10-13`` that means hours of 6,10,11,12,13. The trigger happens on the next minute, hour, day
+  that matches the specification. See any Linux documentation for examples and more details (note:
+  names for days of week and months are not supported; only their integer values are).
 
 When the ``@time_trigger`` occurs and the function is called, the keyword argument ``trigger_type``
 is set to ``"time"``, and ``trigger_time`` is the exact ``datetime`` of the time specification that

tests/test_function.py

@@ -513,7 +513,7 @@ def func_trig(var_name=None, value=None):
 
     return func_trig
 
-f = [factory(50), factory(51), factory(52), factory(53)]
+f = [factory(50), factory(51), factory(52), factory(53), factory(54)]
 """,
     )
     seq_num = 0
@@ -524,49 +524,30 @@ def func_trig(var_name=None, value=None):
     assert literal_eval(await wait_until_done(notify_q)) == seq_num
 
     #
-    # trigger them one at a time
+    # trigger them one at a time to make sure each is working
     #
-    for i in range(3):
+    for i in range(5):
         seq_num += 1
         hass.states.async_set("pyscript.var1", 50 + i)
         assert literal_eval(await wait_until_done(notify_q)) == seq_num
 
-    #
-    # trigger all three together; we don't know the order, so just check
-    # we got all 3
-    #
-    hass.states.async_set("pyscript.var1", 100)
-    seqs = set()
-    expect = set()
-    for i in range(4):
-        seqs.add(literal_eval(await wait_until_done(notify_q)))
-        expect.add(seq_num + 50 + i)
-    assert seqs == expect
-
-    #
-    # now trigger all again, but just the first deletes the last
-    # trigger function and replies
-    #
-    seq_num += 1
-    hass.states.async_set("pyscript.var1", 101)
-    assert literal_eval(await wait_until_done(notify_q)) == seq_num
-
-    #
-    # now trigger all again, and confirm we only get two
-    #
-    hass.states.async_set("pyscript.var1", 100)
-    seqs = set()
-    expect = set()
-    for i in range(3):
-        seqs.add(literal_eval(await wait_until_done(notify_q)))
-        expect.add(seq_num + 50 + i)
-    assert seqs == expect
-
-    #
-    # now trigger all again, but just the first deletes the last
-    # trigger function and replies, just to make sure the last
-    # one didn't trigger
-    #
-    seq_num += 1
-    hass.states.async_set("pyscript.var1", 101)
-    assert literal_eval(await wait_until_done(notify_q)) == seq_num
+    for num_func in range(5, 1, -1):
+        #
+        # trigger all together; we don't know the order, so just check
+        # we got all of the remaining ones
+        #
+        hass.states.async_set("pyscript.var1", 100)
+        seqs = set()
+        expect = set()
+        for i in range(num_func):
+            seqs.add(literal_eval(await wait_until_done(notify_q)))
+            expect.add(seq_num + 50 + i)
+        assert seqs == expect
+
+        #
+        # now trigger all again, but just the first deletes the last
+        # trigger function and replies with the next seq number
+        #
+        seq_num += 1
+        hass.states.async_set("pyscript.var1", 101)
+        assert literal_eval(await wait_until_done(notify_q)) == seq_num