Compare commits

...

1 Commits

Author SHA1 Message Date
Arseny Sher
6d71a4fd31 Ignore zero sized cluster size in pageserver feedback.
To take into account that non zero shards send 0 size. Generally we should make
reporting sharding aware, but for now this will avoid breaking size limit after
https://github.com/neondatabase/neon/pull/6567
2024-02-01 17:07:48 +03:00
2 changed files with 21 additions and 5 deletions

View File

@@ -1804,11 +1804,21 @@ walprop_pg_finish_sync_safekeepers(WalProposer *wp, XLogRecPtr lsn)
static void static void
GetLatestNeonFeedback(PageserverFeedback *rf, WalProposer *wp) GetLatestNeonFeedback(PageserverFeedback *rf, WalProposer *wp)
{ {
int latest_safekeeper = 0; int latest_safekeeper = -1;
XLogRecPtr last_received_lsn = InvalidXLogRecPtr; XLogRecPtr last_received_lsn = InvalidXLogRecPtr;
for (int i = 0; i < wp->n_safekeepers; i++) for (int i = 0; i < wp->n_safekeepers; i++)
{ {
/*
* Non-zero shards don't known the timeline size and send zero.
* TODO: right now we ignore all feedback from non zero shards. We
* should make reporting sharding aware instead and do per shard
* aggregation as any lagging shard should trigger backpressure.
*/
if (wp->safekeeper[i].appendResponse.rf.currentClusterSize == 0)
continue;
if (wp->safekeeper[i].appendResponse.rf.last_received_lsn > last_received_lsn) if (wp->safekeeper[i].appendResponse.rf.last_received_lsn > last_received_lsn)
{ {
latest_safekeeper = i; latest_safekeeper = i;
@@ -1816,6 +1826,10 @@ GetLatestNeonFeedback(PageserverFeedback *rf, WalProposer *wp)
} }
} }
/* no feedback yet */
if (latest_safekeeper == -1)
return;
rf->currentClusterSize = wp->safekeeper[latest_safekeeper].appendResponse.rf.currentClusterSize; rf->currentClusterSize = wp->safekeeper[latest_safekeeper].appendResponse.rf.currentClusterSize;
rf->last_received_lsn = wp->safekeeper[latest_safekeeper].appendResponse.rf.last_received_lsn; rf->last_received_lsn = wp->safekeeper[latest_safekeeper].appendResponse.rf.last_received_lsn;
rf->disk_consistent_lsn = wp->safekeeper[latest_safekeeper].appendResponse.rf.disk_consistent_lsn; rf->disk_consistent_lsn = wp->safekeeper[latest_safekeeper].appendResponse.rf.disk_consistent_lsn;

View File

@@ -265,9 +265,9 @@ impl WalSendersShared {
/// Update aggregated pageserver feedback. LSNs (last_received, /// Update aggregated pageserver feedback. LSNs (last_received,
/// disk_consistent, remote_consistent) and reply timestamp are just /// disk_consistent, remote_consistent) and reply timestamp are just
/// maximized; timeline_size if taken from feedback with highest /// maximized; timeline_size if taken from feedback with highest
/// last_received lsn. This is generally reasonable, but we might want to /// last_received lsn except when it is 0 (sent from non zero shards). This
/// implement other policies once multiple pageservers start to be actively /// is generally reasonable, but we might want to implement other policies
/// used. /// once multiple pageservers start to be actively used.
fn update_ps_feedback(&mut self) { fn update_ps_feedback(&mut self) {
let init = PageserverFeedback::empty(); let init = PageserverFeedback::empty();
let acc = let acc =
@@ -276,7 +276,9 @@ impl WalSendersShared {
.flatten() .flatten()
.fold(init, |mut acc, ws_state| match ws_state.feedback { .fold(init, |mut acc, ws_state| match ws_state.feedback {
ReplicationFeedback::Pageserver(feedback) => { ReplicationFeedback::Pageserver(feedback) => {
if feedback.last_received_lsn > acc.last_received_lsn { if feedback.current_timeline_size != 0
&& feedback.last_received_lsn > acc.last_received_lsn
{
acc.current_timeline_size = feedback.current_timeline_size; acc.current_timeline_size = feedback.current_timeline_size;
} }
acc.last_received_lsn = acc.last_received_lsn =