public inbox for [email protected]help / color / mirror / Atom feed
[PATCH 2/8] Move IS [NOT] NULL handling from BRIN support functions 3+ messages / 3 participants [nested] [flat]
* [PATCH 2/8] Move IS [NOT] NULL handling from BRIN support functions @ 2020-09-17 15:26 Tomas Vondra <[email protected]> 0 siblings, 0 replies; 3+ messages in thread From: Tomas Vondra @ 2020-09-17 15:26 UTC (permalink / raw) The handling of IS [NOT] NULL clauses is independent of an opclass, and most of the code was exactly the same in both minmax and inclusion. So instead move the code from support procedures to the AM methods etc. This simplifies the code quite a bit - especially the support procedures quite a bit, as they don't need to care about NULL values and flags at all. It also means the IS [NOT] NULL clauses can be evaluated without invoking the support procedure at all. Author: Tomas Vondra <[email protected]> Author: Nikita Glukhov <[email protected]> Reviewed-by: Alvaro Herrera <[email protected]> Reviewed-by: Mark Dilger <[email protected]> Reviewed-by: Alexander Korotkov <[email protected]> Reviewed-by: Masahiko Sawada <[email protected]> Reviewed-by: John Naylor <[email protected]> Discussion: https://postgr.es/m/[email protected] --- src/backend/access/brin/brin.c | 293 +++++++++++++++++------ src/backend/access/brin/brin_inclusion.c | 106 +------- src/backend/access/brin/brin_minmax.c | 103 +------- src/include/access/brin_internal.h | 3 + 4 files changed, 239 insertions(+), 266 deletions(-) diff --git a/src/backend/access/brin/brin.c b/src/backend/access/brin/brin.c index dc187153aa..14da9ed17f 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 @@ -179,7 +182,6 @@ brininsert(Relation idxRel, Datum *values, bool *nulls, OffsetNumber off; BrinTuple *brtup; BrinMemTuple *dtup; - int keyno; CHECK_FOR_INTERRUPTS(); @@ -243,31 +245,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) { @@ -390,8 +368,10 @@ bringetbitmap(IndexScanDesc scan, TIDBitmap *tbm) BrinMemTuple *dtup; BrinTuple *btup = NULL; Size btupsz = 0; - ScanKey **keys; - int *nkeys; + ScanKey **keys, + **nullkeys; + int *nkeys, + *nnullkeys; int keyno; opaque = (BrinOpaque *) scan->opaque; @@ -416,10 +396,13 @@ bringetbitmap(IndexScanDesc scan, TIDBitmap *tbm) /* * Make room for per-attribute lists of scan keys that we'll pass to the - * consistent support procedure. + * consistent support procedure. We keep null and regular keys separate, + * so that we can easily pass regular keys to the consistent function. */ keys = palloc0(sizeof(ScanKey *) * bdesc->bd_tupdesc->natts); + nullkeys = palloc0(sizeof(ScanKey *) * bdesc->bd_tupdesc->natts); nkeys = palloc0(sizeof(int) * bdesc->bd_tupdesc->natts); + nnullkeys = palloc0(sizeof(int) * bdesc->bd_tupdesc->natts); /* * Preprocess the scan keys - split them into per-attribute arrays. @@ -440,23 +423,24 @@ bringetbitmap(IndexScanDesc scan, TIDBitmap *tbm) TupleDescAttr(bdesc->bd_tupdesc, keyattno - 1)->attcollation)); - /* First time we see this index attribute, so init as needed. */ - if (!keys[keyattno-1]) + /* + * First time we see this index attribute, so init as needed. + * + * This is a bit of an overkill - we don't know how many scan + * keys are there for a given attribute, so we simply allocate + * the largest number possible (as if all scan keys belonged to + * the same attribute). This may waste a bit of memory, but we + * only expect small number of scan keys in general, so this + * should be negligible, and it's probably cheaper than having + * to repalloc repeatedly. + */ + if (consistentFn[keyattno - 1].fn_oid == InvalidOid) { FmgrInfo *tmp; - /* - * This is a bit of an overkill - we don't know how many - * scan keys are there for this attribute, so we simply - * allocate the largest number possible. This may waste - * a bit of memory, but we only expect small number of - * scan keys in general, so this should be negligible, - * and it's cheaper than having to repalloc repeatedly. - */ - keys[keyattno - 1] = palloc0(sizeof(ScanKey) * scan->numberOfKeys); - - /* First time this column, so look up consistent function */ - Assert(consistentFn[keyattno - 1].fn_oid == InvalidOid); + /* No key/null arrays for this attribute. */ + Assert((keys[keyattno - 1] == NULL) && (nkeys[keyattno - 1] == 0)); + Assert((nullkeys[keyattno - 1] == NULL) && (nnullkeys[keyattno - 1] == 0)); tmp = index_getprocinfo(idxRel, keyattno, BRIN_PROCNUM_CONSISTENT); @@ -464,9 +448,23 @@ bringetbitmap(IndexScanDesc scan, TIDBitmap *tbm) CurrentMemoryContext); } - /* Add key to the per-attribute array. */ - keys[keyattno - 1][nkeys[keyattno - 1]] = key; - nkeys[keyattno - 1]++; + /* Add key to the proper per-attribute array. */ + if (key->sk_flags & SK_ISNULL) + { + if (!nullkeys[keyattno - 1]) + nullkeys[keyattno - 1] = palloc0(sizeof(ScanKey) * scan->numberOfKeys); + + nullkeys[keyattno - 1][nnullkeys[keyattno - 1]] = key; + nnullkeys[keyattno - 1]++; + } + else + { + if (!keys[keyattno - 1]) + keys[keyattno - 1] = palloc0(sizeof(ScanKey) * scan->numberOfKeys); + + keys[keyattno - 1][nkeys[keyattno - 1]] = key; + nkeys[keyattno - 1]++; + } } /* allocate an initial in-memory tuple, out of the per-range memcxt */ @@ -544,15 +542,57 @@ bringetbitmap(IndexScanDesc scan, TIDBitmap *tbm) BrinValues *bval; Datum add; - /* skip attributes without any san keys */ - if (nkeys[attno - 1] == 0) + /* + * skip attributes without any scan keys (both regular + * and IS [NOT] NULL) + */ + if (nkeys[attno - 1] == 0 && nnullkeys[attno - 1] == 0) continue; bval = &dtup->bt_columns[attno - 1]; + /* + * 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. + */ + if (bdesc->bd_info[attno - 1]->oi_regular_nulls && + !check_null_keys(bval, nullkeys[attno - 1], + nnullkeys[attno - 1])) + { + /* + * 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). + */ + + /* No regular scan keys - page range as a whole passes. */ + if (!nkeys[attno - 1]) + continue; + Assert((nkeys[attno - 1] > 0) && (nkeys[attno - 1] <= scan->numberOfKeys)); + /* If it is all nulls, it cannot possibly be consistent. */ + if (bval->bv_allnulls) + { + addrange = false; + break; + } + /* * Check whether the scan key is consistent with the page * range values; if so, have the pages in the range added @@ -695,7 +735,6 @@ brinbuildCallback(Relation index, { BrinBuildState *state = (BrinBuildState *) brstate; BlockNumber thisblock; - int i; thisblock = ItemPointerGetBlockNumber(tid); @@ -724,25 +763,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); } /* @@ -1521,6 +1543,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); @@ -1574,3 +1629,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 215bc794d3..f4730be3b9 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); @@ -264,63 +254,6 @@ brin_inclusion_consistent(PG_FUNCTION_ARGS) int nkeys = PG_GETARG_INT32(3); Oid colloid = PG_GET_COLLATION(); int keyno; - bool regular_keys = false; - - /* - * First check if there are any IS NULL scan keys, and if we're - * violating them. In that case we can terminate early, without - * inspecting the ranges. - */ - for (keyno = 0; keyno < nkeys; keyno++) - { - ScanKey key = keys[keyno]; - - Assert(key->sk_attno == column->bv_attno); - - /* handle IS NULL/IS NOT NULL tests */ - if (key->sk_flags & SK_ISNULL) - { - if (key->sk_flags & SK_SEARCHNULL) - { - if (column->bv_allnulls || column->bv_hasnulls) - continue; /* this key is fine, continue */ - - PG_RETURN_BOOL(false); - } - - /* - * For IS NOT NULL, we can only skip ranges that are known to have - * only nulls. - */ - if (key->sk_flags & SK_SEARCHNOTNULL) - { - if (column->bv_allnulls) - PG_RETURN_BOOL(false); - - continue; - } - - /* - * Neither IS NULL nor IS NOT NULL was used; assume all indexable - * operators are strict and return false. - */ - PG_RETURN_BOOL(false); - } - else - /* note we have regular (non-NULL) scan keys */ - regular_keys = true; - } - - /* - * If the page range is all nulls, it cannot possibly be consistent if - * there are some regular scan keys. - */ - if (column->bv_allnulls && regular_keys) - PG_RETURN_BOOL(false); - - /* If there are no regular keys, the page range is considered consistent. */ - if (!regular_keys) - PG_RETURN_BOOL(true); /* It has to be checked, if it contains elements that are not mergeable. */ if (DatumGetBool(column->bv_values[INCLUSION_UNMERGEABLE])) @@ -331,9 +264,8 @@ brin_inclusion_consistent(PG_FUNCTION_ARGS) { ScanKey key = keys[keyno]; - /* ignore IS NULL/IS NOT NULL tests handled above */ - if (key->sk_flags & SK_ISNULL) - continue; + /* NULL keys are handled and filtered-out in bringetbitmap */ + Assert(!(key->sk_flags & SK_ISNULL)); /* * When there are multiple scan keys, failure to meet the @@ -572,37 +504,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 12878ff3a0..6c8852d404 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); @@ -152,72 +142,14 @@ brin_minmax_consistent(PG_FUNCTION_ARGS) int nkeys = PG_GETARG_INT32(3); Oid colloid = PG_GET_COLLATION(); int keyno; - bool regular_keys = false; - - /* - * First check if there are any IS NULL scan keys, and if we're - * violating them. In that case we can terminate early, without - * inspecting the ranges. - */ - for (keyno = 0; keyno < nkeys; keyno++) - { - ScanKey key = keys[keyno]; - - Assert(key->sk_attno == column->bv_attno); - - /* handle IS NULL/IS NOT NULL tests */ - if (key->sk_flags & SK_ISNULL) - { - if (key->sk_flags & SK_SEARCHNULL) - { - if (column->bv_allnulls || column->bv_hasnulls) - continue; /* this key is fine, continue */ - - PG_RETURN_BOOL(false); - } - - /* - * For IS NOT NULL, we can only skip ranges that are known to have - * only nulls. - */ - if (key->sk_flags & SK_SEARCHNOTNULL) - { - if (column->bv_allnulls) - PG_RETURN_BOOL(false); - - continue; - } - - /* - * Neither IS NULL nor IS NOT NULL was used; assume all indexable - * operators are strict and return false. - */ - PG_RETURN_BOOL(false); - } - else - /* note we have regular (non-NULL) scan keys */ - regular_keys = true; - } - - /* - * If the page range is all nulls, it cannot possibly be consistent if - * there are some regular scan keys. - */ - if (column->bv_allnulls && regular_keys) - PG_RETURN_BOOL(false); - - /* If there are no regular keys, the page range is considered consistent. */ - if (!regular_keys) - PG_RETURN_BOOL(true); /* Check that the range is consistent with all scan keys. */ for (keyno = 0; keyno < nkeys; keyno++) { ScanKey key = keys[keyno]; - /* ignore IS NULL/IS NOT NULL tests handled above */ - if (key->sk_flags & SK_ISNULL) - continue; + /* NULL keys are handled and filtered-out in bringetbitmap */ + Assert(!(key->sk_flags & SK_ISNULL)); /* * When there are multiple scan keys, failure to meet the @@ -308,34 +240,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 78c89a6961..79440ebe7b 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.26.2 --------------FA974B75A0C3E9CA18CE7207 Content-Type: text/x-patch; charset=UTF-8; name="0003-Optimize-allocations-in-bringetbitmap-20210122.patch" Content-Transfer-Encoding: 7bit Content-Disposition: attachment; filename*0="0003-Optimize-allocations-in-bringetbitmap-20210122.patch" ^ permalink raw reply [nested|flat] 3+ messages in thread
* Re: Support load balancing in libpq @ 2023-01-06 17:21 Michael Banck <[email protected]> 0 siblings, 1 reply; 3+ messages in thread From: Michael Banck @ 2023-01-06 17:21 UTC (permalink / raw) To: Jelte Fennema <[email protected]>; +Cc: Maxim Orlov <[email protected]>; Aleksander Alekseev <[email protected]>; pgsql-hackers Hi, On Tue, Nov 29, 2022 at 02:57:08PM +0000, Jelte Fennema wrote: > Attached is a new version with the tests cleaned up a bit (more > comments mostly). > > @Michael, did you have a chance to look at the last version? Because I > feel that the patch is pretty much ready for a committer to look at, > at this point. I had another look; it still applies and tests pass. I still find the distribution over three hosts a bit skewed (even when OpenSSL is enabled, which wasn't the case when I first tested it), but couldn't find a general fault and it worked well enough in my testing. I wonder whether making the parameter a boolean will paint us into a corner, and whether maybe additional modes could be envisioned in the future, but I can't think of some right now (you can pretty neatly restrict load-balancing to standbys by setting target_session_attrs=standby in addition). Maybe a way to change the behaviour if a dns hostname is backed by multiple entries? Some further (mostly nitpicking) comments on the patch: > From 6e20bb223012b666161521b5e7249c066467a5f3 Mon Sep 17 00:00:00 2001 > From: Jelte Fennema <[email protected]> > Date: Mon, 12 Sep 2022 09:44:06 +0200 > Subject: [PATCH v5] Support load balancing in libpq > > Load balancing connections across multiple read replicas is a pretty > common way of scaling out read queries. There are two main ways of doing > so, both with their own advantages and disadvantages: > 1. Load balancing at the client level > 2. Load balancing by connecting to an intermediary load balancer > > Both JBDC (Java) and Npgsql (C#) already support client level load > balancing (option #1). This patch implements client level load balancing > for libpq as well. To stay consistent with the JDBC and Npgsql part of > the ecosystem, a similar implementation and name for the option are > used. I still think all of the above has no business in the commit message, though maybe the first paragraph can stay as introduction. > It contains two levels of load balancing: > 1. The given hosts are randomly shuffled, before resolving them > one-by-one. > 2. Once a host its addresses get resolved, those addresses are shuffled, > before trying to connect to them one-by-one. That's fine. What should be in the commit message is at least a mention of what the new connection parameter is called and possibly what is done to accomplish it. But the committer will pick this up if needed I guess. > diff --git a/doc/src/sgml/libpq.sgml b/doc/src/sgml/libpq.sgml > index f9558dec3b..6ce7a0c9cc 100644 > --- a/doc/src/sgml/libpq.sgml > +++ b/doc/src/sgml/libpq.sgml > @@ -1316,6 +1316,54 @@ postgresql://%2Fvar%2Flib%2Fpostgresql/dbname > </listitem> > </varlistentry> > > + <varlistentry id="libpq-load-balance-hosts" xreflabel="load_balance_hosts"> > + <term><literal>load_balance_hosts</literal></term> > + <listitem> > + <para> > + Controls whether the client load balances connections across hosts and > + adresses. The default value is 0, meaning off, this means that hosts are > + tried in order they are provided and addresses are tried in the order > + they are received from DNS or a hosts file. If this value is set to 1, > + meaning on, the hosts and addresses that they resolve to are tried in > + random order. Subsequent queries once connected will still be sent to > + the same server. Setting this to 1, is mostly useful when opening > + multiple connections at the same time, possibly from different machines. > + This way connections can be load balanced across multiple Postgres > + servers. > + </para> > + <para> > + When providing multiple hosts, these hosts are resolved in random order. > + Then if that host resolves to multiple addresses, these addresses are > + connected to in random order. Only once all addresses for a single host > + have been tried, the addresses for the next random host will be > + resolved. This behaviour can lead to non-uniform address selection in > + certain cases. Such as when not all hosts resolve to the same number of > + addresses, or when multiple hosts resolve to the same address. So if you > + want uniform load balancing, this is something to keep in mind. However, > + non-uniform load balancing also has usecases, e.g. providing the > + hostname of a larger server multiple times in the host string so it gets > + more requests. > + </para> > + <para> > + When using this setting it's recommended to also configure a reasonable > + value for <xref linkend="libpq-connect-connect-timeout"/>. Because then, > + if one of the nodes that are used for load balancing is not responding, > + a new node will be tried. > + </para> > + </listitem> > + </varlistentry> I think this whole section is generally fine, but needs some copyediting. > + <varlistentry id="libpq-random-seed" xreflabel="random_seed"> > + <term><literal>random_seed</literal></term> > + <listitem> > + <para> > + Sets the random seed that is used by <xref linkend="libpq-load-balance-hosts"/> > + to randomize the host order. This option is mostly useful when running > + tests that require a stable random order. > + </para> > + </listitem> > + </varlistentry> I wonder whether this needs to be documented if it is mostly a development/testing parameter? > diff --git a/src/include/libpq/pqcomm.h b/src/include/libpq/pqcomm.h > index fcf68df39b..39e93b1392 100644 > --- a/src/include/libpq/pqcomm.h > +++ b/src/include/libpq/pqcomm.h > @@ -27,6 +27,12 @@ typedef struct > socklen_t salen; > } SockAddr; > > +typedef struct > +{ > + int family; > + SockAddr addr; > +} AddrInfo; That last line looks weirdly indented compared to SockAddr; in the struct above. > /* Configure the UNIX socket location for the well known port. */ > > #define UNIXSOCK_PATH(path, port, sockdir) \ > diff --git a/src/interfaces/libpq/fe-connect.c b/src/interfaces/libpq/fe-connect.c > index f88d672c6c..b4d3613713 100644 > --- a/src/interfaces/libpq/fe-connect.c > +++ b/src/interfaces/libpq/fe-connect.c > @@ -241,6 +241,14 @@ static const internalPQconninfoOption PQconninfoOptions[] = { > "Fallback-Application-Name", "", 64, > offsetof(struct pg_conn, fbappname)}, > > + {"load_balance_hosts", NULL, NULL, NULL, > + "Load-Balance", "", 1, /* should be just '0' or '1' */ > + offsetof(struct pg_conn, loadbalance)}, > + > + {"random_seed", NULL, NULL, NULL, > + "Random-Seed", "", 10, /* strlen(INT32_MAX) == 10 */ > + offsetof(struct pg_conn, randomseed)}, > + > {"keepalives", NULL, NULL, NULL, > "TCP-Keepalives", "", 1, /* should be just '0' or '1' */ > offsetof(struct pg_conn, keepalives)}, > @@ -379,6 +387,7 @@ static bool fillPGconn(PGconn *conn, PQconninfoOption *connOptions); > static void freePGconn(PGconn *conn); > static void closePGconn(PGconn *conn); > static void release_conn_addrinfo(PGconn *conn); > +static bool store_conn_addrinfo(PGconn *conn, struct addrinfo *addrlist); > static void sendTerminateConn(PGconn *conn); > static PQconninfoOption *conninfo_init(PQExpBuffer errorMessage); > static PQconninfoOption *parse_connection_string(const char *connstr, > @@ -424,6 +433,9 @@ static void pgpassfileWarning(PGconn *conn); > static void default_threadlock(int acquire); > static bool sslVerifyProtocolVersion(const char *version); > static bool sslVerifyProtocolRange(const char *min, const char *max); > +static int loadBalance(PGconn *conn); > +static bool parse_int_param(const char *value, int *result, PGconn *conn, > + const char *context); > > > /* global variable because fe-auth.c needs to access it */ > @@ -1007,6 +1019,46 @@ parse_comma_separated_list(char **startptr, bool *more) > return p; > } > > +/* > + * Initializes the prng_state field of the connection. We want something > + * unpredictable, so if possible, use high-quality random bits for the > + * seed. Otherwise, fall back to a seed based on timestamp and PID. > + */ > +static bool > +libpq_prng_init(PGconn *conn) > +{ > + if (unlikely(conn->randomseed)) > + { > + int rseed; > + > + if (!parse_int_param(conn->randomseed, &rseed, conn, "random_seed")) > + { > + return false; > + }; I think it's project policy to drop the braces for single statements in if blocks. > + pg_prng_seed(&conn->prng_state, rseed); > + } > + else if (unlikely(!pg_prng_strong_seed(&conn->prng_state))) > + { > + uint64 rseed; > + time_t now = time(NULL); > + > + /* > + * Since PIDs and timestamps tend to change more frequently in their > + * least significant bits, shift the timestamp left to allow a larger > + * total number of seeds in a given time period. Since that would > + * leave only 20 bits of the timestamp that cycle every ~1 second, > + * also mix in some higher bits. > + */ > + rseed = ((uint64) getpid()) ^ > + ((uint64) now << 12) ^ > + ((uint64) now >> 20); > + > + pg_prng_seed(&conn->prng_state, rseed); > + } > + return true; > +} > + > + > /* Additional newline. > @@ -1164,6 +1217,36 @@ connectOptions2(PGconn *conn) > } > } > > + if (loadbalancehosts < 0) > + { > + appendPQExpBufferStr(&conn->errorMessage, > + libpq_gettext("loadbalance parameter must be an integer\n")); > + return false; > + } > + > + if (loadbalancehosts) > + { > + if (!libpq_prng_init(conn)) > + { > + return false; > + } > + > + /* > + * Shuffle connhost with a Durstenfeld/Knuth version of the > + * Fisher-Yates shuffle. Source: > + * https://en.wikipedia.org/wiki/Fisher%E2%80%93Yates_shuffle#The_modern_algorithm > + */ > + for (i = conn->nconnhost - 1; i > 0; i--) > + { > + int j = pg_prng_uint64_range(&conn->prng_state, 0, i); > + pg_conn_host temp = conn->connhost[j]; > + > + conn->connhost[j] = conn->connhost[i]; > + conn->connhost[i] = temp; > + } > + } > + > + > /* Additional newline. > @@ -1726,6 +1809,27 @@ connectFailureMessage(PGconn *conn, int errorno) > libpq_append_conn_error(conn, "\tIs the server running on that host and accepting TCP/IP connections?"); > } > > +/* > + * Should we load balance across hosts? Returns 1 if yes, 0 if no, and -1 if > + * conn->loadbalance is set to a value which is not parseable as an integer. > + */ > +static int > +loadBalance(PGconn *conn) > +{ > + char *ep; > + int val; > + > + if (conn->loadbalance == NULL) > + { > + return 0; > + } Another case of additional braces. > + val = strtol(conn->loadbalance, &ep, 10); > + if (*ep) > + return -1; > + return val != 0 ? 1 : 0; > +} > + > + > /* Additional newline. > @@ -4041,6 +4154,63 @@ freePGconn(PGconn *conn) > free(conn); > } > > + > +/* Additional newline. > + * Copies over the AddrInfos from addrlist to the PGconn. > + */ > +static bool > +store_conn_addrinfo(PGconn *conn, struct addrinfo *addrlist) > +{ > + struct addrinfo *ai = addrlist; > + > + conn->whichaddr = 0; > + > + conn->naddr = 0; > + while (ai) > + { > + ai = ai->ai_next; > + conn->naddr++; > + } > + > + conn->addr = calloc(conn->naddr, sizeof(AddrInfo)); > + if (conn->addr == NULL) > + { > + return false; > + } Additional braces. > @@ -4048,11 +4218,10 @@ freePGconn(PGconn *conn) > static void > release_conn_addrinfo(PGconn *conn) > { > - if (conn->addrlist) > + if (conn->addr) > { > - pg_freeaddrinfo_all(conn->addrlist_family, conn->addrlist); > - conn->addrlist = NULL; > - conn->addr_cur = NULL; /* for safety */ > + free(conn->addr); > + conn->addr = NULL; > } > } > > diff --git a/src/interfaces/libpq/libpq-int.h b/src/interfaces/libpq/libpq-int.h > index 512762f999..76ee988038 100644 > --- a/src/interfaces/libpq/libpq-int.h > +++ b/src/interfaces/libpq/libpq-int.h > @@ -82,6 +82,8 @@ typedef struct > #endif > #endif /* USE_OPENSSL */ > > +#include "common/pg_prng.h" > + > /* > * POSTGRES backend dependent Constants. > */ > @@ -373,6 +375,8 @@ struct pg_conn > char *pgpassfile; /* path to a file containing password(s) */ > char *channel_binding; /* channel binding mode > * (require,prefer,disable) */ > + char *loadbalance; /* load balance over hosts */ > + char *randomseed; /* seed for randomization of load balancing */ > char *keepalives; /* use TCP keepalives? */ A bit unclear why you put the variables at this point in the list, but the list doesn't seem to be ordered strictly anyway; still, maybe they would fit best at the bottom below target_session_attrs? > char *keepalives_idle; /* time between TCP keepalives */ > char *keepalives_interval; /* time between TCP keepalive > @@ -461,8 +465,10 @@ struct pg_conn > PGTargetServerType target_server_type; /* desired session properties */ > bool try_next_addr; /* time to advance to next address/host? */ > bool try_next_host; /* time to advance to next connhost[]? */ > - struct addrinfo *addrlist; /* list of addresses for current connhost */ > - struct addrinfo *addr_cur; /* the one currently being tried */ > + int naddr; /* number of addrs returned by getaddrinfo */ > + int whichaddr; /* the addr currently being tried */ Address(es) is always spelt out in the comments, those two addr(s) should also I think. > diff --git a/src/interfaces/libpq/t/003_loadbalance.pl b/src/interfaces/libpq/t/003_loadbalance.pl > new file mode 100644 > index 0000000000..07eddbe9cc > --- /dev/null > +++ b/src/interfaces/libpq/t/003_loadbalance.pl > @@ -0,0 +1,167 @@ > +# Copyright (c) 2022, PostgreSQL Global Development Group Copyright bump needed. Cheers, Michael ^ permalink raw reply [nested|flat] 3+ messages in thread
* Re: [EXTERNAL] Re: Support load balancing in libpq @ 2023-01-09 13:00 Jelte Fennema <[email protected]> parent: Michael Banck <[email protected]> 0 siblings, 0 replies; 3+ messages in thread From: Jelte Fennema @ 2023-01-09 13:00 UTC (permalink / raw) To: Michael Banck <[email protected]>; +Cc: Maxim Orlov <[email protected]>; Aleksander Alekseev <[email protected]>; pgsql-hackers Attached an updated patch which should address your feedback and I updated the commit message. > I wonder whether making the parameter a boolean will paint us into a > corner I made it a string option, just like target_session_attrs. I'm pretty sure a round-robin load balancing policy could be implemented in the future given certain constraints, like connections being made within the same process. I adjusted the docs accordingly. > > +typedef struct > > +{ > > + int family; > > + SockAddr addr; > > +} AddrInfo; > > That last line looks weirdly indented compared to SockAddr; in the > struct above. Yes I agree, but for some reason pgindent really badly wants it formatted that way. I now undid the changes made by pgindent manually. > I wonder whether this needs to be documented if it is mostly a > development/testing parameter? I also wasn't sure whether it should be documented or not. I'm fine with either, I'll leave it in for now and let a committer decide if it's wanted or not. > A bit unclear why you put the variables at this point in the list, but > the list doesn't seem to be ordered strictly anyway; still, maybe they > would fit best at the bottom below target_session_attrs? Good point, I added them after target_session_attrs now and also moved docs/parsing accordingly. This makes conceptually to me as well, since target_session_attrs and load_balance_hosts have some interesting sense contextually too. P.S. I also attached the same pgindent run patch that I added in https://www.postgresql.org/message-id/flat/[email protected]... Attachments: [application/octet-stream] v6-0001-libpq-Run-pgindent-after-a9e9a9f32b3.patch (39.3K, ../../DBBPR83MB05079A565E6AB609104D3F9DF7FE9@DBBPR83MB0507.EURPRD83.prod.outlook.com/2-v6-0001-libpq-Run-pgindent-after-a9e9a9f32b3.patch) download | inline diff: From beb005477c16af285df7fc068bfc2def70e5acf8 Mon Sep 17 00:00:00 2001 From: Jelte Fennema <[email protected]> Date: Wed, 30 Nov 2022 10:07:19 +0100 Subject: [PATCH v6 1/2] libpq: Run pgindent after a9e9a9f32b3 It seems that pgindent was not run after the error handling refactor in commit a9e9a9f32b35edf129c88e8b929ef223f8511f59. This fixes that and also addresses a few other things pgindent wanted to change in libpq. --- src/interfaces/libpq/fe-auth-scram.c | 2 +- src/interfaces/libpq/fe-auth.c | 8 +- src/interfaces/libpq/fe-connect.c | 124 +++++++++++------------ src/interfaces/libpq/fe-exec.c | 16 +-- src/interfaces/libpq/fe-lobj.c | 42 ++++---- src/interfaces/libpq/fe-misc.c | 10 +- src/interfaces/libpq/fe-protocol3.c | 2 +- src/interfaces/libpq/fe-secure-common.c | 6 +- src/interfaces/libpq/fe-secure-gssapi.c | 12 +-- src/interfaces/libpq/fe-secure-openssl.c | 64 ++++++------ src/interfaces/libpq/fe-secure.c | 8 +- src/interfaces/libpq/libpq-int.h | 4 +- 12 files changed, 149 insertions(+), 149 deletions(-) diff --git a/src/interfaces/libpq/fe-auth-scram.c b/src/interfaces/libpq/fe-auth-scram.c index 9c42ea4f81..12c3d0bc33 100644 --- a/src/interfaces/libpq/fe-auth-scram.c +++ b/src/interfaces/libpq/fe-auth-scram.c @@ -716,7 +716,7 @@ read_server_final_message(fe_scram_state *state, char *input) return false; } libpq_append_conn_error(conn, "error received from server in SCRAM exchange: %s", - errmsg); + errmsg); return false; } diff --git a/src/interfaces/libpq/fe-auth.c b/src/interfaces/libpq/fe-auth.c index 9afc6f19b9..ab454e6cd0 100644 --- a/src/interfaces/libpq/fe-auth.c +++ b/src/interfaces/libpq/fe-auth.c @@ -73,7 +73,7 @@ pg_GSS_continue(PGconn *conn, int payloadlen) if (!ginbuf.value) { libpq_append_conn_error(conn, "out of memory allocating GSSAPI buffer (%d)", - payloadlen); + payloadlen); return STATUS_ERROR; } if (pqGetnchar(ginbuf.value, payloadlen, conn)) @@ -223,7 +223,7 @@ pg_SSPI_continue(PGconn *conn, int payloadlen) if (!inputbuf) { libpq_append_conn_error(conn, "out of memory allocating SSPI buffer (%d)", - payloadlen); + payloadlen); return STATUS_ERROR; } if (pqGetnchar(inputbuf, payloadlen, conn)) @@ -623,7 +623,7 @@ pg_SASL_continue(PGconn *conn, int payloadlen, bool final) if (!challenge) { libpq_append_conn_error(conn, "out of memory allocating SASL buffer (%d)", - payloadlen); + payloadlen); return STATUS_ERROR; } @@ -1277,7 +1277,7 @@ PQencryptPasswordConn(PGconn *conn, const char *passwd, const char *user, else { libpq_append_conn_error(conn, "unrecognized password encryption algorithm \"%s\"", - algorithm); + algorithm); return NULL; } diff --git a/src/interfaces/libpq/fe-connect.c b/src/interfaces/libpq/fe-connect.c index 50b5df3490..773e9e1f3a 100644 --- a/src/interfaces/libpq/fe-connect.c +++ b/src/interfaces/libpq/fe-connect.c @@ -1079,7 +1079,7 @@ connectOptions2(PGconn *conn) { conn->status = CONNECTION_BAD; libpq_append_conn_error(conn, "could not match %d host names to %d hostaddr values", - count_comma_separated_elems(conn->pghost), conn->nconnhost); + count_comma_separated_elems(conn->pghost), conn->nconnhost); return false; } } @@ -1159,7 +1159,7 @@ connectOptions2(PGconn *conn) { conn->status = CONNECTION_BAD; libpq_append_conn_error(conn, "could not match %d port numbers to %d hosts", - count_comma_separated_elems(conn->pgport), conn->nconnhost); + count_comma_separated_elems(conn->pgport), conn->nconnhost); return false; } } @@ -1248,7 +1248,7 @@ connectOptions2(PGconn *conn) { conn->status = CONNECTION_BAD; libpq_append_conn_error(conn, "invalid %s value: \"%s\"", - "channel_binding", conn->channel_binding); + "channel_binding", conn->channel_binding); return false; } } @@ -1273,7 +1273,7 @@ connectOptions2(PGconn *conn) { conn->status = CONNECTION_BAD; libpq_append_conn_error(conn, "invalid %s value: \"%s\"", - "sslmode", conn->sslmode); + "sslmode", conn->sslmode); return false; } @@ -1293,7 +1293,7 @@ connectOptions2(PGconn *conn) case 'v': /* "verify-ca" or "verify-full" */ conn->status = CONNECTION_BAD; libpq_append_conn_error(conn, "sslmode value \"%s\" invalid when SSL support is not compiled in", - conn->sslmode); + conn->sslmode); return false; } #endif @@ -1313,16 +1313,16 @@ connectOptions2(PGconn *conn) { conn->status = CONNECTION_BAD; libpq_append_conn_error(conn, "invalid %s value: \"%s\"", - "ssl_min_protocol_version", - conn->ssl_min_protocol_version); + "ssl_min_protocol_version", + conn->ssl_min_protocol_version); return false; } if (!sslVerifyProtocolVersion(conn->ssl_max_protocol_version)) { conn->status = CONNECTION_BAD; libpq_append_conn_error(conn, "invalid %s value: \"%s\"", - "ssl_max_protocol_version", - conn->ssl_max_protocol_version); + "ssl_max_protocol_version", + conn->ssl_max_protocol_version); return false; } @@ -1359,7 +1359,7 @@ connectOptions2(PGconn *conn) { conn->status = CONNECTION_BAD; libpq_append_conn_error(conn, "gssencmode value \"%s\" invalid when GSSAPI support is not compiled in", - conn->gssencmode); + conn->gssencmode); return false; } #endif @@ -1392,8 +1392,8 @@ connectOptions2(PGconn *conn) { conn->status = CONNECTION_BAD; libpq_append_conn_error(conn, "invalid %s value: \"%s\"", - "target_session_attrs", - conn->target_session_attrs); + "target_session_attrs", + conn->target_session_attrs); return false; } } @@ -1609,7 +1609,7 @@ connectNoDelay(PGconn *conn) char sebuf[PG_STRERROR_R_BUFLEN]; libpq_append_conn_error(conn, "could not set socket to TCP no delay mode: %s", - SOCK_STRERROR(SOCK_ERRNO, sebuf, sizeof(sebuf))); + SOCK_STRERROR(SOCK_ERRNO, sebuf, sizeof(sebuf))); return 0; } #endif @@ -1787,7 +1787,7 @@ parse_int_param(const char *value, int *result, PGconn *conn, error: libpq_append_conn_error(conn, "invalid integer value \"%s\" for connection option \"%s\"", - value, context); + value, context); return false; } @@ -1816,9 +1816,9 @@ setKeepalivesIdle(PGconn *conn) char sebuf[PG_STRERROR_R_BUFLEN]; libpq_append_conn_error(conn, "%s(%s) failed: %s", - "setsockopt", - PG_TCP_KEEPALIVE_IDLE_STR, - SOCK_STRERROR(SOCK_ERRNO, sebuf, sizeof(sebuf))); + "setsockopt", + PG_TCP_KEEPALIVE_IDLE_STR, + SOCK_STRERROR(SOCK_ERRNO, sebuf, sizeof(sebuf))); return 0; } #endif @@ -1850,9 +1850,9 @@ setKeepalivesInterval(PGconn *conn) char sebuf[PG_STRERROR_R_BUFLEN]; libpq_append_conn_error(conn, "%s(%s) failed: %s", - "setsockopt", - "TCP_KEEPINTVL", - SOCK_STRERROR(SOCK_ERRNO, sebuf, sizeof(sebuf))); + "setsockopt", + "TCP_KEEPINTVL", + SOCK_STRERROR(SOCK_ERRNO, sebuf, sizeof(sebuf))); return 0; } #endif @@ -1885,9 +1885,9 @@ setKeepalivesCount(PGconn *conn) char sebuf[PG_STRERROR_R_BUFLEN]; libpq_append_conn_error(conn, "%s(%s) failed: %s", - "setsockopt", - "TCP_KEEPCNT", - SOCK_STRERROR(SOCK_ERRNO, sebuf, sizeof(sebuf))); + "setsockopt", + "TCP_KEEPCNT", + SOCK_STRERROR(SOCK_ERRNO, sebuf, sizeof(sebuf))); return 0; } #endif @@ -1949,8 +1949,8 @@ prepKeepalivesWin32(PGconn *conn) if (!setKeepalivesWin32(conn->sock, idle, interval)) { libpq_append_conn_error(conn, "%s(%s) failed: error code %d", - "WSAIoctl", "SIO_KEEPALIVE_VALS", - WSAGetLastError()); + "WSAIoctl", "SIO_KEEPALIVE_VALS", + WSAGetLastError()); return 0; } return 1; @@ -1983,9 +1983,9 @@ setTCPUserTimeout(PGconn *conn) char sebuf[256]; libpq_append_conn_error(conn, "%s(%s) failed: %s", - "setsockopt", - "TCP_USER_TIMEOUT", - SOCK_STRERROR(SOCK_ERRNO, sebuf, sizeof(sebuf))); + "setsockopt", + "TCP_USER_TIMEOUT", + SOCK_STRERROR(SOCK_ERRNO, sebuf, sizeof(sebuf))); return 0; } #endif @@ -2354,7 +2354,7 @@ keep_going: /* We will come back to here until there is if (ret || !conn->addrlist) { libpq_append_conn_error(conn, "could not translate host name \"%s\" to address: %s", - ch->host, gai_strerror(ret)); + ch->host, gai_strerror(ret)); goto keep_going; } break; @@ -2366,7 +2366,7 @@ keep_going: /* We will come back to here until there is if (ret || !conn->addrlist) { libpq_append_conn_error(conn, "could not parse network address \"%s\": %s", - ch->hostaddr, gai_strerror(ret)); + ch->hostaddr, gai_strerror(ret)); goto keep_going; } break; @@ -2377,8 +2377,8 @@ keep_going: /* We will come back to here until there is if (strlen(portstr) >= UNIXSOCK_PATH_BUFLEN) { libpq_append_conn_error(conn, "Unix-domain socket path \"%s\" is too long (maximum %d bytes)", - portstr, - (int) (UNIXSOCK_PATH_BUFLEN - 1)); + portstr, + (int) (UNIXSOCK_PATH_BUFLEN - 1)); goto keep_going; } @@ -2391,7 +2391,7 @@ keep_going: /* We will come back to here until there is if (ret || !conn->addrlist) { libpq_append_conn_error(conn, "could not translate Unix-domain socket path \"%s\" to address: %s", - portstr, gai_strerror(ret)); + portstr, gai_strerror(ret)); goto keep_going; } break; @@ -2513,7 +2513,7 @@ keep_going: /* We will come back to here until there is } emitHostIdentityInfo(conn, host_addr); libpq_append_conn_error(conn, "could not create socket: %s", - SOCK_STRERROR(errorno, sebuf, sizeof(sebuf))); + SOCK_STRERROR(errorno, sebuf, sizeof(sebuf))); goto error_return; } @@ -2543,7 +2543,7 @@ keep_going: /* We will come back to here until there is if (!pg_set_noblock(conn->sock)) { libpq_append_conn_error(conn, "could not set socket to nonblocking mode: %s", - SOCK_STRERROR(SOCK_ERRNO, sebuf, sizeof(sebuf))); + SOCK_STRERROR(SOCK_ERRNO, sebuf, sizeof(sebuf))); conn->try_next_addr = true; goto keep_going; } @@ -2552,7 +2552,7 @@ keep_going: /* We will come back to here until there is if (fcntl(conn->sock, F_SETFD, FD_CLOEXEC) == -1) { libpq_append_conn_error(conn, "could not set socket to close-on-exec mode: %s", - SOCK_STRERROR(SOCK_ERRNO, sebuf, sizeof(sebuf))); + SOCK_STRERROR(SOCK_ERRNO, sebuf, sizeof(sebuf))); conn->try_next_addr = true; goto keep_going; } @@ -2581,9 +2581,9 @@ keep_going: /* We will come back to here until there is (char *) &on, sizeof(on)) < 0) { libpq_append_conn_error(conn, "%s(%s) failed: %s", - "setsockopt", - "SO_KEEPALIVE", - SOCK_STRERROR(SOCK_ERRNO, sebuf, sizeof(sebuf))); + "setsockopt", + "SO_KEEPALIVE", + SOCK_STRERROR(SOCK_ERRNO, sebuf, sizeof(sebuf))); err = 1; } else if (!setKeepalivesIdle(conn) @@ -2708,7 +2708,7 @@ keep_going: /* We will come back to here until there is (char *) &optval, &optlen) == -1) { libpq_append_conn_error(conn, "could not get socket error status: %s", - SOCK_STRERROR(SOCK_ERRNO, sebuf, sizeof(sebuf))); + SOCK_STRERROR(SOCK_ERRNO, sebuf, sizeof(sebuf))); goto error_return; } else if (optval != 0) @@ -2735,7 +2735,7 @@ keep_going: /* We will come back to here until there is &conn->laddr.salen) < 0) { libpq_append_conn_error(conn, "could not get client address from socket: %s", - SOCK_STRERROR(SOCK_ERRNO, sebuf, sizeof(sebuf))); + SOCK_STRERROR(SOCK_ERRNO, sebuf, sizeof(sebuf))); goto error_return; } @@ -2775,7 +2775,7 @@ keep_going: /* We will come back to here until there is libpq_append_conn_error(conn, "requirepeer parameter is not supported on this platform"); else libpq_append_conn_error(conn, "could not get peer credentials: %s", - strerror_r(errno, sebuf, sizeof(sebuf))); + strerror_r(errno, sebuf, sizeof(sebuf))); goto error_return; } @@ -2788,7 +2788,7 @@ keep_going: /* We will come back to here until there is if (strcmp(remote_username, conn->requirepeer) != 0) { libpq_append_conn_error(conn, "requirepeer specifies \"%s\", but actual peer user name is \"%s\"", - conn->requirepeer, remote_username); + conn->requirepeer, remote_username); free(remote_username); goto error_return; } @@ -2829,7 +2829,7 @@ keep_going: /* We will come back to here until there is if (pqPacketSend(conn, 0, &pv, sizeof(pv)) != STATUS_OK) { libpq_append_conn_error(conn, "could not send GSSAPI negotiation packet: %s", - SOCK_STRERROR(SOCK_ERRNO, sebuf, sizeof(sebuf))); + SOCK_STRERROR(SOCK_ERRNO, sebuf, sizeof(sebuf))); goto error_return; } @@ -2840,7 +2840,7 @@ keep_going: /* We will come back to here until there is else if (!conn->gctx && conn->gssencmode[0] == 'r') { libpq_append_conn_error(conn, - "GSSAPI encryption required but was impossible (possibly no credential cache, no server support, or using a local socket)"); + "GSSAPI encryption required but was impossible (possibly no credential cache, no server support, or using a local socket)"); goto error_return; } #endif @@ -2882,7 +2882,7 @@ keep_going: /* We will come back to here until there is if (pqPacketSend(conn, 0, &pv, sizeof(pv)) != STATUS_OK) { libpq_append_conn_error(conn, "could not send SSL negotiation packet: %s", - SOCK_STRERROR(SOCK_ERRNO, sebuf, sizeof(sebuf))); + SOCK_STRERROR(SOCK_ERRNO, sebuf, sizeof(sebuf))); goto error_return; } /* Ok, wait for response */ @@ -2911,7 +2911,7 @@ keep_going: /* We will come back to here until there is if (pqPacketSend(conn, 0, startpacket, packetlen) != STATUS_OK) { libpq_append_conn_error(conn, "could not send startup packet: %s", - SOCK_STRERROR(SOCK_ERRNO, sebuf, sizeof(sebuf))); + SOCK_STRERROR(SOCK_ERRNO, sebuf, sizeof(sebuf))); free(startpacket); goto error_return; } @@ -3012,7 +3012,7 @@ keep_going: /* We will come back to here until there is else { libpq_append_conn_error(conn, "received invalid response to SSL negotiation: %c", - SSLok); + SSLok); goto error_return; } } @@ -3123,7 +3123,7 @@ keep_going: /* We will come back to here until there is else if (gss_ok != 'G') { libpq_append_conn_error(conn, "received invalid response to GSSAPI negotiation: %c", - gss_ok); + gss_ok); goto error_return; } } @@ -3201,7 +3201,7 @@ keep_going: /* We will come back to here until there is if (!(beresp == 'R' || beresp == 'v' || beresp == 'E')) { libpq_append_conn_error(conn, "expected authentication request from server, but received %c", - beresp); + beresp); goto error_return; } @@ -3216,17 +3216,17 @@ keep_going: /* We will come back to here until there is * Try to validate message length before using it. * Authentication requests can't be very large, although GSS * auth requests may not be that small. Same for - * NegotiateProtocolVersion. Errors can be a - * little larger, but not huge. If we see a large apparent - * length in an error, it means we're really talking to a - * pre-3.0-protocol server; cope. (Before version 14, the - * server also used the old protocol for errors that happened - * before processing the startup packet.) + * NegotiateProtocolVersion. Errors can be a little larger, + * but not huge. If we see a large apparent length in an + * error, it means we're really talking to a pre-3.0-protocol + * server; cope. (Before version 14, the server also used the + * old protocol for errors that happened before processing the + * startup packet.) */ if ((beresp == 'R' || beresp == 'v') && (msgLength < 8 || msgLength > 2000)) { libpq_append_conn_error(conn, "expected authentication request from server, but received %c", - beresp); + beresp); goto error_return; } @@ -3705,7 +3705,7 @@ keep_going: /* We will come back to here until there is /* Append error report to conn->errorMessage. */ libpq_append_conn_error(conn, "\"%s\" failed", - "SHOW transaction_read_only"); + "SHOW transaction_read_only"); /* Close connection politely. */ conn->status = CONNECTION_OK; @@ -3755,7 +3755,7 @@ keep_going: /* We will come back to here until there is /* Append error report to conn->errorMessage. */ libpq_append_conn_error(conn, "\"%s\" failed", - "SELECT pg_is_in_recovery()"); + "SELECT pg_is_in_recovery()"); /* Close connection politely. */ conn->status = CONNECTION_OK; @@ -3768,8 +3768,8 @@ keep_going: /* We will come back to here until there is default: libpq_append_conn_error(conn, - "invalid connection state %d, probably indicative of memory corruption", - conn->status); + "invalid connection state %d, probably indicative of memory corruption", + conn->status); goto error_return; } @@ -7148,7 +7148,7 @@ pgpassfileWarning(PGconn *conn) if (sqlstate && strcmp(sqlstate, ERRCODE_INVALID_PASSWORD) == 0) libpq_append_conn_error(conn, "password retrieved from file \"%s\"", - conn->pgpassfile); + conn->pgpassfile); } } diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c index ec62550e38..0c2dae6ed9 100644 --- a/src/interfaces/libpq/fe-exec.c +++ b/src/interfaces/libpq/fe-exec.c @@ -1444,7 +1444,7 @@ PQsendQueryInternal(PGconn *conn, const char *query, bool newQuery) if (conn->pipelineStatus != PQ_PIPELINE_OFF) { libpq_append_conn_error(conn, "%s not allowed in pipeline mode", - "PQsendQuery"); + "PQsendQuery"); return 0; } @@ -1512,7 +1512,7 @@ PQsendQueryParams(PGconn *conn, if (nParams < 0 || nParams > PQ_QUERY_PARAM_MAX_LIMIT) { libpq_append_conn_error(conn, "number of parameters must be between 0 and %d", - PQ_QUERY_PARAM_MAX_LIMIT); + PQ_QUERY_PARAM_MAX_LIMIT); return 0; } @@ -1558,7 +1558,7 @@ PQsendPrepare(PGconn *conn, if (nParams < 0 || nParams > PQ_QUERY_PARAM_MAX_LIMIT) { libpq_append_conn_error(conn, "number of parameters must be between 0 and %d", - PQ_QUERY_PARAM_MAX_LIMIT); + PQ_QUERY_PARAM_MAX_LIMIT); return 0; } @@ -1652,7 +1652,7 @@ PQsendQueryPrepared(PGconn *conn, if (nParams < 0 || nParams > PQ_QUERY_PARAM_MAX_LIMIT) { libpq_append_conn_error(conn, "number of parameters must be between 0 and %d", - PQ_QUERY_PARAM_MAX_LIMIT); + PQ_QUERY_PARAM_MAX_LIMIT); return 0; } @@ -2099,10 +2099,9 @@ PQgetResult(PGconn *conn) /* * We're about to return the NULL that terminates the round of - * results from the current query; prepare to send the results - * of the next query, if any, when we're called next. If there's - * no next element in the command queue, this gets us in IDLE - * state. + * results from the current query; prepare to send the results of + * the next query, if any, when we're called next. If there's no + * next element in the command queue, this gets us in IDLE state. */ pqPipelineProcessQueue(conn); res = NULL; /* query is complete */ @@ -3047,6 +3046,7 @@ pqPipelineProcessQueue(PGconn *conn) return; case PGASYNC_IDLE: + /* * If we're in IDLE mode and there's some command in the queue, * get us into PIPELINE_IDLE mode and process normally. Otherwise diff --git a/src/interfaces/libpq/fe-lobj.c b/src/interfaces/libpq/fe-lobj.c index 4cb6a46859..206266fd04 100644 --- a/src/interfaces/libpq/fe-lobj.c +++ b/src/interfaces/libpq/fe-lobj.c @@ -142,7 +142,7 @@ lo_truncate(PGconn *conn, int fd, size_t len) if (conn->lobjfuncs->fn_lo_truncate == 0) { libpq_append_conn_error(conn, "cannot determine OID of function %s", - "lo_truncate"); + "lo_truncate"); return -1; } @@ -205,7 +205,7 @@ lo_truncate64(PGconn *conn, int fd, pg_int64 len) if (conn->lobjfuncs->fn_lo_truncate64 == 0) { libpq_append_conn_error(conn, "cannot determine OID of function %s", - "lo_truncate64"); + "lo_truncate64"); return -1; } @@ -395,7 +395,7 @@ lo_lseek64(PGconn *conn, int fd, pg_int64 offset, int whence) if (conn->lobjfuncs->fn_lo_lseek64 == 0) { libpq_append_conn_error(conn, "cannot determine OID of function %s", - "lo_lseek64"); + "lo_lseek64"); return -1; } @@ -485,7 +485,7 @@ lo_create(PGconn *conn, Oid lobjId) if (conn->lobjfuncs->fn_lo_create == 0) { libpq_append_conn_error(conn, "cannot determine OID of function %s", - "lo_create"); + "lo_create"); return InvalidOid; } @@ -558,7 +558,7 @@ lo_tell64(PGconn *conn, int fd) if (conn->lobjfuncs->fn_lo_tell64 == 0) { libpq_append_conn_error(conn, "cannot determine OID of function %s", - "lo_tell64"); + "lo_tell64"); return -1; } @@ -667,7 +667,7 @@ lo_import_internal(PGconn *conn, const char *filename, Oid oid) if (fd < 0) { /* error */ libpq_append_conn_error(conn, "could not open file \"%s\": %s", - filename, strerror_r(errno, sebuf, sizeof(sebuf))); + filename, strerror_r(errno, sebuf, sizeof(sebuf))); return InvalidOid; } @@ -723,8 +723,8 @@ lo_import_internal(PGconn *conn, const char *filename, Oid oid) /* deliberately overwrite any error from lo_close */ pqClearConnErrorState(conn); libpq_append_conn_error(conn, "could not read from file \"%s\": %s", - filename, - strerror_r(save_errno, sebuf, sizeof(sebuf))); + filename, + strerror_r(save_errno, sebuf, sizeof(sebuf))); return InvalidOid; } @@ -778,8 +778,8 @@ lo_export(PGconn *conn, Oid lobjId, const char *filename) /* deliberately overwrite any error from lo_close */ pqClearConnErrorState(conn); libpq_append_conn_error(conn, "could not open file \"%s\": %s", - filename, - strerror_r(save_errno, sebuf, sizeof(sebuf))); + filename, + strerror_r(save_errno, sebuf, sizeof(sebuf))); return -1; } @@ -799,8 +799,8 @@ lo_export(PGconn *conn, Oid lobjId, const char *filename) /* deliberately overwrite any error from lo_close */ pqClearConnErrorState(conn); libpq_append_conn_error(conn, "could not write to file \"%s\": %s", - filename, - strerror_r(save_errno, sebuf, sizeof(sebuf))); + filename, + strerror_r(save_errno, sebuf, sizeof(sebuf))); return -1; } } @@ -822,7 +822,7 @@ lo_export(PGconn *conn, Oid lobjId, const char *filename) if (close(fd) != 0 && result >= 0) { libpq_append_conn_error(conn, "could not write to file \"%s\": %s", - filename, strerror_r(errno, sebuf, sizeof(sebuf))); + filename, strerror_r(errno, sebuf, sizeof(sebuf))); result = -1; } @@ -954,56 +954,56 @@ lo_initialize(PGconn *conn) if (lobjfuncs->fn_lo_open == 0) { libpq_append_conn_error(conn, "cannot determine OID of function %s", - "lo_open"); + "lo_open"); free(lobjfuncs); return -1; } if (lobjfuncs->fn_lo_close == 0) { libpq_append_conn_error(conn, "cannot determine OID of function %s", - "lo_close"); + "lo_close"); free(lobjfuncs); return -1; } if (lobjfuncs->fn_lo_creat == 0) { libpq_append_conn_error(conn, "cannot determine OID of function %s", - "lo_creat"); + "lo_creat"); free(lobjfuncs); return -1; } if (lobjfuncs->fn_lo_unlink == 0) { libpq_append_conn_error(conn, "cannot determine OID of function %s", - "lo_unlink"); + "lo_unlink"); free(lobjfuncs); return -1; } if (lobjfuncs->fn_lo_lseek == 0) { libpq_append_conn_error(conn, "cannot determine OID of function %s", - "lo_lseek"); + "lo_lseek"); free(lobjfuncs); return -1; } if (lobjfuncs->fn_lo_tell == 0) { libpq_append_conn_error(conn, "cannot determine OID of function %s", - "lo_tell"); + "lo_tell"); free(lobjfuncs); return -1; } if (lobjfuncs->fn_lo_read == 0) { libpq_append_conn_error(conn, "cannot determine OID of function %s", - "loread"); + "loread"); free(lobjfuncs); return -1; } if (lobjfuncs->fn_lo_write == 0) { libpq_append_conn_error(conn, "cannot determine OID of function %s", - "lowrite"); + "lowrite"); free(lobjfuncs); return -1; } diff --git a/src/interfaces/libpq/fe-misc.c b/src/interfaces/libpq/fe-misc.c index 3653a1a8a6..660cdec93c 100644 --- a/src/interfaces/libpq/fe-misc.c +++ b/src/interfaces/libpq/fe-misc.c @@ -749,8 +749,8 @@ retry4: */ definitelyEOF: libpq_append_conn_error(conn, "server closed the connection unexpectedly\n" - "\tThis probably means the server terminated abnormally\n" - "\tbefore or while processing the request."); + "\tThis probably means the server terminated abnormally\n" + "\tbefore or while processing the request."); /* Come here if lower-level code already set a suitable errorMessage */ definitelyFailed: @@ -1067,7 +1067,7 @@ pqSocketCheck(PGconn *conn, int forRead, int forWrite, time_t end_time) char sebuf[PG_STRERROR_R_BUFLEN]; libpq_append_conn_error(conn, "%s() failed: %s", "select", - SOCK_STRERROR(SOCK_ERRNO, sebuf, sizeof(sebuf))); + SOCK_STRERROR(SOCK_ERRNO, sebuf, sizeof(sebuf))); } return result; @@ -1280,7 +1280,7 @@ libpq_ngettext(const char *msgid, const char *msgid_plural, unsigned long n) * newline. */ void -libpq_append_error(PQExpBuffer errorMessage, const char *fmt, ...) +libpq_append_error(PQExpBuffer errorMessage, const char *fmt,...) { int save_errno = errno; bool done; @@ -1309,7 +1309,7 @@ libpq_append_error(PQExpBuffer errorMessage, const char *fmt, ...) * format should not end with a newline. */ void -libpq_append_conn_error(PGconn *conn, const char *fmt, ...) +libpq_append_conn_error(PGconn *conn, const char *fmt,...) { int save_errno = errno; bool done; diff --git a/src/interfaces/libpq/fe-protocol3.c b/src/interfaces/libpq/fe-protocol3.c index 8ab6a88416..b79d74f748 100644 --- a/src/interfaces/libpq/fe-protocol3.c +++ b/src/interfaces/libpq/fe-protocol3.c @@ -466,7 +466,7 @@ static void handleSyncLoss(PGconn *conn, char id, int msgLength) { libpq_append_conn_error(conn, "lost synchronization with server: got message type \"%c\", length %d", - id, msgLength); + id, msgLength); /* build an error result holding the error message */ pqSaveErrorResult(conn); conn->asyncStatus = PGASYNC_READY; /* drop out of PQgetResult wait loop */ diff --git a/src/interfaces/libpq/fe-secure-common.c b/src/interfaces/libpq/fe-secure-common.c index de115b3764..3ecc7bf615 100644 --- a/src/interfaces/libpq/fe-secure-common.c +++ b/src/interfaces/libpq/fe-secure-common.c @@ -226,7 +226,7 @@ pq_verify_peer_name_matches_certificate_ip(PGconn *conn, * wrong given the subject matter. */ libpq_append_conn_error(conn, "certificate contains IP address with invalid length %zu", - iplen); + iplen); return -1; } @@ -235,7 +235,7 @@ pq_verify_peer_name_matches_certificate_ip(PGconn *conn, if (!addrstr) { libpq_append_conn_error(conn, "could not convert certificate's IP address to string: %s", - strerror_r(errno, sebuf, sizeof(sebuf))); + strerror_r(errno, sebuf, sizeof(sebuf))); return -1; } @@ -292,7 +292,7 @@ pq_verify_peer_name_matches_certificate(PGconn *conn) else if (names_examined == 1) { libpq_append_conn_error(conn, "server certificate for \"%s\" does not match host name \"%s\"", - first_name, host); + first_name, host); } else { diff --git a/src/interfaces/libpq/fe-secure-gssapi.c b/src/interfaces/libpq/fe-secure-gssapi.c index 6220e4a101..bed6e62435 100644 --- a/src/interfaces/libpq/fe-secure-gssapi.c +++ b/src/interfaces/libpq/fe-secure-gssapi.c @@ -213,8 +213,8 @@ pg_GSS_write(PGconn *conn, const void *ptr, size_t len) if (output.length > PQ_GSS_SEND_BUFFER_SIZE - sizeof(uint32)) { libpq_append_conn_error(conn, "client tried to send oversize GSSAPI packet (%zu > %zu)", - (size_t) output.length, - PQ_GSS_SEND_BUFFER_SIZE - sizeof(uint32)); + (size_t) output.length, + PQ_GSS_SEND_BUFFER_SIZE - sizeof(uint32)); errno = EIO; /* for lack of a better idea */ goto cleanup; } @@ -349,8 +349,8 @@ pg_GSS_read(PGconn *conn, void *ptr, size_t len) if (input.length > PQ_GSS_RECV_BUFFER_SIZE - sizeof(uint32)) { libpq_append_conn_error(conn, "oversize GSSAPI packet sent by the server (%zu > %zu)", - (size_t) input.length, - PQ_GSS_RECV_BUFFER_SIZE - sizeof(uint32)); + (size_t) input.length, + PQ_GSS_RECV_BUFFER_SIZE - sizeof(uint32)); errno = EIO; /* for lack of a better idea */ return -1; } @@ -588,8 +588,8 @@ pqsecure_open_gss(PGconn *conn) if (input.length > PQ_GSS_RECV_BUFFER_SIZE - sizeof(uint32)) { libpq_append_conn_error(conn, "oversize GSSAPI packet sent by the server (%zu > %zu)", - (size_t) input.length, - PQ_GSS_RECV_BUFFER_SIZE - sizeof(uint32)); + (size_t) input.length, + PQ_GSS_RECV_BUFFER_SIZE - sizeof(uint32)); return PGRES_POLLING_FAILED; } diff --git a/src/interfaces/libpq/fe-secure-openssl.c b/src/interfaces/libpq/fe-secure-openssl.c index 983536de25..ab2cbf045b 100644 --- a/src/interfaces/libpq/fe-secure-openssl.c +++ b/src/interfaces/libpq/fe-secure-openssl.c @@ -213,12 +213,12 @@ rloop: if (result_errno == EPIPE || result_errno == ECONNRESET) libpq_append_conn_error(conn, "server closed the connection unexpectedly\n" - "\tThis probably means the server terminated abnormally\n" - "\tbefore or while processing the request."); + "\tThis probably means the server terminated abnormally\n" + "\tbefore or while processing the request."); else libpq_append_conn_error(conn, "SSL SYSCALL error: %s", - SOCK_STRERROR(result_errno, - sebuf, sizeof(sebuf))); + SOCK_STRERROR(result_errno, + sebuf, sizeof(sebuf))); } else { @@ -313,12 +313,12 @@ pgtls_write(PGconn *conn, const void *ptr, size_t len) result_errno = SOCK_ERRNO; if (result_errno == EPIPE || result_errno == ECONNRESET) libpq_append_conn_error(conn, "server closed the connection unexpectedly\n" - "\tThis probably means the server terminated abnormally\n" - "\tbefore or while processing the request."); + "\tThis probably means the server terminated abnormally\n" + "\tbefore or while processing the request."); else libpq_append_conn_error(conn, "SSL SYSCALL error: %s", - SOCK_STRERROR(result_errno, - sebuf, sizeof(sebuf))); + SOCK_STRERROR(result_errno, + sebuf, sizeof(sebuf))); } else { @@ -410,7 +410,7 @@ pgtls_get_peer_certificate_hash(PGconn *conn, size_t *len) if (algo_type == NULL) { libpq_append_conn_error(conn, "could not find digest for NID %s", - OBJ_nid2sn(algo_nid)); + OBJ_nid2sn(algo_nid)); return NULL; } break; @@ -962,7 +962,7 @@ initialize_SSL(PGconn *conn) if (ssl_min_ver == -1) { libpq_append_conn_error(conn, "invalid value \"%s\" for minimum SSL protocol version", - conn->ssl_min_protocol_version); + conn->ssl_min_protocol_version); SSL_CTX_free(SSL_context); return -1; } @@ -988,7 +988,7 @@ initialize_SSL(PGconn *conn) if (ssl_max_ver == -1) { libpq_append_conn_error(conn, "invalid value \"%s\" for maximum SSL protocol version", - conn->ssl_max_protocol_version); + conn->ssl_max_protocol_version); SSL_CTX_free(SSL_context); return -1; } @@ -1032,7 +1032,7 @@ initialize_SSL(PGconn *conn) char *err = SSLerrmessage(ERR_get_error()); libpq_append_conn_error(conn, "could not read root certificate file \"%s\": %s", - fnbuf, err); + fnbuf, err); SSLerrfree(err); SSL_CTX_free(SSL_context); return -1; @@ -1084,10 +1084,10 @@ initialize_SSL(PGconn *conn) */ if (fnbuf[0] == '\0') libpq_append_conn_error(conn, "could not get home directory to locate root certificate file\n" - "Either provide the file or change sslmode to disable server certificate verification."); + "Either provide the file or change sslmode to disable server certificate verification."); else libpq_append_conn_error(conn, "root certificate file \"%s\" does not exist\n" - "Either provide the file or change sslmode to disable server certificate verification.", fnbuf); + "Either provide the file or change sslmode to disable server certificate verification.", fnbuf); SSL_CTX_free(SSL_context); return -1; } @@ -1117,7 +1117,7 @@ initialize_SSL(PGconn *conn) if (errno != ENOENT && errno != ENOTDIR) { libpq_append_conn_error(conn, "could not open certificate file \"%s\": %s", - fnbuf, strerror_r(errno, sebuf, sizeof(sebuf))); + fnbuf, strerror_r(errno, sebuf, sizeof(sebuf))); SSL_CTX_free(SSL_context); return -1; } @@ -1135,7 +1135,7 @@ initialize_SSL(PGconn *conn) char *err = SSLerrmessage(ERR_get_error()); libpq_append_conn_error(conn, "could not read certificate file \"%s\": %s", - fnbuf, err); + fnbuf, err); SSLerrfree(err); SSL_CTX_free(SSL_context); return -1; @@ -1234,7 +1234,7 @@ initialize_SSL(PGconn *conn) char *err = SSLerrmessage(ERR_get_error()); libpq_append_conn_error(conn, "could not load SSL engine \"%s\": %s", - engine_str, err); + engine_str, err); SSLerrfree(err); free(engine_str); return -1; @@ -1245,7 +1245,7 @@ initialize_SSL(PGconn *conn) char *err = SSLerrmessage(ERR_get_error()); libpq_append_conn_error(conn, "could not initialize SSL engine \"%s\": %s", - engine_str, err); + engine_str, err); SSLerrfree(err); ENGINE_free(conn->engine); conn->engine = NULL; @@ -1260,7 +1260,7 @@ initialize_SSL(PGconn *conn) char *err = SSLerrmessage(ERR_get_error()); libpq_append_conn_error(conn, "could not read private SSL key \"%s\" from engine \"%s\": %s", - engine_colon, engine_str, err); + engine_colon, engine_str, err); SSLerrfree(err); ENGINE_finish(conn->engine); ENGINE_free(conn->engine); @@ -1273,7 +1273,7 @@ initialize_SSL(PGconn *conn) char *err = SSLerrmessage(ERR_get_error()); libpq_append_conn_error(conn, "could not load private SSL key \"%s\" from engine \"%s\": %s", - engine_colon, engine_str, err); + engine_colon, engine_str, err); SSLerrfree(err); ENGINE_finish(conn->engine); ENGINE_free(conn->engine); @@ -1310,10 +1310,10 @@ initialize_SSL(PGconn *conn) { if (errno == ENOENT) libpq_append_conn_error(conn, "certificate present, but not private key file \"%s\"", - fnbuf); + fnbuf); else libpq_append_conn_error(conn, "could not stat private key file \"%s\": %m", - fnbuf); + fnbuf); return -1; } @@ -1321,7 +1321,7 @@ initialize_SSL(PGconn *conn) if (!S_ISREG(buf.st_mode)) { libpq_append_conn_error(conn, "private key file \"%s\" is not a regular file", - fnbuf); + fnbuf); return -1; } @@ -1378,7 +1378,7 @@ initialize_SSL(PGconn *conn) if (SSL_use_PrivateKey_file(conn->ssl, fnbuf, SSL_FILETYPE_ASN1) != 1) { libpq_append_conn_error(conn, "could not load private key file \"%s\": %s", - fnbuf, err); + fnbuf, err); SSLerrfree(err); return -1; } @@ -1394,7 +1394,7 @@ initialize_SSL(PGconn *conn) char *err = SSLerrmessage(ERR_get_error()); libpq_append_conn_error(conn, "certificate does not match private key file \"%s\": %s", - fnbuf, err); + fnbuf, err); SSLerrfree(err); return -1; } @@ -1447,7 +1447,7 @@ open_client_SSL(PGconn *conn) if (r == -1) libpq_append_conn_error(conn, "SSL SYSCALL error: %s", - SOCK_STRERROR(SOCK_ERRNO, sebuf, sizeof(sebuf))); + SOCK_STRERROR(SOCK_ERRNO, sebuf, sizeof(sebuf))); else libpq_append_conn_error(conn, "SSL SYSCALL error: EOF detected"); pgtls_close(conn); @@ -1489,12 +1489,12 @@ open_client_SSL(PGconn *conn) case SSL_R_VERSION_TOO_LOW: #endif libpq_append_conn_error(conn, "This may indicate that the server does not support any SSL protocol version between %s and %s.", - conn->ssl_min_protocol_version ? - conn->ssl_min_protocol_version : - MIN_OPENSSL_TLS_VERSION, - conn->ssl_max_protocol_version ? - conn->ssl_max_protocol_version : - MAX_OPENSSL_TLS_VERSION); + conn->ssl_min_protocol_version ? + conn->ssl_min_protocol_version : + MIN_OPENSSL_TLS_VERSION, + conn->ssl_max_protocol_version ? + conn->ssl_max_protocol_version : + MAX_OPENSSL_TLS_VERSION); break; default: break; diff --git a/src/interfaces/libpq/fe-secure.c b/src/interfaces/libpq/fe-secure.c index 66e401bf3d..8069e38142 100644 --- a/src/interfaces/libpq/fe-secure.c +++ b/src/interfaces/libpq/fe-secure.c @@ -255,14 +255,14 @@ pqsecure_raw_read(PGconn *conn, void *ptr, size_t len) case EPIPE: case ECONNRESET: libpq_append_conn_error(conn, "server closed the connection unexpectedly\n" - "\tThis probably means the server terminated abnormally\n" - "\tbefore or while processing the request."); + "\tThis probably means the server terminated abnormally\n" + "\tbefore or while processing the request."); break; default: libpq_append_conn_error(conn, "could not receive data from server: %s", - SOCK_STRERROR(result_errno, - sebuf, sizeof(sebuf))); + SOCK_STRERROR(result_errno, + sebuf, sizeof(sebuf))); break; } } diff --git a/src/interfaces/libpq/libpq-int.h b/src/interfaces/libpq/libpq-int.h index d94b648ea5..712d572373 100644 --- a/src/interfaces/libpq/libpq-int.h +++ b/src/interfaces/libpq/libpq-int.h @@ -888,8 +888,8 @@ extern char *libpq_ngettext(const char *msgid, const char *msgid_plural, unsigne */ #undef _ -extern void libpq_append_error(PQExpBuffer errorMessage, const char *fmt, ...) pg_attribute_printf(2, 3); -extern void libpq_append_conn_error(PGconn *conn, const char *fmt, ...) pg_attribute_printf(2, 3); +extern void libpq_append_error(PQExpBuffer errorMessage, const char *fmt,...) pg_attribute_printf(2, 3); +extern void libpq_append_conn_error(PGconn *conn, const char *fmt,...) pg_attribute_printf(2, 3); /* * These macros are needed to let error-handling code be portable between -- 2.34.1 [application/octet-stream] v6-0002-Support-load-balancing-in-libpq.patch (28.7K, ../../DBBPR83MB05079A565E6AB609104D3F9DF7FE9@DBBPR83MB0507.EURPRD83.prod.outlook.com/3-v6-0002-Support-load-balancing-in-libpq.patch) download | inline diff: From 83d0abde7db0ddd07f62521c15812f73827f3d61 Mon Sep 17 00:00:00 2001 From: Jelte Fennema <[email protected]> Date: Mon, 12 Sep 2022 09:44:06 +0200 Subject: [PATCH v6 2/2] Support load balancing in libpq This adds support for load balancing to libpq using the newly added load_balance_hosts parameter. When setting the load_balance_hosts parameter to random, hosts and addresses will be connected to in a random order. This then results in load balancing across these hosts/addresses if multiple clients do this at the same time. This patch implements two levels of random load balancing: 1. The given hosts are randomly shuffled, before resolving them one-by-one. 2. Once a host its addresses get resolved, those addresses are shuffled, before trying to connect to them one-by-one. --- .cirrus.yml | 14 ++ doc/src/sgml/libpq.sgml | 69 +++++++ src/include/libpq/pqcomm.h | 6 + src/interfaces/libpq/fe-connect.c | 215 ++++++++++++++++++---- src/interfaces/libpq/libpq-int.h | 21 ++- src/interfaces/libpq/meson.build | 1 + src/interfaces/libpq/t/003_loadbalance.pl | 167 +++++++++++++++++ 7 files changed, 460 insertions(+), 33 deletions(-) create mode 100644 src/interfaces/libpq/t/003_loadbalance.pl diff --git a/.cirrus.yml b/.cirrus.yml index 69837bcd5a..f71e93451f 100644 --- a/.cirrus.yml +++ b/.cirrus.yml @@ -308,6 +308,14 @@ task: mkdir -m 770 /tmp/cores chown root:postgres /tmp/cores sysctl kernel.core_pattern='/tmp/cores/%e-%s-%p.core' + + setup_hosts_file_script: | + cat >> /etc/hosts <<-EOF + 127.0.0.1 pg-loadbalancetest + 127.0.0.2 pg-loadbalancetest + 127.0.0.3 pg-loadbalancetest + EOF + setup_additional_packages_script: | #apt-get update #DEBIAN_FRONTEND=noninteractive apt-get -y install ... @@ -557,6 +565,12 @@ task: setup_additional_packages_script: | REM choco install -y --no-progress ... + setup_hosts_file_script: | + echo 127.0.0.1 pg-loadbalancetest >> c:\Windows\System32\Drivers\etc\hosts + echo 127.0.0.2 pg-loadbalancetest >> c:\Windows\System32\Drivers\etc\hosts + echo 127.0.0.3 pg-loadbalancetest >> c:\Windows\System32\Drivers\etc\hosts + type c:\Windows\System32\Drivers\etc\hosts + # Use /DEBUG:FASTLINK to avoid high memory usage during linking configure_script: | vcvarsall x64 diff --git a/doc/src/sgml/libpq.sgml b/doc/src/sgml/libpq.sgml index af278660eb..03eab1d02e 100644 --- a/doc/src/sgml/libpq.sgml +++ b/doc/src/sgml/libpq.sgml @@ -1964,6 +1964,75 @@ postgresql://%2Fvar%2Flib%2Fpostgresql/dbname </para> </listitem> </varlistentry> + + <varlistentry id="libpq-connect-load-balance-hosts" xreflabel="load_balance_hosts"> + <term><literal>load_balance_hosts</literal></term> + <listitem> + <para> + Controls the order in which the client tries to connect to the available + hosts and addresses. It's typically used in combination with multiple + host names or a DNS record that returns multiple IPs. This parameter be + used in combination with <xref linkend="libpq-connect-target-session-attrs"/> + to, for example, load balance over stanby servers only. Once successfully + connected, subsequent queries on the returned connection will all be + sent to the same server. There are currently two modes: + <variablelist> + <varlistentry> + <term><literal>disable</literal> (default)</term> + <listitem> + <para> + Hosts are tried in the order in which they are provided and + addresses are tried in the order they are received from DNS or a + hosts file. + </para> + </listitem> + </varlistentry> + + <varlistentry> + <term><literal>random</literal></term> + <listitem> + <para> + The provided hosts and the addresses that they resolve to are + tried in random order. This value is mostly useful when opening + multiple connections at the same time, possibly from different + machines. This way connections can be load balanced across multiple + Postgres servers. + </para> + <para> + This algorithm uses two levels of random choices: First the hosts + will be resolved in random order. Then before resolving the next + host, all resolved addresses for the current host will be tried in + random order. This behaviour can lead to non-uniform address + selection in certain cases, for instance when some hosts resolve to + more addresses than others. So if you want uniform load balancing, + this is something to keep in mind. However, non-uniform load + balancing also can be used to your advantage, e.g. by providing the + hostname of a larger server multiple times in the host string so it + gets more connections. + </para> + <para> + When using this value it's recommended to also configure a reasonable + value for <xref linkend="libpq-connect-connect-timeout"/>. Because then, + if one of the nodes that are used for load balancing is not responding, + a new node will be tried. + </para> + </listitem> + </varlistentry> + </variablelist> + </para> + </listitem> + </varlistentry> + + <varlistentry id="libpq-random-seed" xreflabel="random_seed"> + <term><literal>random_seed</literal></term> + <listitem> + <para> + Sets the random seed that is used by <xref linkend="libpq-connect-load-balance-hosts"/> + to randomize the host order. This option is mostly useful when running + tests that require a stable random order. + </para> + </listitem> + </varlistentry> </variablelist> </para> </sect2> diff --git a/src/include/libpq/pqcomm.h b/src/include/libpq/pqcomm.h index 66ba359390..ee28e223bd 100644 --- a/src/include/libpq/pqcomm.h +++ b/src/include/libpq/pqcomm.h @@ -27,6 +27,12 @@ typedef struct socklen_t salen; } SockAddr; +typedef struct +{ + int family; + SockAddr addr; +} AddrInfo; + /* Configure the UNIX socket location for the well known port. */ #define UNIXSOCK_PATH(path, port, sockdir) \ diff --git a/src/interfaces/libpq/fe-connect.c b/src/interfaces/libpq/fe-connect.c index 773e9e1f3a..18a07d810d 100644 --- a/src/interfaces/libpq/fe-connect.c +++ b/src/interfaces/libpq/fe-connect.c @@ -123,6 +123,7 @@ static int ldapServiceLookup(const char *purl, PQconninfoOption *options, #define DefaultChannelBinding "disable" #endif #define DefaultTargetSessionAttrs "any" +#define DefaultLoadBalanceHosts "disable" #ifdef USE_SSL #define DefaultSSLMode "prefer" #else @@ -341,6 +342,15 @@ static const internalPQconninfoOption PQconninfoOptions[] = { "Target-Session-Attrs", "", 15, /* sizeof("prefer-standby") = 15 */ offsetof(struct pg_conn, target_session_attrs)}, + {"load_balance_hosts", "PGLOADBALANCEHOSTS", + DefaultLoadBalanceHosts, NULL, + "Load-Balance-Hosts", "", 8, /* sizeof("disable") = 8 */ + offsetof(struct pg_conn, load_balance_hosts)}, + + {"random_seed", NULL, NULL, NULL, + "Random-Seed", "", 10, /* strlen(INT32_MAX) == 10 */ + offsetof(struct pg_conn, randomseed)}, + /* Terminating entry --- MUST BE LAST */ {NULL, NULL, NULL, NULL, NULL, NULL, 0} @@ -379,6 +389,7 @@ static bool fillPGconn(PGconn *conn, PQconninfoOption *connOptions); static void freePGconn(PGconn *conn); static void closePGconn(PGconn *conn); static void release_conn_addrinfo(PGconn *conn); +static bool store_conn_addrinfo(PGconn *conn, struct addrinfo *addrlist); static void sendTerminateConn(PGconn *conn); static PQconninfoOption *conninfo_init(PQExpBuffer errorMessage); static PQconninfoOption *parse_connection_string(const char *connstr, @@ -424,6 +435,8 @@ static void pgpassfileWarning(PGconn *conn); static void default_threadlock(int acquire); static bool sslVerifyProtocolVersion(const char *version); static bool sslVerifyProtocolRange(const char *min, const char *max); +static bool parse_int_param(const char *value, int *result, PGconn *conn, + const char *context); /* global variable because fe-auth.c needs to access it */ @@ -1007,6 +1020,44 @@ parse_comma_separated_list(char **startptr, bool *more) return p; } +/* + * Initializes the prng_state field of the connection. We want something + * unpredictable, so if possible, use high-quality random bits for the + * seed. Otherwise, fall back to a seed based on timestamp and PID. + */ +static bool +libpq_prng_init(PGconn *conn) +{ + if (unlikely(conn->randomseed)) + { + int rseed; + + if (!parse_int_param(conn->randomseed, &rseed, conn, "random_seed")) + return false; + + pg_prng_seed(&conn->prng_state, rseed); + } + else if (unlikely(!pg_prng_strong_seed(&conn->prng_state))) + { + uint64 rseed; + time_t now = time(NULL); + + /* + * Since PIDs and timestamps tend to change more frequently in their + * least significant bits, shift the timestamp left to allow a larger + * total number of seeds in a given time period. Since that would + * leave only 20 bits of the timestamp that cycle every ~1 second, + * also mix in some higher bits. + */ + rseed = ((uint64) getpid()) ^ + ((uint64) now << 12) ^ + ((uint64) now >> 20); + + pg_prng_seed(&conn->prng_state, rseed); + } + return true; +} + /* * connectOptions2 * @@ -1400,6 +1451,47 @@ connectOptions2(PGconn *conn) else conn->target_server_type = SERVER_TYPE_ANY; + /* + * validate load_balance_hosts option, and set load_balance_type + */ + if (conn->load_balance_hosts) + { + if (strcmp(conn->load_balance_hosts, "disable") == 0) + conn->load_balance_type = LOAD_BALANCE_DISABLE; + else if (strcmp(conn->load_balance_hosts, "random") == 0) + conn->load_balance_type = LOAD_BALANCE_RANDOM; + else + { + conn->status = CONNECTION_BAD; + libpq_append_conn_error(conn, "invalid %s value: \"%s\"", + "load_balance_hosts", + conn->load_balance_hosts); + return false; + } + } + else + conn->load_balance_type = LOAD_BALANCE_DISABLE; + + if (conn->load_balance_type == LOAD_BALANCE_RANDOM) + { + if (!libpq_prng_init(conn)) + return false; + + /* + * Shuffle connhost with a Durstenfeld/Knuth version of the + * Fisher-Yates shuffle. Source: + * https://en.wikipedia.org/wiki/Fisher%E2%80%93Yates_shuffle#The_modern_algorithm + */ + for (i = conn->nconnhost - 1; i > 0; i--) + { + int j = pg_prng_uint64_range(&conn->prng_state, 0, i); + pg_conn_host temp = conn->connhost[j]; + + conn->connhost[j] = conn->connhost[i]; + conn->connhost[i] = temp; + } + } + /* * Resolve special "auto" client_encoding from the locale */ @@ -2077,7 +2169,7 @@ connectDBComplete(PGconn *conn) time_t finish_time = ((time_t) -1); int timeout = 0; int last_whichhost = -2; /* certainly different from whichhost */ - struct addrinfo *last_addr_cur = NULL; + int last_whichaddr = -2; /* certainly different from whichaddr */ if (conn == NULL || conn->status == CONNECTION_BAD) return 0; @@ -2121,11 +2213,11 @@ connectDBComplete(PGconn *conn) if (flag != PGRES_POLLING_OK && timeout > 0 && (conn->whichhost != last_whichhost || - conn->addr_cur != last_addr_cur)) + conn->whichaddr != last_whichaddr)) { finish_time = time(NULL) + timeout; last_whichhost = conn->whichhost; - last_addr_cur = conn->addr_cur; + last_whichaddr = conn->whichaddr; } /* @@ -2272,9 +2364,9 @@ keep_going: /* We will come back to here until there is /* Time to advance to next address, or next host if no more addresses? */ if (conn->try_next_addr) { - if (conn->addr_cur && conn->addr_cur->ai_next) + if (conn->whichaddr < conn->naddr) { - conn->addr_cur = conn->addr_cur->ai_next; + conn->whichaddr++; reset_connection_state_machine = true; } else @@ -2287,6 +2379,7 @@ keep_going: /* We will come back to here until there is { pg_conn_host *ch; struct addrinfo hint; + struct addrinfo *addrlist; int thisport; int ret; char portstr[MAXPGPATH]; @@ -2327,7 +2420,7 @@ keep_going: /* We will come back to here until there is /* Initialize hint structure */ MemSet(&hint, 0, sizeof(hint)); hint.ai_socktype = SOCK_STREAM; - conn->addrlist_family = hint.ai_family = AF_UNSPEC; + hint.ai_family = AF_UNSPEC; /* Figure out the port number we're going to use. */ if (ch->port == NULL || ch->port[0] == '\0') @@ -2350,8 +2443,8 @@ keep_going: /* We will come back to here until there is { case CHT_HOST_NAME: ret = pg_getaddrinfo_all(ch->host, portstr, &hint, - &conn->addrlist); - if (ret || !conn->addrlist) + &addrlist); + if (ret || !addrlist) { libpq_append_conn_error(conn, "could not translate host name \"%s\" to address: %s", ch->host, gai_strerror(ret)); @@ -2362,8 +2455,8 @@ keep_going: /* We will come back to here until there is case CHT_HOST_ADDRESS: hint.ai_flags = AI_NUMERICHOST; ret = pg_getaddrinfo_all(ch->hostaddr, portstr, &hint, - &conn->addrlist); - if (ret || !conn->addrlist) + &addrlist); + if (ret || !addrlist) { libpq_append_conn_error(conn, "could not parse network address \"%s\": %s", ch->hostaddr, gai_strerror(ret)); @@ -2372,7 +2465,7 @@ keep_going: /* We will come back to here until there is break; case CHT_UNIX_SOCKET: - conn->addrlist_family = hint.ai_family = AF_UNIX; + hint.ai_family = AF_UNIX; UNIXSOCK_PATH(portstr, thisport, ch->host); if (strlen(portstr) >= UNIXSOCK_PATH_BUFLEN) { @@ -2387,8 +2480,8 @@ keep_going: /* We will come back to here until there is * name as a Unix-domain socket path. */ ret = pg_getaddrinfo_all(NULL, portstr, &hint, - &conn->addrlist); - if (ret || !conn->addrlist) + &addrlist); + if (ret || !addrlist) { libpq_append_conn_error(conn, "could not translate Unix-domain socket path \"%s\" to address: %s", portstr, gai_strerror(ret)); @@ -2397,8 +2490,14 @@ keep_going: /* We will come back to here until there is break; } - /* OK, scan this addrlist for a working server address */ - conn->addr_cur = conn->addrlist; + if (!store_conn_addrinfo(conn, addrlist)) + { + pg_freeaddrinfo_all(hint.ai_family, addrlist); + libpq_append_conn_error(conn, "out of memory"); + goto error_return; + } + pg_freeaddrinfo_all(hint.ai_family, addrlist); + reset_connection_state_machine = true; conn->try_next_host = false; } @@ -2455,30 +2554,29 @@ keep_going: /* We will come back to here until there is { /* * Try to initiate a connection to one of the addresses - * returned by pg_getaddrinfo_all(). conn->addr_cur is the + * returned by pg_getaddrinfo_all(). conn->whichaddr is the * next one to try. * * The extra level of braces here is historical. It's not * worth reindenting this whole switch case to remove 'em. */ { - struct addrinfo *addr_cur = conn->addr_cur; char host_addr[NI_MAXHOST]; + AddrInfo *addr_cur; /* * Advance to next possible host, if we've tried all of * the addresses for the current host. */ - if (addr_cur == NULL) + if (conn->whichaddr == conn->naddr) { conn->try_next_host = true; goto keep_going; } + addr_cur = &conn->addr[conn->whichaddr]; /* Remember current address for possible use later */ - memcpy(&conn->raddr.addr, addr_cur->ai_addr, - addr_cur->ai_addrlen); - conn->raddr.salen = addr_cur->ai_addrlen; + memcpy(&conn->raddr, &addr_cur->addr, sizeof(SockAddr)); /* * Set connip, too. Note we purposely ignore strdup @@ -2494,7 +2592,7 @@ keep_going: /* We will come back to here until there is conn->connip = strdup(host_addr); /* Try to create the socket */ - conn->sock = socket(addr_cur->ai_family, SOCK_STREAM, 0); + conn->sock = socket(addr_cur->family, SOCK_STREAM, 0); if (conn->sock == PGINVALID_SOCKET) { int errorno = SOCK_ERRNO; @@ -2505,7 +2603,7 @@ keep_going: /* We will come back to here until there is * cases where the address list includes both IPv4 and * IPv6 but kernel only accepts one family. */ - if (addr_cur->ai_next != NULL || + if (conn->whichaddr < conn->naddr || conn->whichhost + 1 < conn->nconnhost) { conn->try_next_addr = true; @@ -2531,7 +2629,7 @@ keep_going: /* We will come back to here until there is * TCP sockets, nonblock mode, close-on-exec. Try the * next address if any of this fails. */ - if (addr_cur->ai_family != AF_UNIX) + if (addr_cur->family != AF_UNIX) { if (!connectNoDelay(conn)) { @@ -2558,7 +2656,7 @@ keep_going: /* We will come back to here until there is } #endif /* F_SETFD */ - if (addr_cur->ai_family != AF_UNIX) + if (addr_cur->family != AF_UNIX) { #ifndef WIN32 int on = 1; @@ -2650,8 +2748,8 @@ keep_going: /* We will come back to here until there is * Start/make connection. This should not block, since we * are in nonblock mode. If it does, well, too bad. */ - if (connect(conn->sock, addr_cur->ai_addr, - addr_cur->ai_addrlen) < 0) + if (connect(conn->sock, (struct sockaddr *) &addr_cur->addr.addr, + addr_cur->addr.salen) < 0) { if (SOCK_ERRNO == EINPROGRESS || #ifdef WIN32 @@ -4035,12 +4133,68 @@ freePGconn(PGconn *conn) free(conn->outBuffer); free(conn->rowBuf); free(conn->target_session_attrs); + free(conn->load_balance_hosts); + free(conn->randomseed); termPQExpBuffer(&conn->errorMessage); termPQExpBuffer(&conn->workBuffer); free(conn); } +/* + * Copies over the AddrInfos from addrlist to the PGconn. + */ +static bool +store_conn_addrinfo(PGconn *conn, struct addrinfo *addrlist) +{ + struct addrinfo *ai = addrlist; + + conn->whichaddr = 0; + + conn->naddr = 0; + while (ai) + { + ai = ai->ai_next; + conn->naddr++; + } + + conn->addr = calloc(conn->naddr, sizeof(AddrInfo)); + if (conn->addr == NULL) + return false; + + ai = addrlist; + for (int i = 0; i < conn->naddr; i++) + { + conn->addr[i].family = ai->ai_family; + + memcpy(&conn->addr[i].addr.addr, ai->ai_addr, + ai->ai_addrlen); + conn->addr[i].addr.salen = ai->ai_addrlen; + ai = ai->ai_next; + } + + if (conn->load_balance_type == LOAD_BALANCE_RANDOM) + { + /* + * Shuffle addr with a Durstenfeld/Knuth version of the Fisher-Yates + * shuffle. Source: + * https://en.wikipedia.org/wiki/Fisher%E2%80%93Yates_shuffle#The_modern_algorithm + * + * We don't need to initialize conn->prng_state here, because that + * already happened in connectOptions2. + */ + for (int i = conn->naddr - 1; i > 0; i--) + { + int j = pg_prng_uint64_range(&conn->prng_state, 0, i); + AddrInfo temp = conn->addr[j]; + + conn->addr[j] = conn->addr[i]; + conn->addr[i] = temp; + } + } + return true; +} + /* * release_conn_addrinfo * - Free any addrinfo list in the PGconn. @@ -4048,11 +4202,10 @@ freePGconn(PGconn *conn) static void release_conn_addrinfo(PGconn *conn) { - if (conn->addrlist) + if (conn->addr) { - pg_freeaddrinfo_all(conn->addrlist_family, conn->addrlist); - conn->addrlist = NULL; - conn->addr_cur = NULL; /* for safety */ + free(conn->addr); + conn->addr = NULL; } } diff --git a/src/interfaces/libpq/libpq-int.h b/src/interfaces/libpq/libpq-int.h index 712d572373..86dd1d6d40 100644 --- a/src/interfaces/libpq/libpq-int.h +++ b/src/interfaces/libpq/libpq-int.h @@ -82,6 +82,8 @@ typedef struct #endif #endif /* USE_OPENSSL */ +#include "common/pg_prng.h" + /* * POSTGRES backend dependent Constants. */ @@ -242,6 +244,13 @@ typedef enum SERVER_TYPE_PREFER_STANDBY_PASS2 /* second pass - behaves same as ANY */ } PGTargetServerType; +/* Target server type (decoded value of load_balance_hosts) */ +typedef enum +{ + LOAD_BALANCE_DISABLE = 0, /* Use the existing host order (default) */ + LOAD_BALANCE_RANDOM, /* Read-write server */ +} PGLoadBalanceType; + /* Boolean value plus a not-known state, for GUCs we might have to fetch */ typedef enum { @@ -396,6 +405,8 @@ struct pg_conn char *ssl_min_protocol_version; /* minimum TLS protocol version */ char *ssl_max_protocol_version; /* maximum TLS protocol version */ char *target_session_attrs; /* desired session properties */ + char *load_balance_hosts; /* load balance over hosts */ + char *randomseed; /* seed for randomization of load balancing */ /* Optional file to write trace info to */ FILE *Pfdebug; @@ -459,10 +470,14 @@ struct pg_conn /* Transient state needed while establishing connection */ PGTargetServerType target_server_type; /* desired session properties */ + PGLoadBalanceType load_balance_type; /* desired load balancing + * algorithm */ bool try_next_addr; /* time to advance to next address/host? */ bool try_next_host; /* time to advance to next connhost[]? */ - struct addrinfo *addrlist; /* list of addresses for current connhost */ - struct addrinfo *addr_cur; /* the one currently being tried */ + int naddr; /* number of addresses returned by getaddrinfo */ + int whichaddr; /* the address currently being tried */ + AddrInfo *addr; /* the array of addresses for the currently + * tried host */ int addrlist_family; /* needed to know how to free addrlist */ bool send_appname; /* okay to send application_name? */ @@ -477,6 +492,8 @@ struct pg_conn PGVerbosity verbosity; /* error/notice message verbosity */ PGContextVisibility show_context; /* whether to show CONTEXT field */ PGlobjfuncs *lobjfuncs; /* private state for large-object access fns */ + pg_prng_state prng_state; /* prng state for load balancing connections */ + /* Buffer for data received from backend and not yet processed */ char *inBuffer; /* currently allocated buffer */ diff --git a/src/interfaces/libpq/meson.build b/src/interfaces/libpq/meson.build index 573fd9b6ea..52b327500c 100644 --- a/src/interfaces/libpq/meson.build +++ b/src/interfaces/libpq/meson.build @@ -116,6 +116,7 @@ tests += { 'tests': [ 't/001_uri.pl', 't/002_api.pl', + 't/003_loadbalance.pl', ], 'env': {'with_ssl': get_option('ssl')}, }, diff --git a/src/interfaces/libpq/t/003_loadbalance.pl b/src/interfaces/libpq/t/003_loadbalance.pl new file mode 100644 index 0000000000..1b4e8fd813 --- /dev/null +++ b/src/interfaces/libpq/t/003_loadbalance.pl @@ -0,0 +1,167 @@ +# Copyright (c) 2023, PostgreSQL Global Development Group +use strict; +use warnings; +use Config; +use PostgreSQL::Test::Utils; +use PostgreSQL::Test::Cluster; +use File::Spec::Functions 'catfile'; +use Test::More; + +# This tests two different methods of load balancing from libpq +# 1. Load balancing by providing multiple host and port combinations in the +# libpq connection string. +# 2. By using a hosts file where hostname maps to multiple different IP +# addresses. Regular Postgres users wouldn't usually use such a host file, +# but this is the easiest way to immitate behaviour of a DNS server that +# returns multiple IP addresses for the same DNS record. +# +# Testing method 1 is supported on all platforms and works out of the box. But +# testing method 2 has some more requirements, both on the platform and on the +# initial setup. If any of these requirements are not met, then method 2 is +# simply not tested. +# +# The requirements to test method 2 are as follows: +# 1. Windows or Linux should be used. +# 2. The OS hosts file at /etc/hosts or c:\Windows\System32\Drivers\etc\hosts +# should contain the following contents: +# +# 127.0.0.1 pg-loadbalancetest +# 127.0.0.2 pg-loadbalancetest +# 127.0.0.3 pg-loadbalancetest +# +# +# Windows or Linux are required to test method 2 because these OSes allow +# binding to 127.0.0.2 and 127.0.0.3 addresess by default, but other OSes +# don't. We need to bind to different IP addresses, so that we can use these +# different IP addresses in the hosts file. +# +# The hosts file needs to be prepared before running this test. We don't do it +# on the fly, because it requires root permissions to change the hosts file. In +# CI we set up the previously mentioned rules in the hosts file, so that this +# load balancing method is tested. + + +# Cluster setup which is shared for testing both load balancing methods +my $can_bind_to_127_0_0_2 = $Config{osname} eq 'linux' || $PostgreSQL::Test::Utils::windows_os; + +if ($can_bind_to_127_0_0_2) +{ + $PostgreSQL::Test::Cluster::use_tcp = 1; + $PostgreSQL::Test::Cluster::test_pghost = '127.0.0.1'; +} +my $port = PostgreSQL::Test::Cluster::get_free_port(); +my $node1 = PostgreSQL::Test::Cluster->new('node1', port => $port); +my $node2 = PostgreSQL::Test::Cluster->new('node2', port => $port, own_host => 1); +my $node3 = PostgreSQL::Test::Cluster->new('node3', port => $port, own_host => 1); + +# Create a data directory with initdb +$node1->init(); +$node2->init(); +$node3->init(); + +# Start the PostgreSQL server +$node1->start(); +$node2->start(); +$node3->start(); + +# Start the tests for load balancing method 1 +my $hostlist = $node1->host . ',' . $node2->host . ',' . $node3->host; +my $portlist = "$port,$port,$port"; + +$node1->connect_ok("host=$hostlist port=$portlist load_balance_hosts=random random_seed=123", + "seed 123 selects node 1 first", + sql => "SELECT 'connect1'", + log_like => [qr/statement: SELECT 'connect1'/]); + +$node2->connect_ok("host=$hostlist port=$portlist load_balance_hosts=random random_seed=123", + "seed 123 does not select node 2 first", + sql => "SELECT 'connect1'", + log_unlike => [qr/statement: SELECT 'connect1'/]); + +$node3->connect_ok("host=$hostlist port=$portlist load_balance_hosts=random random_seed=123", + "seed 123 does not select node 3 first", + sql => "SELECT 'connect1'", + log_unlike => [qr/statement: SELECT 'connect1'/]); + +$node3->connect_ok("host=$hostlist port=$portlist load_balance_hosts=random random_seed=42", + "seed 42 selects node 3 first", + sql => "SELECT 'connect2'", + log_like => [qr/statement: SELECT 'connect2'/]); + +$node1->connect_ok("host=$hostlist port=$portlist load_balance_hosts=random random_seed=42", + "seed 42 does not select node 1 first", + sql => "SELECT 'connect2'", + log_unlike => [qr/statement: SELECT 'connect2'/]); + +$node2->connect_ok("host=$hostlist port=$portlist load_balance_hosts=random random_seed=42", + "seed 42 does not select node 2 first", + sql => "SELECT 'connect2'", + log_unlike => [qr/statement: SELECT 'connect2'/]); + +$node3->stop(); + +$node1->connect_ok("host=$hostlist port=$portlist load_balance_hosts=random random_seed=42", + "seed 42 does select node 1 second", + sql => "SELECT 'connect3'", + log_like => [qr/statement: SELECT 'connect3'/]); + +$node2->connect_ok("host=$hostlist port=$portlist load_balance_hosts=random random_seed=42", + "seed 42 does not select node 2 second", + sql => "SELECT 'connect3'", + log_unlike => [qr/statement: SELECT 'connect3'/]); + +$node3->start(); + +# Checks for the requirements for testing load balancing method 2 +if (!$can_bind_to_127_0_0_2) { + # The OS requirement is not met + done_testing(); + exit; +} + +my $hosts_path; +if ($windows_os) { + $hosts_path = 'c:\Windows\System32\Drivers\etc\hosts'; +} +else +{ + $hosts_path = '/etc/hosts'; +} + +my $hosts_content = PostgreSQL::Test::Utils::slurp_file($hosts_path); + +if ($hosts_content !~ m/pg-loadbalancetest/) { + # Host file is not prepared for this test + done_testing(); + exit; +} + +# Start the tests for load balancing method 2 +$node2->connect_ok("host=pg-loadbalancetest port=$port load_balance_hosts=random random_seed=44", + "seed 44 selects node 2 first", + sql => "SELECT 'connect4'", + log_like => [qr/statement: SELECT 'connect4'/]); + +$node1->connect_ok("host=pg-loadbalancetest port=$port load_balance_hosts=random random_seed=44", + "seed 44 does not select node 1 first", + sql => "SELECT 'connect4'", + log_unlike => [qr/statement: SELECT 'connect4'/]); + +$node3->connect_ok("host=pg-loadbalancetest port=$port load_balance_hosts=random random_seed=44", + "seed 44 does not select node 3 first", + sql => "SELECT 'connect4'", + log_unlike => [qr/statement: SELECT 'connect4'/]); + +$node2->stop(); + +$node1->connect_ok("host=pg-loadbalancetest port=$port load_balance_hosts=random random_seed=44", + "seed 44 does select node 1 second", + sql => "SELECT 'connect5'", + log_like => [qr/statement: SELECT 'connect5'/]); + +$node3->connect_ok("host=pg-loadbalancetest port=$port load_balance_hosts=random random_seed=44", + "seed 44 does not select node 3 second", + sql => "SELECT 'connect5'", + log_unlike => [qr/statement: SELECT 'connect5'/]); + +done_testing(); -- 2.34.1 ^ permalink raw reply [nested|flat] 3+ messages in thread
end of thread, other threads:[~2023-01-09 13:00 UTC | newest] Thread overview: 3+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2020-09-17 15:26 [PATCH 2/8] Move IS [NOT] NULL handling from BRIN support functions Tomas Vondra <[email protected]> 2023-01-06 17:21 Re: Support load balancing in libpq Michael Banck <[email protected]> 2023-01-09 13:00 ` Re: [EXTERNAL] Re: Support load balancing in libpq Jelte Fennema <[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