Prevent storage loads from monopolizing the executor pool (#55389)

* Prevent storage loads from monopolizing the executor pool

- At startup there is an increasing demand to load data
  from storage. Similar to #49451 and #43085, we now prevent
  the thread pool from being monopolized by storage loads and
  allow other consumers that are doing network I/O to proceed
  without having to wait for a free executor thread.

* Only create Semaphore instance when one is not already there

homeassistant/helpers/storage.py

@@ -11,7 +11,7 @@ from typing import Any, Callable
 from homeassistant.const import EVENT_HOMEASSISTANT_FINAL_WRITE
 from homeassistant.core import CALLBACK_TYPE, CoreState, Event, HomeAssistant, callback
 from homeassistant.helpers.event import async_call_later
-from homeassistant.loader import bind_hass
+from homeassistant.loader import MAX_LOAD_CONCURRENTLY, bind_hass
 from homeassistant.util import json as json_util
 
 # mypy: allow-untyped-calls, allow-untyped-defs, no-warn-return-any
@@ -20,6 +20,8 @@ from homeassistant.util import json as json_util
 STORAGE_DIR = ".storage"
 _LOGGER = logging.getLogger(__name__)
 
+STORAGE_SEMAPHORE = "storage_semaphore"
+
 
 @bind_hass
 async def async_migrator(
@@ -109,8 +111,12 @@ class Store:
 
     async def _async_load(self):
         """Load the data and ensure the task is removed."""
+        if STORAGE_SEMAPHORE not in self.hass.data:
+            self.hass.data[STORAGE_SEMAPHORE] = asyncio.Semaphore(MAX_LOAD_CONCURRENTLY)
+
         try:
-            return await self._async_load_data()
+            async with self.hass.data[STORAGE_SEMAPHORE]:
+                return await self._async_load_data()
         finally:
             self._load_task = None