Mercurial > code > home > repos > light9

comparison lib/background_loop.py @ 2215:d8853f173568

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
fork background_loop
author drewp@bigasterisk.com
date Tue, 23 May 2023 14:41:04 -0700
parents
children acf1b68d031a
comparison
equal deleted inserted replaced
2214:d665acb12046 2215:d8853f173568
1 # dev copy
2 import asyncio
3 import logging
4 import time
5 import traceback
6 from typing import Any, Awaitable, Callable, Union
7
8 from prometheus_client import Gauge, Summary
9
10 log = logging.getLogger()
11
12
13 # throw this away (when net_routes is rewritten)
14 def loop_forever_sync(func, sleep_period, up_metric):
15 first_run = True
16 while True:
17 try:
18 func(first_run)
19 up_metric.set(1)
20 first_run = False
21 except Exception as ex:
22 log.error(ex)
23 traceback.print_exc()
24 up_metric.set(0)
25 time.sleep(sleep_period)
26
27
28 _created = []
29
30 # todo: some tricky typing
31 F_RET = Any
32 F_KWARGS = Any
33
34 UserAsyncFunc = Callable[..., # always called with at least kwarg first_run=bool
35 Awaitable[F_RET]]
36 UserSyncFunc = Callable[..., # see above
37 F_RET]
38 UserFunc = Union[UserAsyncFunc, UserSyncFunc]
39
40
41 class Loop:
42
43 def __init__(
44 self,
45 func: UserFunc,
46 sleep_period: float,
47 metric_prefix: str,
48 ):
49 self.func = func
50 self.sleep_period = sleep_period
51
52 self.up_metric = Gauge(f'{metric_prefix}_up', 'not erroring')
53 self.call_metric = Summary(f'{metric_prefix}_calls', 'calls')
54 self.lastSuccessRun = 0
55 self.everSucceeded = False
56
57 async def call_func(self, first_run: bool, call_kwargs={}) -> F_RET:
58 with self.call_metric.time():
59 if asyncio.iscoroutinefunction(self.func):
60 ret = await self.func(first_run=first_run, **call_kwargs)
61 else:
62 ret = self.func(first_run=first_run, **call_kwargs)
63 return ret
64
65 async def _run(self):
66 self.first_run = True
67 while True:
68 await self._runOne()
69 await asyncio.sleep(self.sleep_period)
70
71 async def runNow(self, **kwargs: F_KWARGS) -> F_RET:
72 """unlike background runs, you get to pass more args and get your func's
73 return value.
74 """
75 return await self._runOne(call_kwargs=kwargs)
76
77 async def _runOne(self, call_kwargs={}):
78 try:
79 result = await self.call_func(self.first_run, call_kwargs)
80 self.lastSuccessRun = time.time()
81 self.up_metric.set(1)
82 self.everSucceeded = True
83 self.first_run = False
84 except Exception as ex:
85 log.error(ex)
86 traceback.print_exc()
87 self.up_metric.set(0)
88 result = None
89 # todo: something that reveals error ratio
90 return result
91
92
93 def loop_forever(
94 func: UserFunc,
95 sleep_period: float,
96 metric_prefix='background_loop',
97 up_metric=None,
98 call_metric=None,
99 ):
100 """
101 sleep_period is the sleep time after however long func takes to run
102 """
103 if up_metric is not None or call_metric is not None:
104 raise NotImplementedError('remove old-style metrics')
105
106 loop = Loop(func, sleep_period, metric_prefix)
107 _created.append(asyncio.create_task(loop._run()))
108 return loop