public inbox for [email protected]help / color / mirror / Atom feed
[PATCH 3/8] Move processing of NULLs from BRIN support functions 4+ messages / 4 participants [nested] [flat]
* [PATCH 3/8] Move processing of NULLs from BRIN support functions @ 2020-04-02 00:56 Tomas Vondra <[email protected]> 0 siblings, 0 replies; 4+ messages in thread From: Tomas Vondra @ 2020-04-02 00:56 UTC (permalink / raw) --- src/backend/access/brin/brin.c | 260 ++++++++++++++--------- src/backend/access/brin/brin_inclusion.c | 44 +--- src/backend/access/brin/brin_minmax.c | 41 +--- src/include/access/brin_internal.h | 3 + 4 files changed, 169 insertions(+), 179 deletions(-) diff --git a/src/backend/access/brin/brin.c b/src/backend/access/brin/brin.c index caf7b62688..a9c44c0b82 100644 --- a/src/backend/access/brin/brin.c +++ b/src/backend/access/brin/brin.c @@ -35,6 +35,7 @@ #include "storage/freespace.h" #include "utils/acl.h" #include "utils/builtins.h" +#include "utils/datum.h" #include "utils/index_selfuncs.h" #include "utils/memutils.h" #include "utils/rel.h" @@ -77,7 +78,9 @@ static void form_and_insert_tuple(BrinBuildState *state); static void union_tuples(BrinDesc *bdesc, BrinMemTuple *a, BrinTuple *b); static void brin_vacuum_scan(Relation idxrel, BufferAccessStrategy strategy); - +static bool add_values_to_range(Relation idxRel, BrinDesc *bdesc, + BrinMemTuple *dtup, Datum *values, bool *nulls); +static bool check_null_keys(BrinValues *bval, ScanKey *nullkeys, int nnullkeys); /* * BRIN handler function: return IndexAmRoutine with access method parameters @@ -178,7 +181,6 @@ brininsert(Relation idxRel, Datum *values, bool *nulls, OffsetNumber off; BrinTuple *brtup; BrinMemTuple *dtup; - int keyno; CHECK_FOR_INTERRUPTS(); @@ -242,31 +244,7 @@ brininsert(Relation idxRel, Datum *values, bool *nulls, dtup = brin_deform_tuple(bdesc, brtup, NULL); - /* - * Compare the key values of the new tuple to the stored index values; - * our deformed tuple will get updated if the new tuple doesn't fit - * the original range (note this means we can't break out of the loop - * early). Make a note of whether this happens, so that we know to - * insert the modified tuple later. - */ - for (keyno = 0; keyno < bdesc->bd_tupdesc->natts; keyno++) - { - Datum result; - BrinValues *bval; - FmgrInfo *addValue; - - bval = &dtup->bt_columns[keyno]; - addValue = index_getprocinfo(idxRel, keyno + 1, - BRIN_PROCNUM_ADDVALUE); - result = FunctionCall4Coll(addValue, - idxRel->rd_indcollation[keyno], - PointerGetDatum(bdesc), - PointerGetDatum(bval), - values[keyno], - nulls[keyno]); - /* if that returned true, we need to insert the updated tuple */ - need_insert |= DatumGetBool(result); - } + need_insert = add_values_to_range(idxRel, bdesc, dtup, values, nulls); if (!need_insert) { @@ -359,6 +337,7 @@ brinbeginscan(Relation r, int nkeys, int norderbys) return scan; } + /* * Execute the index scan. * @@ -562,69 +541,31 @@ bringetbitmap(IndexScanDesc scan, TIDBitmap *tbm) (nkeys[attno - 1] <= scan->numberOfKeys)); /* - * First check if there are any IS [NOT] NULL scan keys, and - * if we're violating them. In that case we can terminate - * early, without invoking the support function. + * First check if there are any IS [NOT] NULL scan keys, + * and if we're violating them. In that case we can + * terminate early, without invoking the support function. * * As there may be more keys, we can only detemine mismatch * within this loop. */ - for (keyno = 0; (keyno < nnullkeys[attno - 1]); keyno++) + if (bdesc->bd_info[attno - 1]->oi_regular_nulls && + !check_null_keys(bval, nullkeys[attno - 1], + nnullkeys[attno - 1])) { - ScanKey key = nullkeys[attno - 1][keyno]; - - Assert(key->sk_attno == bval->bv_attno); - - /* interrupt the loop as soon as we find a mismatch */ - if (!addrange) - break; - - /* handle IS NULL/IS NOT NULL tests */ - if (key->sk_flags & SK_ISNULL) - { - /* IS NULL scan key, but range has no NULLs */ - if (key->sk_flags & SK_SEARCHNULL) - { - if (!bval->bv_allnulls && !bval->bv_hasnulls) - addrange = false; - - continue; - } - - /* - * For IS NOT NULL, we can only skip ranges that are - * known to have only nulls. - */ - if (key->sk_flags & SK_SEARCHNOTNULL) - { - if (bval->bv_allnulls) - addrange = false; - - continue; - } - - /* - * Neither IS NULL nor IS NOT NULL was used; assume all - * indexable operators are strict and thus return false - * with NULL value in the scan key. - */ - addrange = false; - } - } - - /* - * If any of the IS [NOT] NULL keys failed, the page range as - * a whole can't pass. So terminate the loop. - */ - if (!addrange) + /* + * If any of the IS [NOT] NULL keys failed, the page + * range as a whole can't pass. So terminate the loop. + */ + addrange = false; break; + } /* - * So either there are no IS [NOT] NULL keys, or all passed. If - * there are no regular scan keys, we're done - the page range - * matches. If there are regular keys, but the page range is - * marked as 'all nulls' it can't possibly pass (we're assuming - * the operators are strict). + * So either there are no IS [NOT] NULL keys, or all passed. + * If there are no regular scan keys, we're done - the page + * range matches. If there are regular keys, but the page + * range is marked as 'all nulls' it can't possibly pass + * (we're assuming the operators are strict). */ /* No regular scan keys - page range as a whole passes. */ @@ -772,7 +713,6 @@ brinbuildCallback(Relation index, { BrinBuildState *state = (BrinBuildState *) brstate; BlockNumber thisblock; - int i; thisblock = ItemPointerGetBlockNumber(tid); @@ -801,25 +741,8 @@ brinbuildCallback(Relation index, } /* Accumulate the current tuple into the running state */ - for (i = 0; i < state->bs_bdesc->bd_tupdesc->natts; i++) - { - FmgrInfo *addValue; - BrinValues *col; - Form_pg_attribute attr = TupleDescAttr(state->bs_bdesc->bd_tupdesc, i); - - col = &state->bs_dtuple->bt_columns[i]; - addValue = index_getprocinfo(index, i + 1, - BRIN_PROCNUM_ADDVALUE); - - /* - * Update dtuple state, if and as necessary. - */ - FunctionCall4Coll(addValue, - attr->attcollation, - PointerGetDatum(state->bs_bdesc), - PointerGetDatum(col), - values[i], isnull[i]); - } + (void) add_values_to_range(index, state->bs_bdesc, state->bs_dtuple, + values, isnull); } /* @@ -1598,6 +1521,39 @@ union_tuples(BrinDesc *bdesc, BrinMemTuple *a, BrinTuple *b) FmgrInfo *unionFn; BrinValues *col_a = &a->bt_columns[keyno]; BrinValues *col_b = &db->bt_columns[keyno]; + BrinOpcInfo *opcinfo = bdesc->bd_info[keyno]; + + if (opcinfo->oi_regular_nulls) + { + /* Adjust "hasnulls". */ + if (!col_a->bv_hasnulls && col_b->bv_hasnulls) + col_a->bv_hasnulls = true; + + /* If there are no values in B, there's nothing left to do. */ + if (col_b->bv_allnulls) + continue; + + /* + * Adjust "allnulls". If A doesn't have values, just copy the + * values from B into A, and we're done. We cannot run the + * operators in this case, because values in A might contain + * garbage. Note we already established that B contains values. + */ + if (col_a->bv_allnulls) + { + int i; + + col_a->bv_allnulls = false; + + for (i = 0; i < opcinfo->oi_nstored; i++) + col_a->bv_values[i] = + datumCopy(col_b->bv_values[i], + opcinfo->oi_typcache[i]->typbyval, + opcinfo->oi_typcache[i]->typlen); + + continue; + } + } unionFn = index_getprocinfo(bdesc->bd_index, keyno + 1, BRIN_PROCNUM_UNION); @@ -1651,3 +1607,103 @@ brin_vacuum_scan(Relation idxrel, BufferAccessStrategy strategy) */ FreeSpaceMapVacuum(idxrel); } + +static bool +add_values_to_range(Relation idxRel, BrinDesc *bdesc, BrinMemTuple *dtup, + Datum *values, bool *nulls) +{ + int keyno; + bool modified = false; + + /* + * Compare the key values of the new tuple to the stored index values; + * our deformed tuple will get updated if the new tuple doesn't fit + * the original range (note this means we can't break out of the loop + * early). Make a note of whether this happens, so that we know to + * insert the modified tuple later. + */ + for (keyno = 0; keyno < bdesc->bd_tupdesc->natts; keyno++) + { + Datum result; + BrinValues *bval; + FmgrInfo *addValue; + + bval = &dtup->bt_columns[keyno]; + + if (bdesc->bd_info[keyno]->oi_regular_nulls && nulls[keyno]) + { + /* + * If the new value is null, we record that we saw it if it's + * the first one; otherwise, there's nothing to do. + */ + if (!bval->bv_hasnulls) + { + bval->bv_hasnulls = true; + modified = true; + } + + continue; + } + + addValue = index_getprocinfo(idxRel, keyno + 1, + BRIN_PROCNUM_ADDVALUE); + result = FunctionCall4Coll(addValue, + idxRel->rd_indcollation[keyno], + PointerGetDatum(bdesc), + PointerGetDatum(bval), + values[keyno], + nulls[keyno]); + /* if that returned true, we need to insert the updated tuple */ + modified |= DatumGetBool(result); + } + + return modified; +} + +static bool +check_null_keys(BrinValues *bval, ScanKey *nullkeys, int nnullkeys) +{ + int keyno; + + /* + * First check if there are any IS [NOT] NULL scan keys, and if we're + * violating them. + */ + for (keyno = 0; keyno < nnullkeys; keyno++) + { + ScanKey key = nullkeys[keyno]; + + Assert(key->sk_attno == bval->bv_attno); + + /* Handle only IS NULL/IS NOT NULL tests */ + if (!(key->sk_flags & SK_ISNULL)) + continue; + + if (key->sk_flags & SK_SEARCHNULL) + { + /* IS NULL scan key, but range has no NULLs */ + if (!bval->bv_allnulls && !bval->bv_hasnulls) + return false; + } + else if (key->sk_flags & SK_SEARCHNOTNULL) + { + /* + * For IS NOT NULL, we can only skip ranges that are known to + * have only nulls. + */ + if (bval->bv_allnulls) + return false; + } + else + { + /* + * Neither IS NULL nor IS NOT NULL was used; assume all indexable + * operators are strict and thus return false with NULL value in + * the scan key. + */ + return false; + } + } + + return true; +} diff --git a/src/backend/access/brin/brin_inclusion.c b/src/backend/access/brin/brin_inclusion.c index 22edc6b46f..59503b6f68 100644 --- a/src/backend/access/brin/brin_inclusion.c +++ b/src/backend/access/brin/brin_inclusion.c @@ -110,6 +110,7 @@ brin_inclusion_opcinfo(PG_FUNCTION_ARGS) */ result = palloc0(MAXALIGN(SizeofBrinOpcInfo(3)) + sizeof(InclusionOpaque)); result->oi_nstored = 3; + result->oi_regular_nulls = true; result->oi_opaque = (InclusionOpaque *) MAXALIGN((char *) result + SizeofBrinOpcInfo(3)); @@ -141,7 +142,7 @@ brin_inclusion_add_value(PG_FUNCTION_ARGS) BrinDesc *bdesc = (BrinDesc *) PG_GETARG_POINTER(0); BrinValues *column = (BrinValues *) PG_GETARG_POINTER(1); Datum newval = PG_GETARG_DATUM(2); - bool isnull = PG_GETARG_BOOL(3); + bool isnull PG_USED_FOR_ASSERTS_ONLY = PG_GETARG_BOOL(3); Oid colloid = PG_GET_COLLATION(); FmgrInfo *finfo; Datum result; @@ -149,18 +150,7 @@ brin_inclusion_add_value(PG_FUNCTION_ARGS) AttrNumber attno; Form_pg_attribute attr; - /* - * If the new value is null, we record that we saw it if it's the first - * one; otherwise, there's nothing to do. - */ - if (isnull) - { - if (column->bv_hasnulls) - PG_RETURN_BOOL(false); - - column->bv_hasnulls = true; - PG_RETURN_BOOL(true); - } + Assert(!isnull); attno = column->bv_attno; attr = TupleDescAttr(bdesc->bd_tupdesc, attno - 1); @@ -510,37 +500,11 @@ brin_inclusion_union(PG_FUNCTION_ARGS) Datum result; Assert(col_a->bv_attno == col_b->bv_attno); - - /* Adjust "hasnulls". */ - if (!col_a->bv_hasnulls && col_b->bv_hasnulls) - col_a->bv_hasnulls = true; - - /* If there are no values in B, there's nothing left to do. */ - if (col_b->bv_allnulls) - PG_RETURN_VOID(); + Assert(!col_a->bv_allnulls && !col_b->bv_allnulls); attno = col_a->bv_attno; attr = TupleDescAttr(bdesc->bd_tupdesc, attno - 1); - /* - * Adjust "allnulls". If A doesn't have values, just copy the values from - * B into A, and we're done. We cannot run the operators in this case, - * because values in A might contain garbage. Note we already established - * that B contains values. - */ - if (col_a->bv_allnulls) - { - col_a->bv_allnulls = false; - col_a->bv_values[INCLUSION_UNION] = - datumCopy(col_b->bv_values[INCLUSION_UNION], - attr->attbyval, attr->attlen); - col_a->bv_values[INCLUSION_UNMERGEABLE] = - col_b->bv_values[INCLUSION_UNMERGEABLE]; - col_a->bv_values[INCLUSION_CONTAINS_EMPTY] = - col_b->bv_values[INCLUSION_CONTAINS_EMPTY]; - PG_RETURN_VOID(); - } - /* If B includes empty elements, mark A similarly, if needed. */ if (!DatumGetBool(col_a->bv_values[INCLUSION_CONTAINS_EMPTY]) && DatumGetBool(col_b->bv_values[INCLUSION_CONTAINS_EMPTY])) diff --git a/src/backend/access/brin/brin_minmax.c b/src/backend/access/brin/brin_minmax.c index 7a7bd21cec..8882eec12c 100644 --- a/src/backend/access/brin/brin_minmax.c +++ b/src/backend/access/brin/brin_minmax.c @@ -48,6 +48,7 @@ brin_minmax_opcinfo(PG_FUNCTION_ARGS) result = palloc0(MAXALIGN(SizeofBrinOpcInfo(2)) + sizeof(MinmaxOpaque)); result->oi_nstored = 2; + result->oi_regular_nulls = true; result->oi_opaque = (MinmaxOpaque *) MAXALIGN((char *) result + SizeofBrinOpcInfo(2)); result->oi_typcache[0] = result->oi_typcache[1] = @@ -69,7 +70,7 @@ brin_minmax_add_value(PG_FUNCTION_ARGS) BrinDesc *bdesc = (BrinDesc *) PG_GETARG_POINTER(0); BrinValues *column = (BrinValues *) PG_GETARG_POINTER(1); Datum newval = PG_GETARG_DATUM(2); - bool isnull = PG_GETARG_DATUM(3); + bool isnull PG_USED_FOR_ASSERTS_ONLY = PG_GETARG_DATUM(3); Oid colloid = PG_GET_COLLATION(); FmgrInfo *cmpFn; Datum compar; @@ -77,18 +78,7 @@ brin_minmax_add_value(PG_FUNCTION_ARGS) Form_pg_attribute attr; AttrNumber attno; - /* - * If the new value is null, we record that we saw it if it's the first - * one; otherwise, there's nothing to do. - */ - if (isnull) - { - if (column->bv_hasnulls) - PG_RETURN_BOOL(false); - - column->bv_hasnulls = true; - PG_RETURN_BOOL(true); - } + Assert(!isnull); attno = column->bv_attno; attr = TupleDescAttr(bdesc->bd_tupdesc, attno - 1); @@ -245,34 +235,11 @@ brin_minmax_union(PG_FUNCTION_ARGS) bool needsadj; Assert(col_a->bv_attno == col_b->bv_attno); - - /* Adjust "hasnulls" */ - if (!col_a->bv_hasnulls && col_b->bv_hasnulls) - col_a->bv_hasnulls = true; - - /* If there are no values in B, there's nothing left to do */ - if (col_b->bv_allnulls) - PG_RETURN_VOID(); + Assert(!col_a->bv_allnulls && !col_b->bv_allnulls); attno = col_a->bv_attno; attr = TupleDescAttr(bdesc->bd_tupdesc, attno - 1); - /* - * Adjust "allnulls". If A doesn't have values, just copy the values from - * B into A, and we're done. We cannot run the operators in this case, - * because values in A might contain garbage. Note we already established - * that B contains values. - */ - if (col_a->bv_allnulls) - { - col_a->bv_allnulls = false; - col_a->bv_values[0] = datumCopy(col_b->bv_values[0], - attr->attbyval, attr->attlen); - col_a->bv_values[1] = datumCopy(col_b->bv_values[1], - attr->attbyval, attr->attlen); - PG_RETURN_VOID(); - } - /* Adjust minimum, if B's min is less than A's min */ finfo = minmax_get_strategy_procinfo(bdesc, attno, attr->atttypid, BTLessStrategyNumber); diff --git a/src/include/access/brin_internal.h b/src/include/access/brin_internal.h index 9ffc9100c0..7c4f3da0a0 100644 --- a/src/include/access/brin_internal.h +++ b/src/include/access/brin_internal.h @@ -27,6 +27,9 @@ typedef struct BrinOpcInfo /* Number of columns stored in an index column of this opclass */ uint16 oi_nstored; + /* Regular processing of NULLs in BrinValues? */ + bool oi_regular_nulls; + /* Opaque pointer for the opclass' private use */ void *oi_opaque; -- 2.25.4 --jsivyprvf2oxfjz3 Content-Type: text/plain; charset=us-ascii Content-Disposition: attachment; filename="0004-BRIN-bloom-indexes-20200807.patch" ^ permalink raw reply [nested|flat] 4+ messages in thread
* Allow async standbys wait for sync replication (was: Disallow quorum uncommitted (with synchronous standbys) txns in logical replication subscribers) @ 2022-02-25 15:01 Bharath Rupireddy <[email protected]> 0 siblings, 1 reply; 4+ messages in thread From: Bharath Rupireddy @ 2022-02-25 15:01 UTC (permalink / raw) To: SATYANARAYANA NARLAPURAM <[email protected]>; +Cc: PostgreSQL Hackers <[email protected]> On Thu, Jan 6, 2022 at 1:29 PM SATYANARAYANA NARLAPURAM <[email protected]> wrote: > > Consider a cluster formation where we have a Primary(P), Sync Replica(S1), and multiple async replicas for disaster recovery and read scaling (within the region and outside the region). In this setup, S1 is the preferred failover target in an event of the primary failure. When a transaction is committed on the primary, it is not acknowledged to the client until the primary gets an acknowledgment from the sync standby that the WAL is flushed to the disk (assume synchrnous_commit configuration is remote_flush). However, walsenders corresponds to the async replica on the primary don't wait for the flush acknowledgment from the primary and send the WAL to the async standbys (also any logical replication/decoding clients). So it is possible for the async replicas and logical client ahead of the sync replica. If a failover is initiated in such a scenario, to bring the formation into a healthy state we have to either > > run the pg_rewind on the async replicas for them to reconnect with the new primary or > collect the latest WAL across the replicas and feed the standby. > > Both these operations are involved, error prone, and can cause multiple minutes of downtime if done manually. In addition, there is a window where the async replicas can show the data that was neither acknowledged to the client nor committed on standby. Logical clients if they are ahead may need to reseed the data as no easy rewind option for them. > > I would like to propose a GUC send_Wal_after_quorum_committed which when set to ON, walsenders corresponds to async standbys and logical replication workers wait until the LSN is quorum committed on the primary before sending it to the standby. This not only simplifies the post failover steps but avoids unnecessary downtime for the async replicas. Thoughts? Thanks Satya and others for the inputs. Here's the v1 patch that basically allows async wal senders to wait until the sync standbys report their flush lsn back to the primary. Please let me know your thoughts. I've done pgbench testing to see if the patch causes any problems. I ran tests two times, there isn't much difference in the txns per seconds (tps), although there's a delay in the async standby receiving the WAL, after all, that's the feature we are pursuing. [1] HEAD or WITHOUT PATCH: ./pgbench -c 10 -t 500 -P 10 testdb transaction type: <builtin: TPC-B (sort of)> scaling factor: 100 query mode: simple number of clients: 10 number of threads: 1 number of transactions per client: 500 number of transactions actually processed: 5000/5000 latency average = 247.395 ms latency stddev = 74.409 ms initial connection time = 13.622 ms tps = 39.713114 (without initial connection time) PATCH: ./pgbench -c 10 -t 500 -P 10 testdb transaction type: <builtin: TPC-B (sort of)> scaling factor: 100 query mode: simple number of clients: 10 number of threads: 1 number of transactions per client: 500 number of transactions actually processed: 5000/5000 latency average = 251.757 ms latency stddev = 72.846 ms initial connection time = 13.025 ms tps = 39.315862 (without initial connection time) TEST SETUP: primary in region 1 async standby 1 in the same region as that of the primary region 1 i.e. close to primary sync standby 1 in region 2 sync standby 2 in region 3 an archive location in a region different from the primary and standbys regions, region 4 Note that I intentionally kept sync standbys in regions far from primary because it allows sync standbys to receive WAL a bit late by default, which works well for our testing. PGBENCH SETUP: ./psql -d postgres -c "drop database testdb" ./psql -d postgres -c "create database testdb" ./pgbench -i -s 100 testdb ./psql -d testdb -c "\dt" ./psql -d testdb -c "SELECT pg_size_pretty(pg_database_size('testdb'))" ./pgbench -c 10 -t 500 -P 10 testdb Regards, Bharath Rupireddy. Attachments: [application/octet-stream] v1-0001-Allow-async-standbys-wait-for-sync-replication.patch (11.6K, ../../CALj2ACVUa8WddVDS20QmVKNwTbeOQqy4zy59NPzh8NnLipYZGw@mail.gmail.com/2-v1-0001-Allow-async-standbys-wait-for-sync-replication.patch) download | inline diff: From b7abfc772ce1b0ce66a0854241a290cf3beb91eb Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy <[email protected]> Date: Tue, 15 Feb 2022 14:56:33 +0000 Subject: [PATCH v1] Allow async standbys wait for sync replication --- doc/src/sgml/config.sgml | 24 +++ doc/src/sgml/monitoring.sgml | 4 + src/backend/replication/walsender.c | 155 ++++++++++++++++++ src/backend/utils/activity/wait_event.c | 3 + src/backend/utils/misc/guc.c | 9 + src/backend/utils/misc/postgresql.conf.sample | 3 + src/include/replication/walsender.h | 1 + src/include/utils/wait_event.h | 3 +- 8 files changed, 201 insertions(+), 1 deletion(-) diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index 53b361e7a9..63e8c226cb 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -4245,6 +4245,30 @@ restore_command = 'copy "C:\\server\\archivedir\\%f" "%p"' # Windows </listitem> </varlistentry> + <varlistentry id="guc-async-standbys-wait-for-sync_replication" xreflabel="async_standbys_wait_for_sync_replication"> + <term><varname>async_standbys_wait_for_sync_replication</varname> (<type>boolean</type>) + <indexterm> + <primary><varname>async_standbys_wait_for_sync_replication</varname> configuration parameter</primary> + </indexterm> + </term> + <listitem> + <para> + When set, the primary will never let asynchronous standbys fall ahead + of the synchronous standbys. In other words, the asynchronous standbys + will wait until the synchronous replication i.e., after at least the + standbys (either in quorum or priority based) receive the WAL, flush + and acknowledge the primary with the flush LSN. This behvaiour is + particularly important as it avoids extra manual steps required on the + asynchronous standbys which are ahead of the synchronous standbys in + the event of failover. + </para> + <para> + This parameter can only be set in the <filename>postgresql.conf</filename> + file or on the server command line. + </para> + </listitem> + </varlistentry> + </variablelist> </sect2> diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml index 62f2a3332b..52d58d79c4 100644 --- a/doc/src/sgml/monitoring.sgml +++ b/doc/src/sgml/monitoring.sgml @@ -1170,6 +1170,10 @@ postgres 27093 0.0 0.0 30096 2752 ? Ss 11:34 0:00 postgres: ser </thead> <tbody> + <row> + <entry><literal>AsyncWalSenderWaitForSyncReplication</literal></entry> + <entry>Asynchronous standby waiting in WAL sender, before sending WAL, for synchronous standbys to flush the WAL.</entry> + </row> <row> <entry><literal>ClientRead</literal></entry> <entry>Waiting to read data from the client.</entry> diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 655760fee3..79920f0212 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -124,6 +124,8 @@ int wal_sender_timeout = 60 * 1000; /* maximum time to send one WAL * data message */ bool log_replication_commands = false; +bool async_standbys_wait_for_sync_replication = true; + /* * State for WalSndWakeupRequest */ @@ -257,6 +259,9 @@ static bool TransactionIdInRecentPast(TransactionId xid, uint32 epoch); static void WalSndSegmentOpen(XLogReaderState *state, XLogSegNo nextSegNo, TimeLineID *tli_p); +static bool AmAsyncStandbyWALSender(void); +static bool ShouldWaitForSyncRepl(void); +static void WaitForSyncRepl(XLogRecPtr sendRqstPtr); /* Initialize walsender process before entering the main command loop */ void @@ -2836,6 +2841,8 @@ XLogSendPhysical(void) return; } + WaitForSyncRepl(SendRqstPtr); + /* * Figure out how much to send in one message. If there's no more than * MAX_SEND_SIZE bytes to send, send everything. Otherwise send @@ -2994,6 +3001,22 @@ XLogSendLogical(void) if (record != NULL) { + /* + * At this point, we do not know whether the current LSN (ReadRecPtr) + * is required by any of the logical decoding output plugins which is + * only known at the plugin level. If we were to decide whether to wait + * or not for the synchronous standbys flush LSN at the plugin level, + * we might have to pass extra information to it which doesn't sound an + * elegant way. + * + * Another way the output plugins can wait there before sending the WAL + * is by reading the flush LSN from the logical replication slots. + * + * Waiting here i.e. before even the logical decoding kicks in, makes + * the code clean. + */ + WaitForSyncRepl(logical_decoding_ctx->reader->ReadRecPtr); + /* * Note the lack of any call to LagTrackerWrite() which is handled by * WalSndUpdateProgress which is called by output plugin through @@ -3816,3 +3839,135 @@ LagTrackerRead(int head, XLogRecPtr lsn, TimestampTz now) Assert(time != 0); return now - time; } + +/* + * Check if the WAL sender is serving an asynchronous standby at the moment. + */ +static bool +AmAsyncStandbyWALSender(void) +{ + int priority; + + SpinLockAcquire(&MyWalSnd->mutex); + priority = MyWalSnd->sync_standby_priority; + SpinLockRelease(&MyWalSnd->mutex); + + Assert(priority >= 0); + + if (priority > 0) + ereport(DEBUG3, + (errmsg("WAL sender is serving a sync standby at the moment with priority %d", + priority))); + else if (priority == 0) + ereport(DEBUG3, + (errmsg("WAL sender is serving an async standby at the moment with priority %d", + priority))); + + return priority == 0 ? true : false; +} + +/* + * Check if the WAL sender should wait for synchronous replication. + */ +static bool +ShouldWaitForSyncRepl(void) +{ + /* GUC is set to off, so no need to wait */ + if (!async_standbys_wait_for_sync_replication) + return false; + + /* Synchronous replication is not requested, so no need to wait */ + if (SyncRepStandbyNames == NULL || + SyncRepStandbyNames[0] == '\0' || + SyncRepConfig == NULL) + return false; + + /* + * WAL sender is serving a sync standby at the moment, so no need to wait. + */ + if (!AmAsyncStandbyWALSender()) + return false; + + /* + * WAL sender is serving an async standby at the moment, so it should wait. + */ + return true; +} + +/* + * Wait until the sendRqstPtr is at least the flush LSN of all synchronous + * standbys. + */ +static void +WaitForSyncRepl(XLogRecPtr sendRqstPtr) +{ + XLogRecPtr flushLSN; + volatile WalSndCtlData *walsndctl = WalSndCtl; + + Assert(walsndctl != NULL); + Assert(!XLogRecPtrIsInvalid(sendRqstPtr)); + + if (!ShouldWaitForSyncRepl()) + return; + + LWLockAcquire(SyncRepLock, LW_SHARED); + flushLSN = walsndctl->lsn[SYNC_REP_WAIT_FLUSH]; + LWLockRelease(SyncRepLock); + + /* + * This WAL sender is allowed to send the WAL as its request LSN is equal + * or behind the sync standbys flush LSN. + */ + if (sendRqstPtr <= flushLSN) + { + ereport(DEBUG3, + (errmsg("async standby WAL sender is allowed to send WAL as its request LSN %X/%X is equal or behind sync standbys flush LSN %X/%X", + LSN_FORMAT_ARGS(sendRqstPtr), LSN_FORMAT_ARGS(flushLSN)), + errhidestmt(true))); + + return; + } + + ereport(DEBUG3, + (errmsg("async standby WAL sender with request LSN %X/%X is waiting as sync standbys are ahead with flush LSN %X/%X", + LSN_FORMAT_ARGS(flushLSN), LSN_FORMAT_ARGS(sendRqstPtr)), + errhidestmt(true))); + + for (;;) + { + /* + * It is enough to wait until the flush LSN of sync standbys as it + * guarantees local durable commit, standby durable commit after both + * server and OS crash, thus no transaction losses. + * + * XXX: we could go further to make this optional with + * synchronous_commit setting. That is, the async standbys can be made + * to wait either for remote write or flush or apply LSN. But waiting + * for remote flush LSN suffices to give maximum protection for us. + */ + LWLockAcquire(SyncRepLock, LW_SHARED); + flushLSN = walsndctl->lsn[SYNC_REP_WAIT_FLUSH]; + LWLockRelease(SyncRepLock); + + if (sendRqstPtr <= flushLSN) + break; + + /* + * Wait for a brief moment i.e. 10 msec before trying again. + * Intentionally chose a shorter wait time as we don't want to wait + * longer durations and be aggressive in reading the flush LSN. + */ + WalSndWait(WL_SOCKET_WRITEABLE | WL_SOCKET_READABLE, 10L, + WAIT_EVENT_ASYNC_WAL_SENDER_WAIT_FOR_SYNC_REPLICATION); + + /* Clear any already-pending wakeups */ + ResetLatch(MyLatch); + + CHECK_FOR_INTERRUPTS(); + } + + ereport(DEBUG3, + (errmsg("async standby WAL sender with request LSN %X/%X can now send WAL up to sync standbys flush LSN %X/%X", + LSN_FORMAT_ARGS(sendRqstPtr), LSN_FORMAT_ARGS(flushLSN)), + errhidestmt(true))); +} diff --git a/src/backend/utils/activity/wait_event.c b/src/backend/utils/activity/wait_event.c index 60972c3a75..745c3976a8 100644 --- a/src/backend/utils/activity/wait_event.c +++ b/src/backend/utils/activity/wait_event.c @@ -267,6 +267,9 @@ pgstat_get_wait_client(WaitEventClient w) switch (w) { + case WAIT_EVENT_ASYNC_WAL_SENDER_WAIT_FOR_SYNC_REPLICATION: + event_name = "AsyncWalSenderWaitForSyncReplication"; + break; case WAIT_EVENT_CLIENT_READ: event_name = "ClientRead"; break; diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c index e2fe219aa8..a4050ba8d5 100644 --- a/src/backend/utils/misc/guc.c +++ b/src/backend/utils/misc/guc.c @@ -1381,6 +1381,15 @@ static struct config_bool ConfigureNamesBool[] = false, NULL, NULL, NULL }, + { + {"async_standbys_wait_for_sync_replication", PGC_SIGHUP, REPLICATION_SENDING, + gettext_noop("Sets whether asynchronous standbys should wait until synchronous standbys receive and flush WAL."), + NULL + }, + &async_standbys_wait_for_sync_replication, + true, + NULL, NULL, NULL + }, { {"debug_assertions", PGC_INTERNAL, PRESET_OPTIONS, gettext_noop("Shows whether the running server has assertion checks enabled."), diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index 56d0bee6d9..f9de74d4e8 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -308,6 +308,9 @@ #wal_sender_timeout = 60s # in milliseconds; 0 disables #track_commit_timestamp = off # collect timestamp of transaction commit # (change requires restart) +#async_standbys_wait_for_sync_replication = on # asynchronous standbys wait + # until synchronous standbys receive, flush WAL and + # confirm primary # - Primary Server - diff --git a/src/include/replication/walsender.h b/src/include/replication/walsender.h index b1892e9e4b..61b113ae48 100644 --- a/src/include/replication/walsender.h +++ b/src/include/replication/walsender.h @@ -34,6 +34,7 @@ extern bool wake_wal_senders; extern int max_wal_senders; extern int wal_sender_timeout; extern bool log_replication_commands; +extern bool async_standbys_wait_for_sync_replication; extern void InitWalSender(void); extern bool exec_replication_command(const char *query_string); diff --git a/src/include/utils/wait_event.h b/src/include/utils/wait_event.h index 395d325c5f..fbe2b66368 100644 --- a/src/include/utils/wait_event.h +++ b/src/include/utils/wait_event.h @@ -60,7 +60,8 @@ typedef enum */ typedef enum { - WAIT_EVENT_CLIENT_READ = PG_WAIT_CLIENT, + WAIT_EVENT_ASYNC_WAL_SENDER_WAIT_FOR_SYNC_REPLICATION = PG_WAIT_CLIENT, + WAIT_EVENT_CLIENT_READ, WAIT_EVENT_CLIENT_WRITE, WAIT_EVENT_GSS_OPEN_SERVER, WAIT_EVENT_LIBPQWALRECEIVER_CONNECT, -- 2.25.1 ^ permalink raw reply [nested|flat] 4+ messages in thread
* Re: Allow async standbys wait for sync replication (was: Disallow quorum uncommitted (with synchronous standbys) txns in logical replication subscribers) @ 2022-02-25 19:38 Nathan Bossart <[email protected]> parent: Bharath Rupireddy <[email protected]> 0 siblings, 1 reply; 4+ messages in thread From: Nathan Bossart @ 2022-02-25 19:38 UTC (permalink / raw) To: Bharath Rupireddy <[email protected]>; +Cc: SATYANARAYANA NARLAPURAM <[email protected]>; PostgreSQL Hackers <[email protected]> On Fri, Feb 25, 2022 at 08:31:37PM +0530, Bharath Rupireddy wrote: > Thanks Satya and others for the inputs. Here's the v1 patch that > basically allows async wal senders to wait until the sync standbys > report their flush lsn back to the primary. Please let me know your > thoughts. I haven't had a chance to look too closely yet, but IIUC this adds a new function that waits for synchronous replication. This new function essentially spins until the synchronous LSN has advanced. I don't think it's a good idea to block sending any WAL like this. AFAICT it is possible that there will be a lot of synchronously replicated WAL that we can send, and it might just be the last several bytes that cannot yet be replicated to the asynchronous standbys. І believe this patch will cause the server to avoid sending _any_ WAL until the synchronous LSN advances. Perhaps we should instead just choose the SendRqstPtr based on the current synchronous LSN. Presumably there are other things we'd need to consider, but in general, I think we ought to send as much WAL as possible for a given call to XLogSendPhysical(). > I've done pgbench testing to see if the patch causes any problems. I > ran tests two times, there isn't much difference in the txns per > seconds (tps), although there's a delay in the async standby receiving > the WAL, after all, that's the feature we are pursuing. I'm curious what a longer pgbench run looks like when the synchronous replicas are in the same region. That is probably a more realistic use-case. -- Nathan Bossart Amazon Web Services: https://aws.amazon.com ^ permalink raw reply [nested|flat] 4+ messages in thread
* Re: Allow async standbys wait for sync replication (was: Disallow quorum uncommitted (with synchronous standbys) txns in logical replication subscribers) @ 2022-02-25 21:52 Hsu, John <[email protected]> parent: Nathan Bossart <[email protected]> 0 siblings, 0 replies; 4+ messages in thread From: Hsu, John @ 2022-02-25 21:52 UTC (permalink / raw) To: Nathan Bossart <[email protected]>; Bharath Rupireddy <[email protected]>; +Cc: SATYANARAYANA NARLAPURAM <[email protected]>; PostgreSQL Hackers <[email protected]> Hello, On 2/25/22 11:38 AM, Nathan Bossart wrote: > CAUTION: This email originated from outside of the organization. Do not click links or open attachments unless you can confirm the sender and know the content is safe. > > > > On Fri, Feb 25, 2022 at 08:31:37PM +0530, Bharath Rupireddy wrote: >> Thanks Satya and others for the inputs. Here's the v1 patch that >> basically allows async wal senders to wait until the sync standbys >> report their flush lsn back to the primary. Please let me know your >> thoughts. > I haven't had a chance to look too closely yet, but IIUC this adds a new > function that waits for synchronous replication. This new function > essentially spins until the synchronous LSN has advanced. > > I don't think it's a good idea to block sending any WAL like this. AFAICT > it is possible that there will be a lot of synchronously replicated WAL > that we can send, and it might just be the last several bytes that cannot > yet be replicated to the asynchronous standbys. І believe this patch will > cause the server to avoid sending _any_ WAL until the synchronous LSN > advances. > > Perhaps we should instead just choose the SendRqstPtr based on the current > synchronous LSN. Presumably there are other things we'd need to consider, > but in general, I think we ought to send as much WAL as possible for a > given call to XLogSendPhysical(). I think you're right that we'll avoid sending any WAL until sync_lsn advances. We could setup a contrived situation where the async-walsender never advances because it terminates before the flush_lsn of the synchronous_node catches up. And when the async-walsender restarts, it'll start with the latest flushed on the primary and we could go into a perpetual loop. I took a look at the patch and tested basic streaming with async replicas ahead of the synchronous standby and with logical clients as well and it works as expected. > > ereport(LOG, > (errmsg("async standby WAL sender with request LSN %X/%X is waiting as sync standbys are ahead with flush LSN %X/%X", > LSN_FORMAT_ARGS(flushLSN), LSN_FORMAT_ARGS(sendRqstPtr)), > errhidestmt(true))); I think this log formatting is incorrect. s/sync standbys are ahead/sync standbys are behind/ and I think you need to swap flushLsn and sendRqstPtr When a walsender is waiting for the lsn on the synchronous replica to advance and a database stop is issued to the writer, the pg_ctl stop isn't able to proceed and the database seems to never shutdown. > Assert(priority >= 0); What's the point of the assert here? Also the comments/code refer to AsyncStandbys, however it's also used for logical clients, which may or may not be standbys. Don't feel too strongly about the naming here but something to note. > if (!ShouldWaitForSyncRepl()) > return; > ... > for (;;) > { > // rest of work > } If we had a walsender already waiting for an ack, and the conditions of ShouldWaitForSyncRepl() change, such as disabling async_standbys_wait_for_sync_replication or synchronous replication it'll still wait since we never re-check the condition. postgres=# select wait_event from pg_stat_activity where wait_event like 'AsyncWal%'; wait_event -------------------------------------- AsyncWalSenderWaitForSyncReplication AsyncWalSenderWaitForSyncReplication AsyncWalSenderWaitForSyncReplication (3 rows) postgres=# show synchronous_standby_names; synchronous_standby_names --------------------------- (1 row) postgres=# show async_standbys_wait_for_sync_replication; async_standbys_wait_for_sync_replication ------------------------------------------ off (1 row) > LWLockAcquire(SyncRepLock, LW_SHARED); > flushLSN = walsndctl->lsn[SYNC_REP_WAIT_FLUSH]; > LWLockRelease(SyncRepLock); Should we configure this similar to the user's setting of synchronous_commit instead of just flush? (SYNC_REP_WAIT_WRITE, SYNC_REP_WAIT_APPLY) Thanks, John H ^ permalink raw reply [nested|flat] 4+ messages in thread
end of thread, other threads:[~2022-02-25 21:52 UTC | newest] Thread overview: 4+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2020-04-02 00:56 [PATCH 3/8] Move processing of NULLs from BRIN support functions Tomas Vondra <[email protected]> 2022-02-25 15:01 Allow async standbys wait for sync replication (was: Disallow quorum uncommitted (with synchronous standbys) txns in logical replication subscribers) Bharath Rupireddy <[email protected]> 2022-02-25 19:38 ` Re: Allow async standbys wait for sync replication (was: Disallow quorum uncommitted (with synchronous standbys) txns in logical replication subscribers) Nathan Bossart <[email protected]> 2022-02-25 21:52 ` Re: Allow async standbys wait for sync replication (was: Disallow quorum uncommitted (with synchronous standbys) txns in logical replication subscribers) Hsu, John <[email protected]>
This inbox is served by agora; see mirroring instructions for how to clone and mirror all data and code used for this inbox