Set executor to 15 and help to reduce flooting async core with updates (#4252)

* Set executor to 15 and help to reduce flooting async core with udpates

* fix typing

* if it a executor, wait

* address comments from paulus

* add space for style :)

* fix spell

* Update entity_component.py

* Update entity_component.py

homeassistant/core.py

@@ -57,7 +57,7 @@ SERVICE_CALL_LIMIT = 10  # seconds
 ENTITY_ID_PATTERN = re.compile(r"^(\w+)\.(\w+)$")
 
 # Size of a executor pool
-EXECUTOR_POOL_SIZE = 10
+EXECUTOR_POOL_SIZE = 15
 
 _LOGGER = logging.getLogger(__name__)
 

homeassistant/helpers/entity_component.py

@@ -289,6 +289,7 @@ class EntityPlatform(object):
         self.entity_namespace = entity_namespace
         self.platform_entities = []
         self._async_unsub_polling = None
+        self._process_updates = False
 
     def add_entities(self, new_entities, update_before_add=False):
         """Add entities for a single platform."""
@@ -348,14 +349,37 @@ class EntityPlatform(object):
             self._async_unsub_polling()
             self._async_unsub_polling = None
 
-    @callback
+    @asyncio.coroutine
     def _update_entity_states(self, now):
         """Update the states of all the polling entities.
 
+        To protect from flooding the executor, we will update async entities
+        in parallel and other entities sequential.
+
         This method must be run in the event loop.
         """
-        for entity in self.platform_entities:
-            if entity.should_poll:
-                self.component.hass.async_add_job(
-                    entity.async_update_ha_state(True)
-                )
+        if self._process_updates:
+            return
+        self._process_updates = True
+
+        try:
+            tasks = []
+            to_update = []
+
+            for entity in self.platform_entities:
+                if not entity.should_poll:
+                    continue
+
+                update_coro = entity.async_update_ha_state(True)
+                if hasattr(entity, 'async_update'):
+                    tasks.append(update_coro)
+                else:
+                    to_update.append(update_coro)
+
+            for update_coro in to_update:
+                yield from update_coro
+
+            if tasks:
+                yield from asyncio.wait(tasks, loop=self.component.hass.loop)
+        finally:
+            self._process_updates = False