mirror of
https://github.com/outbackdingo/patroni.git
synced 2026-09-02 01:29:36 +00:00
Alexander KukushkinandGitHub
038b5aed72
Previously replicas were always watching for leader key (even if the postgres was not in the running there). It was not a big issue, but it was not possible to interrupt such watch in cases if the postgres started up or stopped successfully. Also it was delaying update_member call and we had kind of stale information in DCS up to `loop_wait` seconds. This commit changes such behavior. If the async_executor is busy by starting/stopping or restarting postgres we will not watch for leader key but waiting for event from async_executor up to `loop_wait` seconds. Async executor will fire such event only in case if the function it was calling returned something what could be evaluated to boolean True. Such functionality is really needed to change the way how we are making decision about necessity of pg_rewind. It will require to have a local postgres running and for us it is really important to get such notification as soon as possible.
57 lines
1.6 KiB
Python
57 lines
1.6 KiB
Python
import logging
|
|
from threading import RLock, Thread
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class AsyncExecutor(object):
|
|
|
|
def __init__(self, ha_wakeup):
|
|
self._ha_wakeup = ha_wakeup
|
|
self._thread_lock = RLock()
|
|
self._scheduled_action = None
|
|
self._scheduled_action_lock = RLock()
|
|
|
|
@property
|
|
def busy(self):
|
|
return self.scheduled_action is not None
|
|
|
|
def schedule(self, action, immediately=False):
|
|
with self._scheduled_action_lock:
|
|
if self._scheduled_action is not None:
|
|
return self._scheduled_action
|
|
self._scheduled_action = action
|
|
return None
|
|
|
|
@property
|
|
def scheduled_action(self):
|
|
with self._scheduled_action_lock:
|
|
return self._scheduled_action
|
|
|
|
def reset_scheduled_action(self):
|
|
with self._scheduled_action_lock:
|
|
self._scheduled_action = None
|
|
|
|
def run(self, func, args=()):
|
|
wakeup = False
|
|
try:
|
|
# if the func returned something (not None) - wake up main HA loop
|
|
wakeup = func(*args) if args else func()
|
|
return wakeup
|
|
except:
|
|
logger.exception('Exception during execution of long running task %s', self.scheduled_action)
|
|
finally:
|
|
with self:
|
|
self.reset_scheduled_action()
|
|
if wakeup is not None:
|
|
self._ha_wakeup()
|
|
|
|
def run_async(self, func, args=()):
|
|
Thread(target=self.run, args=(func, args)).start()
|
|
|
|
def __enter__(self):
|
|
self._thread_lock.acquire()
|
|
|
|
def __exit__(self, *args):
|
|
self._thread_lock.release()
|