Async exception handling (#3731)

* remove unused exception

* add logging

* disable pylint broad-except

* add exception handler

* fix lint

* update log output

* change log message in async with exc_info

* Add exc_info to asyncio exception handler

homeassistant/core.py

@@ -136,6 +136,7 @@ class HomeAssistant(object):
         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.pool = pool = create_worker_pool()
         self.bus = EventBus(pool, self.loop)
         self.services = ServiceRegistry(self.bus, self.add_job, self.loop)
@@ -318,6 +319,25 @@ class HomeAssistant(object):
         self.state = CoreState.not_running
         self.loop.stop()
 
+    # pylint: disable=no-self-use
+    def _async_exception_handler(self, loop, context):
+        """Handle all exception inside the core loop."""
+        message = context.get('message')
+        if message:
+            _LOGGER.warning(
+                "Error inside async loop: %s",
+                message
+            )
+
+        # for debug modus
+        exception = context.get('exception')
+        if exception is not None:
+            exc_info = (type(exception), exception, exception.__traceback__)
+            _LOGGER.debug(
+                "Exception inside async loop: ",
+                exc_info=exc_info
+            )
+
 
 class EventOrigin(enum.Enum):
     """Represent the origin of an event."""

homeassistant/util/async.py

@@ -1,6 +1,7 @@
 """Asyncio backports for Python 3.4.3 compatibility."""
 import concurrent.futures
 import threading
+import logging
 from asyncio import coroutines
 from asyncio.futures import Future
 
@@ -13,6 +14,9 @@ except ImportError:
     ensure_future = async
 
 
+_LOGGER = logging.getLogger(__name__)
+
+
 def _set_result_unless_cancelled(fut, result):
     """Helper setting the result only if the future was not cancelled."""
     if fut.cancelled():
@@ -111,10 +115,12 @@ def run_coroutine_threadsafe(coro, loop):
         try:
             # pylint: disable=deprecated-method
             _chain_future(ensure_future(coro, loop=loop), future)
+        # pylint: disable=broad-except
         except Exception as exc:
             if future.set_running_or_notify_cancel():
                 future.set_exception(exc)
-            raise
+            else:
+                _LOGGER.warning("Exception on lost future: ", exc_info=True)
 
     loop.call_soon_threadsafe(callback)
     return future
@@ -158,10 +164,12 @@ def run_callback_threadsafe(loop, callback, *args):
         """Run callback and store result."""
         try:
             future.set_result(callback(*args))
+        # pylint: disable=broad-except
         except Exception as exc:
             if future.set_running_or_notify_cancel():
                 future.set_exception(exc)
-            raise
+            else:
+                _LOGGER.warning("Exception on lost future: ", exc_info=True)
 
     loop.call_soon_threadsafe(run_callback)
     return future