Fixes issue with too many tasks piled in event loop (due to inactivity) and issue with scheduler not restarting tasks on hot-reload
@@ -5,6 +5,7 @@ import pytz
import tzlocal
import traceback
import inspect
+import logging
from typing import Any, Dict, List, Union, Optional, Callable, Tuple # noqa
try:
from typing import Awaitable
@@ -190,7 +191,7 @@ class Scheduler(Invoker):
return timezone
- async def start_schedule_loop(cls: Any, obj: Any, context: Dict, handler: Callable, interval: Optional[Union[str, int]]=None, timestamp: Optional[str]=None, timezone: Optional[str]=None) -> None:
+ async def start_schedule_loop(cls: Any, obj: Any, context: Dict, handler: Callable, func: Callable, interval: Optional[Union[str, int]]=None, timestamp: Optional[str]=None, timezone: Optional[str]=None) -> None:
timezone = cls.get_timezone(timezone)
if not cls.close_waiter:
@@ -204,6 +205,7 @@ class Scheduler(Invoker):
prev_call_at = None
tasks = [] # type: List
current_time = time.time()
+ too_many_tasks = False
while not cls.close_waiter.done():
try:
last_time = current_time
@@ -224,6 +226,13 @@ class Scheduler(Invoker):
prev_call_at = next_call_at
next_call_at = None
tasks = [task for task in tasks if not task.done()]
+ if len(tasks) >= 20:
+ if not too_many_tasks:
+ too_many_tasks = True
+ logging.getLogger('transport.schedule').warning('Too many scheduled tasks (20) for function "{}"'.format(func.__name__))
+ await asyncio.sleep(1)
+ continue
+ too_many_tasks = False
tasks.append(asyncio.ensure_future(asyncio.shield(handler())))
except Exception as e:
if not context.get('log_level') or context.get('log_level') in ['DEBUG']:
@@ -273,11 +282,13 @@ class Scheduler(Invoker):
context['_schedule_loop_started'] = True
async def _schedule() -> None:
+ cls.close_waiter = asyncio.Future()
+
for interval, timestamp, timezone, func, handler in context.get('_schedule_scheduled_functions', []):
cls.next_call_at(time.time(), interval, timestamp, cls.get_timezone(timezone)) # test provided interval/timestamp on init
for interval, timestamp, timezone, func, handler in context.get('_schedule_scheduled_functions', []):
- await cls.start_schedule_loop(cls, obj, context, handler, interval, timestamp, timezone)
+ await cls.start_schedule_loop(cls, obj, context, handler, func, interval, timestamp, timezone)
return _schedule
