From e6ef3c340a7c8e87d6f78af449bbc5ae10868608 Mon Sep 17 00:00:00 2001 From: Alexander Kukushkin Date: Fri, 15 May 2020 16:02:17 +0200 Subject: [PATCH] Wake up the main thread after checkpoint is done (#1524) Replicas are waiting for checkpoint indication via member key of the leader in DCS. The key is normally updated only one time per HA loop. Without waking the main thread up replicas will have to wait up to `loop_wait` seconds longer than necessary. --- patroni/ha.py | 2 +- patroni/postgresql/rewind.py | 8 +++++--- tests/test_rewind.py | 12 ++++++------ 3 files changed, 12 insertions(+), 10 deletions(-) diff --git a/patroni/ha.py b/patroni/ha.py index 7a0e0adb..07a8122e 100644 --- a/patroni/ha.py +++ b/patroni/ha.py @@ -945,7 +945,7 @@ class Ha(object): return msg # check if the node is ready to be used by pg_rewind - self._rewind.ensure_checkpoint_after_promote() + self._rewind.ensure_checkpoint_after_promote(self.wakeup) if self.is_standby_cluster(): # in case of standby cluster we don't really need to diff --git a/patroni/postgresql/rewind.py b/patroni/postgresql/rewind.py index 9dd30c7a..87a3626b 100644 --- a/patroni/postgresql/rewind.py +++ b/patroni/postgresql/rewind.py @@ -135,15 +135,17 @@ class Rewind(object): self._check_timeline_and_lsn(leader) return leader and leader.conn_url and self._state == REWIND_STATUS.NEED - def __checkpoint(self, task): + def __checkpoint(self, task, wakeup): try: result = self._postgresql.checkpoint() except Exception as e: result = 'Exception: ' + str(e) with task: task.complete(not bool(result)) + if task.result: + wakeup() - def ensure_checkpoint_after_promote(self): + def ensure_checkpoint_after_promote(self, wakeup): """After promote issue a CHECKPOINT from a new thread and asynchronously check the result. In case if CHECKPOINT failed, just check that timeline in pg_control was updated.""" @@ -157,7 +159,7 @@ class Rewind(object): return else: self._checkpoint_task = CriticalTask() - return Thread(target=self.__checkpoint, args=(self._checkpoint_task,)).start() + return Thread(target=self.__checkpoint, args=(self._checkpoint_task, wakeup)).start() if self._postgresql.get_master_timeline() == self._postgresql.pg_control_timeline(): self._state = REWIND_STATUS.CHECKPOINT diff --git a/tests/test_rewind.py b/tests/test_rewind.py index 88715beb..20739903 100644 --- a/tests/test_rewind.py +++ b/tests/test_rewind.py @@ -117,16 +117,16 @@ class TestRewind(BaseTestPostgresql): @patch.object(Postgresql, 'checkpoint') def test_ensure_checkpoint_after_promote(self, mock_checkpoint, mock_controldata): mock_checkpoint.return_value = None - self.r.ensure_checkpoint_after_promote() - self.r.ensure_checkpoint_after_promote() + self.r.ensure_checkpoint_after_promote(Mock()) + self.r.ensure_checkpoint_after_promote(Mock()) self.r.reset_state() mock_controldata.return_value = {"Latest checkpoint's TimeLineID": 1} mock_checkpoint.side_effect = Exception - self.r.ensure_checkpoint_after_promote() - self.r.ensure_checkpoint_after_promote() + self.r.ensure_checkpoint_after_promote(Mock()) + self.r.ensure_checkpoint_after_promote(Mock()) self.r.reset_state() mock_controldata.side_effect = TypeError - self.r.ensure_checkpoint_after_promote() - self.r.ensure_checkpoint_after_promote() + self.r.ensure_checkpoint_after_promote(Mock()) + self.r.ensure_checkpoint_after_promote(Mock())