Events: Periodically witch out of the event parsing loop

inbox/events/google.py

@@ -27,6 +27,7 @@
 from inbox.models import Account, Calendar
 from inbox.models.backends.oauth import token_manager
 from inbox.models.event import EVENT_STATUSES, Event
+from inbox.util.concurrency import iterate_and_periodically_switch_to_gevent
 
 CALENDARS_URL = "https://www.googleapis.com/calendar/v3/users/me/calendarList"
 STATUS_MAP = {
@@ -92,7 +93,7 @@ def sync_events(
         updates = []
         items = self._get_raw_events(calendar_uid, sync_from_time)
         read_only_calendar = self.calendars_table.get(calendar_uid, True)
-        for item in items:
+        for item in iterate_and_periodically_switch_to_gevent(items):
             try:
                 parsed = parse_event_response(item, read_only_calendar)
                 updates.append(parsed)

inbox/events/microsoft/events_provider.py

@@ -27,6 +27,7 @@
 from inbox.models.calendar import Calendar
 from inbox.models.event import Event, RecurringEvent
 from inbox.models.session import session_scope
+from inbox.util.concurrency import iterate_and_periodically_switch_to_gevent
 
 URL_PREFIX = config.get("API_URL", "")
 
@@ -135,7 +136,7 @@ def sync_events(
             ),
         )
         read_only = self.calendars_table.get(calendar_uid, True)
-        for raw_event in raw_events:
+        for raw_event in iterate_and_periodically_switch_to_gevent(raw_events):
             if not validate_event(raw_event):
                 self.log.warning("Invalid event", raw_event=raw_event)
                 continue

inbox/util/concurrency.py

@@ -1,7 +1,10 @@
+import datetime
 import functools
 import random
 import ssl
 import sys
+import time
+from typing import Iterable, TypeVar
 
 import _mysql_exceptions
 import gevent
@@ -152,3 +155,30 @@ def callback(e):
         fail_classes=fail_classes,
         backoff_delay=backoff_delay,
     )()
+
+
+IterableItemT = TypeVar("IterableItemT")
+
+
+DEFAULT_SWITCH_PERIOD = datetime.timedelta(seconds=1)
+
+
+def iterate_and_periodically_switch_to_gevent(
+    iterable: Iterable[IterableItemT],
+    *,
+    switch_period: datetime.timedelta = DEFAULT_SWITCH_PERIOD
+) -> Iterable[IterableItemT]:
+    """
+    Given an iterable, yield each item, and periodically switch to the gevent
+    event loop to allow other greenlets to run.
+
+    Use this with CPU-bound loops to avoid blocking the event loop for too long.
+    Otherwise the greenlet might get killed by KillerGreenletTracer.
+    """
+    last_sleep_time = time.monotonic()
+    for item in iterable:
+        if time.monotonic() - last_sleep_time >= switch_period.total_seconds():
+            gevent.sleep(0)
+            last_sleep_time = time.monotonic()
+
+        yield item