From e8db6b91c5fe03dc8245695efe1c93213e7277c7 Mon Sep 17 00:00:00 2001 From: Arthur Petukhovsky Date: Mon, 25 Sep 2023 17:31:05 +0000 Subject: [PATCH] Move safekeeper feedback handling --- pgxn/neon/walproposer.c | 141 ++++-------------------------------- pgxn/neon/walproposer.h | 5 ++ pgxn/neon/walproposer_pg.c | 142 +++++++++++++++++++++++++++++++++++++ 3 files changed, 159 insertions(+), 129 deletions(-) diff --git a/pgxn/neon/walproposer.c b/pgxn/neon/walproposer.c index 0e79d07c76..2924fe78aa 100644 --- a/pgxn/neon/walproposer.c +++ b/pgxn/neon/walproposer.c @@ -84,7 +84,6 @@ static XLogRecPtr lastSentCommitLsn; /* last commitLsn broadcast to* * safekeepers */ static ProposerGreeting greetRequest; static VoteRequest voteRequest; /* Vote request for safekeeper */ -static AppendResponse quorumFeedback; /* * Minimal LSN which may be needed for recovery of some safekeeper, * record-aligned (first record which might not yet received by someone). @@ -137,7 +136,6 @@ static void BroadcastAppendRequest(void); static void HandleActiveState(Safekeeper *sk, uint32 events); static bool SendAppendRequests(Safekeeper *sk); static bool RecvAppendResponses(Safekeeper *sk); -static void CombineHotStanbyFeedbacks(HotStandbyFeedback * hs); static XLogRecPtr CalculateMinFlushLsn(void); static XLogRecPtr GetAcknowledgedByQuorumWALPosition(void); static void HandleSafekeeperResponse(void); @@ -341,7 +339,7 @@ WalProposerInit(XLogRecPtr flushRecPtr, uint64 systemId) greetRequest.tag = 'g'; greetRequest.protocolVersion = SK_PROTOCOL_VERSION; greetRequest.pgVersion = PG_VERSION_NUM; - pg_strong_random(&greetRequest.proposerId, sizeof(greetRequest.proposerId)); + walprop_pg.strong_random(&greetRequest.proposerId, sizeof(greetRequest.proposerId)); greetRequest.systemId = systemId; if (!neon_timeline) elog(FATAL, "neon.timeline_id is not provided"); @@ -959,8 +957,8 @@ HandleElectedProposer(void) else if (syncSafekeepers) { /* Sync is not needed: just exit */ - fprintf(stdout, "%X/%X\n", LSN_FORMAT_ARGS(propEpochStartLsn)); - exit(0); + walprop_pg.finish_sync_safekeepers(propEpochStartLsn); + /* unreachable */ } for (int i = 0; i < n_safekeepers; i++) @@ -1076,10 +1074,10 @@ DetermineEpochStartLsn(void) */ if (propEpochStartLsn == InvalidXLogRecPtr && !syncSafekeepers) { - propEpochStartLsn = truncateLsn = GetRedoStartLsn(); + propEpochStartLsn = truncateLsn = walprop_pg.get_redo_start_lsn(); if (timelineStartLsn == InvalidXLogRecPtr) { - timelineStartLsn = GetRedoStartLsn(); + timelineStartLsn = walprop_pg.get_redo_start_lsn(); } elog(LOG, "bumped epochStartLsn to the first record %X/%X", LSN_FORMAT_ARGS(propEpochStartLsn)); } @@ -1130,7 +1128,7 @@ DetermineEpochStartLsn(void) * Safekeepers don't skip header as they need continious stream of * data, so correct LSN for comparison. */ - if (SkipXLogPageHeader(propEpochStartLsn) != GetRedoStartLsn()) + if (SkipXLogPageHeader(propEpochStartLsn) != walprop_pg.get_redo_start_lsn()) { /* * However, allow to proceed if previously elected leader was me; @@ -1143,7 +1141,7 @@ DetermineEpochStartLsn(void) elog(PANIC, "collected propEpochStartLsn %X/%X, but basebackup LSN %X/%X", LSN_FORMAT_ARGS(propEpochStartLsn), - LSN_FORMAT_ARGS(GetRedoStartLsn())); + LSN_FORMAT_ARGS(walprop_pg.get_redo_start_lsn())); } } walprop_shared->mineLastElectedTerm = propTerm; @@ -1429,7 +1427,7 @@ SendAppendRequests(Safekeeper *sk) /* write the WAL itself */ enlargeStringInfo(&sk->outbuf, req->endLsn - req->beginLsn); - /* will raise error on failure */ + /* wal_read will raise error on failure */ walprop_pg.wal_read(sk->xlogreader, &sk->outbuf.data[sk->outbuf.len], req->beginLsn, @@ -1611,42 +1609,6 @@ ParsePageserverFeedbackMessage(StringInfo reply_message, PageserverFeedback * rf } } -/* - * Combine hot standby feedbacks from all safekeepers. - */ -static void -CombineHotStanbyFeedbacks(HotStandbyFeedback * hs) -{ - hs->ts = 0; - hs->xmin.value = ~0; /* largest unsigned value */ - hs->catalog_xmin.value = ~0; /* largest unsigned value */ - - for (int i = 0; i < n_safekeepers; i++) - { - if (safekeeper[i].appendResponse.hs.ts != 0) - { - HotStandbyFeedback *skhs = &safekeeper[i].appendResponse.hs; - if (FullTransactionIdIsNormal(skhs->xmin) - && FullTransactionIdPrecedes(skhs->xmin, hs->xmin)) - { - hs->xmin = skhs->xmin; - hs->ts = skhs->ts; - } - if (FullTransactionIdIsNormal(skhs->catalog_xmin) - && FullTransactionIdPrecedes(skhs->catalog_xmin, hs->xmin)) - { - hs->catalog_xmin = skhs->catalog_xmin; - hs->ts = skhs->ts; - } - } - } - - if (hs->xmin.value == ~0) - hs->xmin = InvalidFullTransactionId; - if (hs->catalog_xmin.value == ~0) - hs->catalog_xmin = InvalidFullTransactionId; -} - /* * Get minimum of flushed LSNs of all safekeepers, which is the LSN of the * last WAL record that can be safely discarded. @@ -1692,92 +1654,14 @@ GetAcknowledgedByQuorumWALPosition(void) return responses[n_safekeepers - quorum]; } -/* - * Get PageserverFeedback fields from the most advanced safekeeper - */ -static void -GetLatestNeonFeedback(PageserverFeedback * rf) -{ - int latest_safekeeper = 0; - XLogRecPtr last_received_lsn = InvalidXLogRecPtr; - - for (int i = 0; i < n_safekeepers; i++) - { - if (safekeeper[i].appendResponse.rf.last_received_lsn > last_received_lsn) - { - latest_safekeeper = i; - last_received_lsn = safekeeper[i].appendResponse.rf.last_received_lsn; - } - } - - rf->currentClusterSize = safekeeper[latest_safekeeper].appendResponse.rf.currentClusterSize; - rf->last_received_lsn = safekeeper[latest_safekeeper].appendResponse.rf.last_received_lsn; - rf->disk_consistent_lsn = safekeeper[latest_safekeeper].appendResponse.rf.disk_consistent_lsn; - rf->remote_consistent_lsn = safekeeper[latest_safekeeper].appendResponse.rf.remote_consistent_lsn; - rf->replytime = safekeeper[latest_safekeeper].appendResponse.rf.replytime; - - elog(DEBUG2, "GetLatestNeonFeedback: currentClusterSize %lu," - " last_received_lsn %X/%X, disk_consistent_lsn %X/%X, remote_consistent_lsn %X/%X, replytime %lu", - rf->currentClusterSize, - LSN_FORMAT_ARGS(rf->last_received_lsn), - LSN_FORMAT_ARGS(rf->disk_consistent_lsn), - LSN_FORMAT_ARGS(rf->remote_consistent_lsn), - rf->replytime); - - walprop_pg.replication_feedback_set(rf); -} - static void HandleSafekeeperResponse(void) { - HotStandbyFeedback hsFeedback; XLogRecPtr minQuorumLsn; - XLogRecPtr diskConsistentLsn; XLogRecPtr minFlushLsn; minQuorumLsn = GetAcknowledgedByQuorumWALPosition(); - diskConsistentLsn = quorumFeedback.rf.disk_consistent_lsn; - - if (!syncSafekeepers) - { - /* Get PageserverFeedback fields from the most advanced safekeeper */ - GetLatestNeonFeedback(&quorumFeedback.rf); - SetZenithCurrentClusterSize(quorumFeedback.rf.currentClusterSize); - } - - if (minQuorumLsn > quorumFeedback.flushLsn || diskConsistentLsn != quorumFeedback.rf.disk_consistent_lsn) - { - - if (minQuorumLsn > quorumFeedback.flushLsn) - quorumFeedback.flushLsn = minQuorumLsn; - - /* advance the replication slot */ - if (!syncSafekeepers) - ProcessStandbyReply( - /* write_lsn - This is what durably stored in WAL service. */ - quorumFeedback.flushLsn, - /* flush_lsn - This is what durably stored in WAL service. */ - quorumFeedback.flushLsn, - - /* - * apply_lsn - This is what processed and durably saved at* - * pageserver. - */ - quorumFeedback.rf.disk_consistent_lsn, - walprop_pg.get_current_timestamp(), false); - } - - CombineHotStanbyFeedbacks(&hsFeedback); - if (hsFeedback.ts != 0 && memcmp(&hsFeedback, &quorumFeedback.hs, sizeof hsFeedback) != 0) - { - quorumFeedback.hs = hsFeedback; - if (!syncSafekeepers) - ProcessStandbyHSFeedback(hsFeedback.ts, - XidFromFullTransactionId(hsFeedback.xmin), - EpochFromFullTransactionId(hsFeedback.xmin), - XidFromFullTransactionId(hsFeedback.catalog_xmin), - EpochFromFullTransactionId(hsFeedback.catalog_xmin)); - } + walprop_pg.process_safekeeper_feedback(safekeeper, n_safekeepers, syncSafekeepers, minQuorumLsn); /* * Try to advance truncateLsn to minFlushLsn, which is the last record @@ -1802,8 +1686,7 @@ HandleSafekeeperResponse(void) * Advance the replication slot to free up old WAL files. Note that * slot doesn't exist if we are in syncSafekeepers mode. */ - if (MyReplicationSlot) - PhysicalConfirmReceivedLocation(truncateLsn); + walprop_pg.confirm_wal_streamed(truncateLsn); } /* @@ -1850,8 +1733,8 @@ HandleSafekeeperResponse(void) */ BroadcastAppendRequest(); - fprintf(stdout, "%X/%X\n", LSN_FORMAT_ARGS(propEpochStartLsn)); - exit(0); + walprop_pg.finish_sync_safekeepers(propEpochStartLsn); + /* unreachable */ } } } diff --git a/pgxn/neon/walproposer.h b/pgxn/neon/walproposer.h index c2b08af3e7..98cf6c6806 100644 --- a/pgxn/neon/walproposer.h +++ b/pgxn/neon/walproposer.h @@ -481,6 +481,11 @@ typedef struct walproposer_api void (*update_event_set) (Safekeeper * sk, uint32 events); void (*add_safekeeper_event_set) (Safekeeper * sk, uint32 events); int (*wait_event_set) (long timeout, Safekeeper **sk, uint32 *events); + bool (*strong_random) (void *buf, size_t len); + XLogRecPtr (*get_redo_start_lsn) (void); + void (*finish_sync_safekeepers) (XLogRecPtr lsn); + void (*process_safekeeper_feedback) (Safekeeper * safekeepers, int n_safekeepers, bool isSync, XLogRecPtr commitLsn); + void (*confirm_wal_streamed) (XLogRecPtr lsn); } walproposer_api; extern const walproposer_api walprop_pg; diff --git a/pgxn/neon/walproposer_pg.c b/pgxn/neon/walproposer_pg.c index 13f8160ce0..8995c9d6b8 100644 --- a/pgxn/neon/walproposer_pg.c +++ b/pgxn/neon/walproposer_pg.c @@ -43,6 +43,8 @@ char *wal_acceptors_list = ""; int wal_acceptor_reconnect_timeout = 1000; int wal_acceptor_connection_timeout = 10000; +static AppendResponse quorumFeedback; + #define WAL_PROPOSER_SLOT_NAME "wal_proposer_slot" static WalproposerShmemState * walprop_shared; @@ -1432,6 +1434,141 @@ walprop_pg_wait_event_set(long timeout, Safekeeper **sk, uint32 *events) return rc; } +static void +walprop_pg_finish_sync_safekeepers(XLogRecPtr lsn) +{ + fprintf(stdout, "%X/%X\n", LSN_FORMAT_ARGS(lsn)); + exit(0); +} + +/* + * Get PageserverFeedback fields from the most advanced safekeeper + */ +static void +GetLatestNeonFeedback(PageserverFeedback * rf, Safekeeper * safekeepers, int n_safekeepers) +{ + int latest_safekeeper = 0; + XLogRecPtr last_received_lsn = InvalidXLogRecPtr; + + for (int i = 0; i < n_safekeepers; i++) + { + if (safekeepers[i].appendResponse.rf.last_received_lsn > last_received_lsn) + { + latest_safekeeper = i; + last_received_lsn = safekeepers[i].appendResponse.rf.last_received_lsn; + } + } + + rf->currentClusterSize = safekeepers[latest_safekeeper].appendResponse.rf.currentClusterSize; + rf->last_received_lsn = safekeepers[latest_safekeeper].appendResponse.rf.last_received_lsn; + rf->disk_consistent_lsn = safekeepers[latest_safekeeper].appendResponse.rf.disk_consistent_lsn; + rf->remote_consistent_lsn = safekeepers[latest_safekeeper].appendResponse.rf.remote_consistent_lsn; + rf->replytime = safekeepers[latest_safekeeper].appendResponse.rf.replytime; + + elog(DEBUG2, "GetLatestNeonFeedback: currentClusterSize %lu," + " last_received_lsn %X/%X, disk_consistent_lsn %X/%X, remote_consistent_lsn %X/%X, replytime %lu", + rf->currentClusterSize, + LSN_FORMAT_ARGS(rf->last_received_lsn), + LSN_FORMAT_ARGS(rf->disk_consistent_lsn), + LSN_FORMAT_ARGS(rf->remote_consistent_lsn), + rf->replytime); + + walprop_pg.replication_feedback_set(rf); +} + +/* + * Combine hot standby feedbacks from all safekeepers. + */ +static void +CombineHotStanbyFeedbacks(HotStandbyFeedback * hs, Safekeeper * safekeepers, int n_safekeepers) +{ + hs->ts = 0; + hs->xmin.value = ~0; /* largest unsigned value */ + hs->catalog_xmin.value = ~0; /* largest unsigned value */ + + for (int i = 0; i < n_safekeepers; i++) + { + if (safekeepers[i].appendResponse.hs.ts != 0) + { + HotStandbyFeedback *skhs = &safekeepers[i].appendResponse.hs; + if (FullTransactionIdIsNormal(skhs->xmin) + && FullTransactionIdPrecedes(skhs->xmin, hs->xmin)) + { + hs->xmin = skhs->xmin; + hs->ts = skhs->ts; + } + if (FullTransactionIdIsNormal(skhs->catalog_xmin) + && FullTransactionIdPrecedes(skhs->catalog_xmin, hs->xmin)) + { + hs->catalog_xmin = skhs->catalog_xmin; + hs->ts = skhs->ts; + } + } + } + + if (hs->xmin.value == ~0) + hs->xmin = InvalidFullTransactionId; + if (hs->catalog_xmin.value == ~0) + hs->catalog_xmin = InvalidFullTransactionId; +} + +static void +walprop_pg_process_safekeeper_feedback(Safekeeper * safekeepers, int n_safekeepers, bool isSync, XLogRecPtr commitLsn) +{ + HotStandbyFeedback hsFeedback; + XLogRecPtr diskConsistentLsn; + + diskConsistentLsn = quorumFeedback.rf.disk_consistent_lsn; + + if (!isSync) + { + /* Get PageserverFeedback fields from the most advanced safekeeper */ + GetLatestNeonFeedback(&quorumFeedback.rf, safekeepers, n_safekeepers); + SetZenithCurrentClusterSize(quorumFeedback.rf.currentClusterSize); + } + + if (commitLsn > quorumFeedback.flushLsn || diskConsistentLsn != quorumFeedback.rf.disk_consistent_lsn) + { + + if (commitLsn > quorumFeedback.flushLsn) + quorumFeedback.flushLsn = commitLsn; + + /* advance the replication slot */ + if (!isSync) + ProcessStandbyReply( + /* write_lsn - This is what durably stored in WAL service. */ + quorumFeedback.flushLsn, + /* flush_lsn - This is what durably stored in WAL service. */ + quorumFeedback.flushLsn, + + /* + * apply_lsn - This is what processed and durably saved at* + * pageserver. + */ + quorumFeedback.rf.disk_consistent_lsn, + walprop_pg.get_current_timestamp(), false); + } + + CombineHotStanbyFeedbacks(&hsFeedback, safekeepers, n_safekeepers); + if (hsFeedback.ts != 0 && memcmp(&hsFeedback, &quorumFeedback.hs, sizeof hsFeedback) != 0) + { + quorumFeedback.hs = hsFeedback; + if (!isSync) + ProcessStandbyHSFeedback(hsFeedback.ts, + XidFromFullTransactionId(hsFeedback.xmin), + EpochFromFullTransactionId(hsFeedback.xmin), + XidFromFullTransactionId(hsFeedback.catalog_xmin), + EpochFromFullTransactionId(hsFeedback.catalog_xmin)); + } +} + +static void +walprop_pg_confirm_wal_streamed(XLogRecPtr lsn) +{ + if (MyReplicationSlot) + PhysicalConfirmReceivedLocation(lsn); +} + /* * Temporary globally exported walproposer API for postgres. */ @@ -1466,4 +1603,9 @@ const walproposer_api walprop_pg = { .update_event_set = walprop_pg_update_event_set, .add_safekeeper_event_set = walprop_pg_add_safekeeper_event_set, .wait_event_set = walprop_pg_wait_event_set, + .strong_random = pg_strong_random, + .get_redo_start_lsn = GetRedoStartLsn, + .finish_sync_safekeepers = walprop_pg_finish_sync_safekeepers, + .process_safekeeper_feedback = walprop_pg_process_safekeeper_feedback, + .confirm_wal_streamed = walprop_pg_confirm_wal_streamed, };