Coverage for core / event_loop.py: 100.0%

1""" 

2Singleton event loop management. 

3 

4Replaces repeated `asyncio.new_event_loop()` calls (7+ across codebase) 

5with a single reusable event loop. 

6 

7Before: Each async call creates a new event loop, causing resource leaks. 

8After: One event loop per thread, properly managed. 

9""" 

10 

11import asyncio 

12import threading 

13import logging 

14 

15logger = logging.getLogger('hevolve_core') 

16 

17_thread_loops = threading.local() 

18 

19 

20def get_or_create_event_loop() -> asyncio.AbstractEventLoop: 

21 """ 

22 Get or create an event loop for the current thread. 

23 Reuses existing loops instead of creating new ones each time. 

24 """ 

25 # Check for thread-local loop first 

26 loop = getattr(_thread_loops, 'loop', None) 

27 if loop is not None and not loop.is_closed(): 

28 return loop 

29 

30 # Try to get existing loop 

31 try: 

32 loop = asyncio.get_event_loop() 

33 if not loop.is_closed(): 

34 _thread_loops.loop = loop 

35 return loop 

36 except RuntimeError: 

37 pass 

38 

39 # Create new loop for this thread 

40 loop = asyncio.new_event_loop() 

41 asyncio.set_event_loop(loop) 

42 _thread_loops.loop = loop 

43 logger.debug(f"Created new event loop for thread {threading.current_thread().name}") 

44 return loop 

45 

46 

47def run_async(coro): 

48 """ 

49 Run an async coroutine from synchronous code. 

50 Uses the singleton event loop for the current thread. 

51 """ 

52 loop = get_or_create_event_loop() 

53 return loop.run_until_complete(coro)