Core: cleanup timer (#5825)

* Minor core cleanup

* Cleanup timer

* Lint

* timeout with correct loop

* Improve timer thanks to pvizeli

* Update core.py

* More tests

homeassistant/core.py

@@ -13,6 +13,7 @@ import os
 import re
 import sys
 import threading
+from time import monotonic
 
 from types import MappingProxyType
 from typing import Optional, Any, Callable, List  # NOQA
@@ -43,9 +44,6 @@ except ImportError:
 
 DOMAIN = 'homeassistant'
 
-# How often time_changed event should fire
-TIMER_INTERVAL = 1  # seconds
-
 # How long we wait for the result of a service call
 SERVICE_CALL_LIMIT = 10  # seconds
 
@@ -83,6 +81,22 @@ def is_callback(func: Callable[..., Any]) -> bool:
     return '_hass_callback' in func.__dict__
 
 
+@callback
+def async_loop_exception_handler(loop, context):
+    """Handle all exception inside the core loop."""
+    kwargs = {}
+    exception = context.get('exception')
+    if exception:
+        # Do not report on shutting down exceptions.
+        if isinstance(exception, ShuttingDown):
+            return
+
+        kwargs['exc_info'] = (type(exception), exception,
+                              exception.__traceback__)
+
+    _LOGGER.error("Error doing job: %s", context['message'], **kwargs)
+
+
 class CoreState(enum.Enum):
     """Represent the current state of Home Assistant."""
 
@@ -108,7 +122,7 @@ class HomeAssistant(object):
 
         self.executor = ThreadPoolExecutor(max_workers=EXECUTOR_POOL_SIZE)
         self.loop.set_default_executor(self.executor)
-        self.loop.set_exception_handler(self._async_exception_handler)
+        self.loop.set_exception_handler(async_loop_exception_handler)
         self._pending_tasks = []
         self.bus = EventBus(self)
         self.services = ServiceRegistry(self)
@@ -286,22 +300,6 @@ class HomeAssistant(object):
         self.exit_code = exit_code
         self.loop.stop()
 
-    # pylint: disable=no-self-use
-    @callback
-    def _async_exception_handler(self, loop, context):
-        """Handle all exception inside the core loop."""
-        kwargs = {}
-        exception = context.get('exception')
-        if exception:
-            # Do not report on shutting down exceptions.
-            if isinstance(exception, ShuttingDown):
-                return
-
-            kwargs['exc_info'] = (type(exception), exception,
-                                  exception.__traceback__)
-
-        _LOGGER.error("Error doing job: %s", context['message'], **kwargs)
-
 
 class EventOrigin(enum.Enum):
     """Represent the origin of an event."""
@@ -494,7 +492,6 @@ class EventBus(object):
             # This will make sure the second time it does nothing.
             setattr(onetime_listener, 'run', True)
             self._async_remove_listener(event_type, onetime_listener)
-
             self._hass.async_run_job(listener, event)
 
         return self.async_listen(event_type, onetime_listener)
@@ -542,7 +539,6 @@ class State(object):
         self.state = str(state)
         self.attributes = MappingProxyType(attributes or {})
         self.last_updated = last_updated or dt_util.utcnow()
-
         self.last_changed = last_changed or self.last_updated
 
     @property
@@ -673,7 +669,6 @@ class StateMachine(object):
         Async friendly.
         """
         state_obj = self.get(entity_id)
-
         return state_obj and state_obj.state == state
 
     def is_state_attr(self, entity_id, name, value):
@@ -682,7 +677,6 @@ class StateMachine(object):
         Async friendly.
         """
         state_obj = self.get(entity_id)
-
         return state_obj and state_obj.attributes.get(name, None) == value
 
     def remove(self, entity_id):
@@ -702,20 +696,16 @@ class StateMachine(object):
         This method must be run in the event loop.
         """
         entity_id = entity_id.lower()
-
         old_state = self._states.pop(entity_id, None)
 
         if old_state is None:
             return False
 
-        event_data = {
+        self._bus.async_fire(EVENT_STATE_CHANGED, {
             'entity_id': entity_id,
             'old_state': old_state,
             'new_state': None,
-        }
+        })
-
-        self._bus.async_fire(EVENT_STATE_CHANGED, event_data)
-
         return True
 
     def set(self, entity_id, new_state, attributes=None, force_update=False):
@@ -746,9 +736,7 @@ class StateMachine(object):
         entity_id = entity_id.lower()
         new_state = str(new_state)
         attributes = attributes or {}
-
         old_state = self._states.get(entity_id)
-
         is_existing = old_state is not None
         same_state = (is_existing and old_state.state == new_state and
                       not force_update)
@@ -757,19 +745,14 @@ class StateMachine(object):
         if same_state and same_attr:
             return
 
-        # If state did not exist or is different, set it
         last_changed = old_state.last_changed if same_state else None
-
         state = State(entity_id, new_state, attributes, last_changed)
         self._states[entity_id] = state
-
+        self._bus.async_fire(EVENT_STATE_CHANGED, {
-        event_data = {
             'entity_id': entity_id,
             'old_state': old_state,
             'new_state': state,
-        }
+        })
-
-        self._bus.async_fire(EVENT_STATE_CHANGED, event_data)
 
 
 class Service(object):
@@ -823,9 +806,17 @@ class ServiceRegistry(object):
         """Initialize a service registry."""
         self._services = {}
         self._hass = hass
-        self._cur_id = 0
         self._async_unsub_call_event = None
 
+        def _gen_unique_id():
+            cur_id = 1
+            while True:
+                yield '{}-{}'.format(id(self), cur_id)
+                cur_id += 1
+
+        gen = _gen_unique_id()
+        self._generate_unique_id = lambda: next(gen)
+
     @property
     def services(self):
         """Dict with per domain a list of available services."""
@@ -1025,11 +1016,6 @@ class ServiceRegistry(object):
 
             self._hass.async_add_job(execute_service)
 
-    def _generate_unique_id(self):
-        """Generate a unique service call id."""
-        self._cur_id += 1
-        return '{}-{}'.format(id(self), self._cur_id)
-
 
 class Config(object):
     """Configuration settings for Home Assistant."""
@@ -1092,66 +1078,38 @@ class Config(object):
         }
 
 
-def _async_create_timer(hass, interval=TIMER_INTERVAL):
+def _async_create_timer(hass):
     """Create a timer that will start on HOMEASSISTANT_START."""
-    stop_event = asyncio.Event(loop=hass.loop)
+    handle = None
+
+    @callback
+    def fire_time_event(nxt):
+        """Fire next time event."""
+        nonlocal handle
+
+        hass.bus.async_fire(EVENT_TIME_CHANGED,
+                            {ATTR_NOW: dt_util.utcnow()})
+        nxt += 1
+        slp_seconds = nxt - monotonic()
+
+        if slp_seconds < 0:
+            _LOGGER.error('Timer got out of sync. Resetting')
+            nxt = monotonic() + 1
+            slp_seconds = 1
+
+        handle = hass.loop.call_later(slp_seconds, fire_time_event, nxt)
+
+    @callback
+    def start_timer(event):
+        """Create an async timer."""
+        _LOGGER.info("Timer:starting")
+        hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, stop_timer)
+        fire_time_event(monotonic())
 
-    # Setting the Event inside the loop by marking it as a coroutine
     @callback
     def stop_timer(event):
         """Stop the timer."""
-        stop_event.set()
+        if handle is not None:
-
+            handle.cancel()
-    hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, stop_timer)
-
-    @asyncio.coroutine
-    def timer(interval, stop_event):
-        """Create an async timer."""
-        _LOGGER.info("Timer:starting")
-
-        last_fired_on_second = -1
-
-        calc_now = dt_util.utcnow
-
-        while not stop_event.is_set():
-            now = calc_now()
-
-            # First check checks if we are not on a second matching the
-            # timer interval. Second check checks if we did not already fire
-            # this interval.
-            if now.second % interval or \
-               now.second == last_fired_on_second:
-
-                # Sleep till it is the next time that we have to fire an event.
-                # Aim for halfway through the second that fits TIMER_INTERVAL.
-                # If TIMER_INTERVAL is 10 fire at .5, 10.5, 20.5, etc seconds.
-                # This will yield the best results because time.sleep() is not
-                # 100% accurate because of non-realtime OS's
-                slp_seconds = interval - now.second % interval + \
-                    .5 - now.microsecond/1000000.0
-
-                yield from asyncio.sleep(slp_seconds, loop=hass.loop)
-
-                now = calc_now()
-
-            last_fired_on_second = now.second
-
-            # Event might have been set while sleeping
-            if not stop_event.is_set():
-                try:
-                    # Schedule the bus event
-                    hass.loop.call_soon(
-                        hass.bus.async_fire,
-                        EVENT_TIME_CHANGED,
-                        {ATTR_NOW: now}
-                    )
-                except ShuttingDown:
-                    # HA raises error if firing event after it has shut down
-                    break
-
-    @asyncio.coroutine
-    def start_timer(event):
-        """Start our async timer."""
-        hass.loop.create_task(timer(interval, stop_event))
 
     hass.bus.async_listen_once(EVENT_HOMEASSISTANT_START, start_timer)

homeassistant/remote.py

@@ -133,7 +133,7 @@ class HomeAssistant(ha.HomeAssistant):
         self.loop = loop or asyncio.get_event_loop()
         self.executor = ThreadPoolExecutor(max_workers=5)
         self.loop.set_default_executor(self.executor)
-        self.loop.set_exception_handler(self._async_exception_handler)
+        self.loop.set_exception_handler(ha.async_loop_exception_handler)
         self._pending_tasks = []
         self._pending_sheduler = None
 

tests/test_core.py

@@ -2,7 +2,7 @@
 # pylint: disable=protected-access
 import asyncio
 import unittest
-from unittest.mock import patch, MagicMock
+from unittest.mock import patch, MagicMock, sentinel
 from datetime import datetime, timedelta
 
 import pytz
@@ -14,7 +14,9 @@ from homeassistant.util.async import run_coroutine_threadsafe
 import homeassistant.util.dt as dt_util
 from homeassistant.util.unit_system import (METRIC_SYSTEM)
 from homeassistant.const import (
-    __version__, EVENT_STATE_CHANGED, ATTR_FRIENDLY_NAME, CONF_UNIT_SYSTEM)
+    __version__, EVENT_STATE_CHANGED, ATTR_FRIENDLY_NAME, CONF_UNIT_SYSTEM,
+    ATTR_NOW, EVENT_TIME_CHANGED, EVENT_HOMEASSISTANT_STOP,
+    EVENT_HOMEASSISTANT_START)
 
 from tests.common import get_test_home_assistant
 
@@ -736,37 +738,78 @@ class TestConfig(unittest.TestCase):
         self.assertEqual(expected, self.config.as_dict())
 
 
-class TestAsyncCreateTimer(object):
+@patch('homeassistant.core.monotonic')
+def test_create_timer(mock_monotonic, loop):
     """Test create timer."""
-
-    @patch('homeassistant.core.asyncio.Event')
-    @patch('homeassistant.core.dt_util.utcnow')
-    def test_create_timer(self, mock_utcnow, mock_event, event_loop):
-        """Test create timer fires correctly."""
     hass = MagicMock()
-        now = mock_utcnow()
+    funcs = []
-        event = mock_event()
+    orig_callback = ha.callback
-        now.second = 1
-        mock_utcnow.reset_mock()
 
+    def mock_callback(func):
+        funcs.append(func)
+        return orig_callback(func)
+
+    with patch.object(ha, 'callback', mock_callback):
         ha._async_create_timer(hass)
+
+        assert len(funcs) == 3
+        fire_time_event, start_timer, stop_timer = funcs
+
+    assert len(hass.bus.async_listen_once.mock_calls) == 1
+    event_type, callback = hass.bus.async_listen_once.mock_calls[0][1]
+    assert event_type == EVENT_HOMEASSISTANT_START
+    assert callback is start_timer
+
+    mock_monotonic.side_effect = 10.2, 10.3
+
+    with patch('homeassistant.core.dt_util.utcnow',
+               return_value=sentinel.mock_date):
+        start_timer(None)
+
     assert len(hass.bus.async_listen_once.mock_calls) == 2
-        start_timer = hass.bus.async_listen_once.mock_calls[1][1][1]
+    assert len(hass.bus.async_fire.mock_calls) == 1
+    assert len(hass.loop.call_later.mock_calls) == 1
 
-        event_loop.run_until_complete(start_timer(None))
+    event_type, callback = hass.bus.async_listen_once.mock_calls[1][1]
-        assert hass.loop.create_task.called
+    assert event_type == EVENT_HOMEASSISTANT_STOP
+    assert callback is stop_timer
 
-        timer = hass.loop.create_task.mock_calls[0][1][0]
+    slp_seconds, callback, nxt = hass.loop.call_later.mock_calls[0][1]
-        event.is_set.side_effect = False, False, True
+    assert abs(slp_seconds - 0.9) < 0.001
-        event_loop.run_until_complete(timer)
+    assert callback is fire_time_event
-        assert len(mock_utcnow.mock_calls) == 1
+    assert abs(nxt - 11.2) < 0.001
 
-        assert hass.loop.call_soon.called
+    event_type, event_data = hass.bus.async_fire.mock_calls[0][1]
-        event_type, event_data = hass.loop.call_soon.mock_calls[0][1][1:]
+    assert event_type == EVENT_TIME_CHANGED
+    assert event_data[ATTR_NOW] is sentinel.mock_date
 
-        assert ha.EVENT_TIME_CHANGED == event_type
-        assert {ha.ATTR_NOW: now} == event_data
 
-        stop_timer = hass.bus.async_listen_once.mock_calls[0][1][1]
+@patch('homeassistant.core.monotonic')
-        stop_timer(None)
+def test_timer_out_of_sync(mock_monotonic, loop):
-        assert event.set.called
+    """Test create timer."""
+    hass = MagicMock()
+    funcs = []
+    orig_callback = ha.callback
+
+    def mock_callback(func):
+        funcs.append(func)
+        return orig_callback(func)
+
+    with patch.object(ha, 'callback', mock_callback):
+        ha._async_create_timer(hass)
+
+        assert len(funcs) == 3
+        fire_time_event, start_timer, stop_timer = funcs
+
+    mock_monotonic.side_effect = 10.2, 11.3, 11.3
+
+    with patch('homeassistant.core.dt_util.utcnow',
+               return_value=sentinel.mock_date):
+        start_timer(None)
+
+    assert len(hass.loop.call_later.mock_calls) == 1
+
+    slp_seconds, callback, nxt = hass.loop.call_later.mock_calls[0][1]
+    assert slp_seconds == 1
+    assert callback is fire_time_event
+    assert abs(nxt - 12.3) < 0.001