agora inbox for [email protected]help / color / mirror / Atom feed
[PATCH] Remove the NODROP SLOT option from DROP SUBSCRIPTION 27+ messages / 4 participants [nested] [flat]
* [PATCH] Remove the NODROP SLOT option from DROP SUBSCRIPTION @ 2017-05-05 16:14 Petr Jelinek <[email protected]> 0 siblings, 0 replies; 27+ messages in thread From: Petr Jelinek @ 2017-05-05 16:14 UTC (permalink / raw) --- doc/src/sgml/ref/create_subscription.sgml | 7 + doc/src/sgml/ref/drop_subscription.sgml | 22 +-- src/backend/catalog/pg_subscription.c | 9 +- src/backend/commands/subscriptioncmds.c | 195 ++++++++++++++++----- src/backend/nodes/copyfuncs.c | 2 +- src/backend/nodes/equalfuncs.c | 2 +- src/backend/parser/gram.y | 43 ++--- src/backend/replication/logical/worker.c | 35 ++-- src/backend/tcop/utility.c | 3 + src/include/catalog/pg_subscription.h | 2 +- src/include/nodes/parsenodes.h | 2 +- .../dummy_seclabel/expected/dummy_seclabel.out | 4 +- .../modules/dummy_seclabel/sql/dummy_seclabel.sql | 4 +- src/test/regress/expected/object_address.out | 4 +- src/test/regress/expected/subscription.out | 13 +- src/test/regress/sql/object_address.sql | 4 +- src/test/regress/sql/subscription.sql | 12 +- src/test/subscription/t/001_rep_changes.pl | 2 +- src/test/subscription/t/004_sync.pl | 8 +- 19 files changed, 241 insertions(+), 132 deletions(-) diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml index fcec254..c9256e7 100644 --- a/doc/src/sgml/ref/create_subscription.sgml +++ b/doc/src/sgml/ref/create_subscription.sgml @@ -129,6 +129,13 @@ CREATE SUBSCRIPTION <replaceable class="PARAMETER">subscription_name</replaceabl Name of the replication slot to use. The default behavior is to use <literal>subscription_name</> for slot name. </para> + <para> + When <literal>slot_name</literal> is set to + <literal>NONE</literal> there will be no slot associated with the + subscription. Such subscriptions must also have both + <literal>enabled</literal> and <literal>create_slot</literal> set + to <literal>false</literal>. + </para> </listitem> </varlistentry> diff --git a/doc/src/sgml/ref/drop_subscription.sgml b/doc/src/sgml/ref/drop_subscription.sgml index f1ac125..8b00c45 100644 --- a/doc/src/sgml/ref/drop_subscription.sgml +++ b/doc/src/sgml/ref/drop_subscription.sgml @@ -21,7 +21,7 @@ PostgreSQL documentation <refsynopsisdiv> <synopsis> -DROP SUBSCRIPTION [ IF EXISTS ] <replaceable class="parameter">name</replaceable> [ DROP SLOT | NODROP SLOT ] +DROP SUBSCRIPTION [ IF EXISTS ] <replaceable class="parameter">name</replaceable> [ CASCADE | RESTRICT ] </synopsis> </refsynopsisdiv> @@ -57,20 +57,16 @@ DROP SUBSCRIPTION [ IF EXISTS ] <replaceable class="parameter">name</replaceable </varlistentry> <varlistentry> - <term><literal>DROP SLOT</literal></term> - <term><literal>NODROP SLOT</literal></term> - <listitem> - <para> - Specifies whether to drop the replication slot on the publisher. The - default is - <literal>DROP SLOT</literal>. - </para> + <term><literal>CASCADE</literal></term> + <term><literal>RESTRICT</literal></term> + <listitem> <para> - If the publisher is not reachable when the subscription is to be - dropped, then it is useful to specify <literal>NODROP SLOT</literal>. - But the replication slot on the publisher will then have to be removed - manually. + These key words are used to determine what to do with when there is a + replication slot associated with the subscription. The + <literal>RESTRICT</literal> will refuse to drop the subscription in + such case, while <literal>CASCADE</literal> will drop the associated + slot. <literal>RESTRICT</literal> is the default. </para> </listitem> </varlistentry> diff --git a/src/backend/catalog/pg_subscription.c b/src/backend/catalog/pg_subscription.c index 22587a4..7dc21f1 100644 --- a/src/backend/catalog/pg_subscription.c +++ b/src/backend/catalog/pg_subscription.c @@ -82,8 +82,10 @@ GetSubscription(Oid subid, bool missing_ok) tup, Anum_pg_subscription_subslotname, &isnull); - Assert(!isnull); - sub->slotname = pstrdup(NameStr(*DatumGetName(datum))); + if (!isnull) + sub->slotname = pstrdup(NameStr(*DatumGetName(datum))); + else + sub->slotname = NULL; /* Get synccommit */ datum = SysCacheGetAttr(SUBSCRIPTIONOID, @@ -147,7 +149,8 @@ FreeSubscription(Subscription *sub) { pfree(sub->name); pfree(sub->conninfo); - pfree(sub->slotname); + if (sub->slotname) + pfree(sub->slotname); list_free_deep(sub->publications); pfree(sub); } diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index c00981e..afb116a 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -60,7 +60,8 @@ static List *fetch_table_list(WalReceiverConn *wrconn, List *publications); */ static void parse_subscription_options(List *options, bool *connect, bool *enabled_given, - bool *enabled, bool *create_slot, char **slot_name, + bool *enabled, bool *create_slot, + bool *slot_name_given, char **slot_name, bool *copy_data, char **synchronous_commit) { ListCell *lc; @@ -78,7 +79,10 @@ parse_subscription_options(List *options, bool *connect, bool *enabled_given, if (create_slot) *create_slot = true; if (slot_name) + { + *slot_name_given = false; *slot_name = NULL; + } if (copy_data) *copy_data = true; if (synchronous_commit) @@ -121,12 +125,17 @@ parse_subscription_options(List *options, bool *connect, bool *enabled_given, } else if (strcmp(defel->defname, "slot_name") == 0 && slot_name) { - if (*slot_name) + if (*slot_name_given) ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), errmsg("conflicting or redundant options"))); + *slot_name_given = true; *slot_name = defGetString(defel); + + /* Setting slot_name = NONE is treated as no slot name. */ + if (strcmp(*slot_name, "none") == 0) + *slot_name = NULL; } else if (strcmp(defel->defname, "copy_data") == 0 && copy_data) { @@ -164,26 +173,43 @@ parse_subscription_options(List *options, bool *connect, bool *enabled_given, if (connect && !*connect) { /* Check for incompatible options from the user. */ - if (*enabled_given && *enabled) + if (enabled && *enabled_given && *enabled) ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), - errmsg("noconnect and enabled are mutually exclusive options"))); + errmsg("connect = false and enabled are mutually exclusive options"))); - if (create_slot_given && *create_slot) + if (create_slot && create_slot_given && *create_slot) ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), - errmsg("noconnect and create slot are mutually exclusive options"))); + errmsg("connect = false and create_slot are mutually exclusive options"))); - if (copy_data_given && *copy_data) + if (copy_data && copy_data_given && *copy_data) ereport(ERROR, (errcode(ERRCODE_SYNTAX_ERROR), - errmsg("noconnect and copy data are mutually exclusive options"))); + errmsg("connect = false and copy_data are mutually exclusive options"))); /* Change the defaults of other options. */ *enabled = false; *create_slot = false; *copy_data = false; } + + /* + * Do additional checking for disallowed combination when + * slot_name = NONE was used. + */ + if (slot_name && *slot_name_given && !*slot_name) + { + if (enabled && *enabled_given && *enabled) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("slot_name = NONE and enabled are mutually exclusive options"))); + + if (create_slot && create_slot_given && *create_slot) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("slot_name = NONE and create_slot are mutually exclusive options"))); + } } /* @@ -260,6 +286,7 @@ CreateSubscription(CreateSubscriptionStmt *stmt, bool isTopLevel) char *synchronous_commit; char *conninfo; char *slotname; + bool slotname_given; char originname[NAMEDATALEN]; bool create_slot; List *publications; @@ -269,8 +296,8 @@ CreateSubscription(CreateSubscriptionStmt *stmt, bool isTopLevel) * Connection and publication should not be specified here. */ parse_subscription_options(stmt->options, &connect, &enabled_given, - &enabled, &create_slot, &slotname, ©_data, - &synchronous_commit); + &enabled, &create_slot, &slotname_given, + &slotname, ©_data, &synchronous_commit); /* * Since creating a replication slot is not transactional, rolling back @@ -299,8 +326,9 @@ CreateSubscription(CreateSubscriptionStmt *stmt, bool isTopLevel) stmt->subname))); } - if (slotname == NULL) + if (!slotname_given && slotname == NULL) slotname = stmt->subname; + /* The default for synchronous_commit of subscriptions is off. */ if (synchronous_commit == NULL) synchronous_commit = "off"; @@ -325,8 +353,11 @@ CreateSubscription(CreateSubscriptionStmt *stmt, bool isTopLevel) values[Anum_pg_subscription_subenabled - 1] = BoolGetDatum(enabled); values[Anum_pg_subscription_subconninfo - 1] = CStringGetTextDatum(conninfo); - values[Anum_pg_subscription_subslotname - 1] = - DirectFunctionCall1(namein, CStringGetDatum(slotname)); + if (slotname) + values[Anum_pg_subscription_subslotname - 1] = + DirectFunctionCall1(namein, CStringGetDatum(slotname)); + else + nulls[Anum_pg_subscription_subslotname - 1] = true; values[Anum_pg_subscription_subsynccommit - 1] = CStringGetTextDatum(synchronous_commit); values[Anum_pg_subscription_subpublications - 1] = @@ -396,6 +427,8 @@ CreateSubscription(CreateSubscriptionStmt *stmt, bool isTopLevel) */ if (create_slot) { + Assert(slotname); + walrcv_create_slot(wrconn, slotname, false, CRS_NOEXPORT_SNAPSHOT, &lsn); ereport(NOTICE, @@ -548,6 +581,7 @@ AlterSubscription(AlterSubscriptionStmt *stmt) HeapTuple tup; Oid subid; bool update_tuple = false; + Subscription *sub; rel = heap_open(SubscriptionRelationId, RowExclusiveLock); @@ -567,6 +601,7 @@ AlterSubscription(AlterSubscriptionStmt *stmt) stmt->subname); subid = HeapTupleGetOid(tup); + sub = GetSubscription(subid, false); /* Form a new tuple. */ memset(values, 0, sizeof(values)); @@ -577,19 +612,29 @@ AlterSubscription(AlterSubscriptionStmt *stmt) { case ALTER_SUBSCRIPTION_OPTIONS: { - char *slot_name; - char *synchronous_commit; + char *slotname; + bool slotname_given; + char *synchronous_commit; parse_subscription_options(stmt->options, NULL, NULL, NULL, - NULL, &slot_name, NULL, - &synchronous_commit); + NULL, &slotname_given, &slotname, + NULL, &synchronous_commit); - if (slot_name) + if (slotname_given) { - values[Anum_pg_subscription_subslotname - 1] = - DirectFunctionCall1(namein, CStringGetDatum(slot_name)); + if (sub->enabled && !slotname) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("cannot set slot_name = NONE for enabled subscription"))); + + if (slotname) + values[Anum_pg_subscription_subslotname - 1] = + DirectFunctionCall1(namein, CStringGetDatum(slotname)); + else + nulls[Anum_pg_subscription_subslotname - 1] = true; replaces[Anum_pg_subscription_subslotname - 1] = true; } + if (synchronous_commit) { values[Anum_pg_subscription_subsynccommit - 1] = @@ -608,9 +653,14 @@ AlterSubscription(AlterSubscriptionStmt *stmt) parse_subscription_options(stmt->options, NULL, &enabled_given, &enabled, NULL, - NULL, NULL, NULL); + NULL, NULL, NULL, NULL); Assert(enabled_given); + if (!sub->slotname && enabled) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("cannot enable subscription which does not have a slot_name"))); + values[Anum_pg_subscription_subenabled - 1] = BoolGetDatum(enabled); replaces[Anum_pg_subscription_subenabled - 1] = true; @@ -633,10 +683,10 @@ AlterSubscription(AlterSubscriptionStmt *stmt) case ALTER_SUBSCRIPTION_PUBLICATION_REFRESH: { bool copy_data; - Subscription *sub = GetSubscription(subid, false); parse_subscription_options(stmt->options, NULL, NULL, NULL, - NULL, NULL, ©_data, NULL); + NULL, NULL, NULL, ©_data, + NULL); values[Anum_pg_subscription_subpublications - 1] = publicationListToArray(stmt->publication); @@ -647,6 +697,11 @@ AlterSubscription(AlterSubscriptionStmt *stmt) /* Refresh if user asked us to. */ if (stmt->kind == ALTER_SUBSCRIPTION_PUBLICATION_REFRESH) { + if (!sub->enabled) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("ALTER SUBSCRIPTION ... REFRESH is not allowed for disabled subscriptions"))); + /* Make sure refresh sees the new list of publications. */ sub->publications = stmt->publication; @@ -659,10 +714,15 @@ AlterSubscription(AlterSubscriptionStmt *stmt) case ALTER_SUBSCRIPTION_REFRESH: { bool copy_data; - Subscription *sub = GetSubscription(subid, false); + + if (!sub->enabled) + ereport(ERROR, + (errcode(ERRCODE_SYNTAX_ERROR), + errmsg("ALTER SUBSCRIPTION ... REFRESH is not allowed for disabled subscriptions"))); parse_subscription_options(stmt->options, NULL, NULL, NULL, - NULL, NULL, ©_data, NULL); + NULL, NULL, NULL, ©_data, + NULL); AlterSubscription_refresh(sub, copy_data); @@ -721,8 +781,8 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel) * run DROP SUBSCRIPTION inside a transaction block if dropping the * replication slot. */ - if (stmt->drop_slot) - PreventTransactionChain(isTopLevel, "DROP SUBSCRIPTION ... DROP SLOT"); + if (stmt->behavior == DROP_CASCADE) + PreventTransactionChain(isTopLevel, "DROP SUBSCRIPTION ... CASCADE"); /* * Lock pg_subscription with AccessExclusiveLock to ensure @@ -782,8 +842,10 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel) /* Get slotname */ datum = SysCacheGetAttr(SUBSCRIPTIONOID, tup, Anum_pg_subscription_subslotname, &isnull); - Assert(!isnull); - slotname = pstrdup(NameStr(*DatumGetName(datum))); + if (!isnull) + slotname = pstrdup(NameStr(*DatumGetName(datum))); + else + slotname = NULL; ObjectAddressSet(myself, SubscriptionRelationId, subid); EventTriggerSQLDropAddObject(&myself, true, true); @@ -808,43 +870,84 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel) if (originid != InvalidRepOriginId) replorigin_drop(originid); - /* If the user asked to not drop the slot, we are done mow.*/ - if (!stmt->drop_slot) + /* If there is no slot associated with subscription we can finish here. */ + if (!slotname) { heap_close(rel, NoLock); return; } /* - * Otherwise drop the replication slot at the publisher node using + * Otherwise check for the replication slot at the publisher node using * the replication connection. */ load_file("libpqwalreceiver", false); - initStringInfo(&cmd); - appendStringInfo(&cmd, "DROP_REPLICATION_SLOT \"%s\"", slotname); - wrconn = walrcv_connect(conninfo, true, subname, &err); if (wrconn == NULL) ereport(ERROR, - (errmsg("could not connect to publisher when attempting to " - "drop the replication slot \"%s\"", slotname), - errdetail("The error was: %s", err))); + (errmsg("could not connect to publisher when attempting to fetch " + "information about replication slot \"%s\"", slotname), + errdetail("The error was: %s", err), + errhint("Use ALTER SUBSCRIPTION ... WITH (slot_name = NONE) " + "to disassociate the subscription from slot."))); PG_TRY(); { WalRcvExecResult *res; - res = walrcv_exec(wrconn, cmd.data, 0, NULL); - - if (res->status != WALRCV_OK_COMMAND) + TupleTableSlot *tupslot; + Oid slot_row_desc[1] = {BOOLOID}; + bool found; + + initStringInfo(&cmd); + appendStringInfo(&cmd, + "SELECT true FROM pg_catalog.pg_replication_slots WHERE slot_name = %s", + quote_literal_cstr(slotname)); + res = walrcv_exec(wrconn, cmd.data, 1, slot_row_desc); + if (res->status != WALRCV_OK_TUPLES) ereport(ERROR, - (errmsg("could not drop the replication slot \"%s\" on publisher", - slotname), + (errmsg("could not fetch the replication slot info from publisher"), errdetail("The error was: %s", res->err))); + + tupslot = MakeSingleTupleTableSlot(res->tupledesc); + found = tuplestore_gettupleslot(res->tuplestore, true, false, + tupslot); + ExecDropSingleTupleTableSlot(tupslot); + walrcv_clear_result(res); + + /* If slot was not found on publisher, we are done. */ + if (!found) + { + walrcv_disconnect(wrconn); + pfree(cmd.data); + heap_close(rel, NoLock); + return; + } + + /* Otherwise the next action depends on the drop_behavior. */ + if (stmt->behavior == DROP_CASCADE) + { + resetStringInfo(&cmd); + appendStringInfo(&cmd, "DROP_REPLICATION_SLOT %s", + quote_identifier(slotname)); + res = walrcv_exec(wrconn, cmd.data, 0, NULL); + + if (res->status != WALRCV_OK_COMMAND) + ereport(ERROR, + (errmsg("could not drop the replication slot \"%s\" on publisher", + slotname), + errdetail("The error was: %s", res->err))); + else + ereport(NOTICE, + (errmsg("dropped replication slot \"%s\" on publisher", + slotname))); + } else - ereport(NOTICE, - (errmsg("dropped replication slot \"%s\" on publisher", - slotname))); + ereport(ERROR, + (errcode(ERRCODE_DEPENDENT_OBJECTS_STILL_EXIST), + errmsg("cannot drop subscription \"%s\" because there is still replication slot associated with it", + subname), + errhint("Use DROP ... CASCADE to drop the slot too."))); walrcv_clear_result(res); } diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c index 35a237a..2d2a9d0 100644 --- a/src/backend/nodes/copyfuncs.c +++ b/src/backend/nodes/copyfuncs.c @@ -4537,8 +4537,8 @@ _copyDropSubscriptionStmt(const DropSubscriptionStmt *from) DropSubscriptionStmt *newnode = makeNode(DropSubscriptionStmt); COPY_STRING_FIELD(subname); - COPY_SCALAR_FIELD(drop_slot); COPY_SCALAR_FIELD(missing_ok); + COPY_SCALAR_FIELD(behavior); return newnode; } diff --git a/src/backend/nodes/equalfuncs.c b/src/backend/nodes/equalfuncs.c index 21dfbb0..b5459cd 100644 --- a/src/backend/nodes/equalfuncs.c +++ b/src/backend/nodes/equalfuncs.c @@ -2246,8 +2246,8 @@ _equalDropSubscriptionStmt(const DropSubscriptionStmt *a, const DropSubscriptionStmt *b) { COMPARE_STRING_FIELD(subname); - COMPARE_SCALAR_FIELD(drop_slot); COMPARE_SCALAR_FIELD(missing_ok); + COMPARE_SCALAR_FIELD(behavior); return true; } diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y index 732647b..bdd2f6d 100644 --- a/src/backend/parser/gram.y +++ b/src/backend/parser/gram.y @@ -415,7 +415,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query); %type <fun_param_mode> arg_class %type <typnam> func_return func_type -%type <boolean> opt_trusted opt_restart_seqs opt_drop_slot +%type <boolean> opt_trusted opt_restart_seqs %type <ival> OptTemp %type <ival> OptNoLog %type <oncommit> OnCommitOption @@ -467,7 +467,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query); %type <node> def_arg columnElem where_clause where_or_current_clause a_expr b_expr c_expr AexprConst indirection_el opt_slice_bound columnref in_expr having_clause func_table xmltable array_expr - ExclusionWhereClause + ExclusionWhereClause operator_def_arg %type <list> rowsfrom_item rowsfrom_list opt_col_def_list %type <boolean> opt_ordinality %type <list> ExclusionConstraintList ExclusionConstraintElem @@ -5684,6 +5684,7 @@ def_arg: func_type { $$ = (Node *)$1; } | qual_all_Op { $$ = (Node *)$1; } | NumericOnly { $$ = (Node *)$1; } | Sconst { $$ = (Node *)makeString($1); } + | NONE { $$ = (Node *)makeString(pstrdup($1)); } ; old_aggr_definition: '(' old_aggr_list ')' { $$ = $2; } @@ -8923,8 +8924,16 @@ operator_def_list: operator_def_elem { $$ = list_make1($1); } operator_def_elem: ColLabel '=' NONE { $$ = makeDefElem($1, NULL, @1); } - | ColLabel '=' def_arg - { $$ = makeDefElem($1, (Node *) $3, @1); } + | ColLabel '='operator_def_arg + { $$ = makeDefElem($1, (Node *) $3, @1); } + ; + +operator_def_arg: + func_type { $$ = (Node *)$1; } + | reserved_keyword { $$ = (Node *)makeString(pstrdup($1)); } + | qual_all_Op { $$ = (Node *)$1; } + | NumericOnly { $$ = (Node *)$1; } + | Sconst { $$ = (Node *)makeString($1); } ; /***************************************************************************** @@ -9315,42 +9324,24 @@ AlterSubscriptionStmt: * *****************************************************************************/ -DropSubscriptionStmt: DROP SUBSCRIPTION name opt_drop_slot +DropSubscriptionStmt: DROP SUBSCRIPTION name opt_drop_behavior { DropSubscriptionStmt *n = makeNode(DropSubscriptionStmt); n->subname = $3; - n->drop_slot = $4; n->missing_ok = false; + n->behavior = $4; $$ = (Node *) n; } - | DROP SUBSCRIPTION IF_P EXISTS name opt_drop_slot + | DROP SUBSCRIPTION IF_P EXISTS name opt_drop_behavior { DropSubscriptionStmt *n = makeNode(DropSubscriptionStmt); n->subname = $5; - n->drop_slot = $6; n->missing_ok = true; + n->behavior = $6; $$ = (Node *) n; } ; -opt_drop_slot: - DROP SLOT - { - $$ = TRUE; - } - | IDENT SLOT - { - if (strcmp($1, "nodrop") == 0) - $$ = FALSE; - else - ereport(ERROR, - (errcode(ERRCODE_SYNTAX_ERROR), - errmsg("unrecognized option \"%s\"", $1), - parser_errposition(@1))); - } - | /*EMPTY*/ { $$ = TRUE; } - ; - /***************************************************************************** * * QUERY: Define Rewrite Rule diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index 2d7770d..a4b7dcb 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -1329,6 +1329,22 @@ reread_subscription(void) } /* + * Exit if the subscription was disabled. + * This normally should not happen as the worker gets killed + * during ALTER SUBSCRIPTION ... DISABLE. + */ + if (!newsub->enabled) + { + ereport(LOG, + (errmsg("logical replication worker for subscription \"%s\" will " + "stop because the subscription was disabled", + MySubscription->name))); + + walrcv_disconnect(wrconn); + proc_exit(0); + } + + /* * Exit if connection string was changed. The launcher will start * new worker. */ @@ -1358,6 +1374,9 @@ reread_subscription(void) proc_exit(0); } + /* !slotname should never happen when enabled is true. */ + Assert(newsub->slotname); + /* * We need to make new connection to new slot if slot name has changed * so exit here as well if that's the case. @@ -1388,22 +1407,6 @@ reread_subscription(void) proc_exit(0); } - /* - * Exit if the subscription was disabled. - * This normally should not happen as the worker gets killed - * during ALTER SUBSCRIPTION ... DISABLE. - */ - if (!newsub->enabled) - { - ereport(LOG, - (errmsg("logical replication worker for subscription \"%s\" will " - "stop because the subscription was disabled", - MySubscription->name))); - - walrcv_disconnect(wrconn); - proc_exit(0); - } - /* Check for other changes that should never happen too. */ if (newsub->dbid != MySubscription->dbid) { diff --git a/src/backend/tcop/utility.c b/src/backend/tcop/utility.c index 24e5c42..d4fa5a7 100644 --- a/src/backend/tcop/utility.c +++ b/src/backend/tcop/utility.c @@ -2303,6 +2303,9 @@ CreateCommandTag(Node *parsetree) case OBJECT_PUBLICATION: tag = "DROP PUBLICATION"; break; + case OBJECT_SUBSCRIPTION: + tag = "DROP SUBSCRIPTION"; + break; case OBJECT_STATISTIC_EXT: tag = "DROP STATISTICS"; break; diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h index 5550f19..d4f3979 100644 --- a/src/include/catalog/pg_subscription.h +++ b/src/include/catalog/pg_subscription.h @@ -45,7 +45,7 @@ CATALOG(pg_subscription,6100) BKI_SHARED_RELATION BKI_ROWTYPE_OID(6101) BKI_SCHE text subconninfo BKI_FORCE_NOT_NULL; /* Slot name on publisher */ - NameData subslotname BKI_FORCE_NOT_NULL; + NameData subslotname; /* Synchronous commit setting for worker */ text subsynccommit BKI_FORCE_NOT_NULL; diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h index e1d454a..46c23c2 100644 --- a/src/include/nodes/parsenodes.h +++ b/src/include/nodes/parsenodes.h @@ -3393,8 +3393,8 @@ typedef struct DropSubscriptionStmt { NodeTag type; char *subname; /* Name of of the subscription */ - bool drop_slot; /* Should we drop the slot on remote side? */ bool missing_ok; /* Skip error if missing? */ + DropBehavior behavior; /* RESTRICT or CASCADE behavior */ } DropSubscriptionStmt; #endif /* PARSENODES_H */ diff --git a/src/test/modules/dummy_seclabel/expected/dummy_seclabel.out b/src/test/modules/dummy_seclabel/expected/dummy_seclabel.out index f6fc29a..77bdc93 100644 --- a/src/test/modules/dummy_seclabel/expected/dummy_seclabel.out +++ b/src/test/modules/dummy_seclabel/expected/dummy_seclabel.out @@ -69,7 +69,7 @@ CREATE SCHEMA dummy_seclabel_test; SECURITY LABEL ON SCHEMA dummy_seclabel_test IS 'unclassified'; -- OK SET client_min_messages = error; CREATE PUBLICATION dummy_pub; -CREATE SUBSCRIPTION dummy_sub CONNECTION '' PUBLICATION foo WITH (connect = false); +CREATE SUBSCRIPTION dummy_sub CONNECTION '' PUBLICATION foo WITH (connect = false, slot_name = NONE); RESET client_min_messages; SECURITY LABEL ON PUBLICATION dummy_pub IS 'classified'; SECURITY LABEL ON SUBSCRIPTION dummy_sub IS 'classified'; @@ -111,7 +111,7 @@ NOTICE: event ddl_command_end: SECURITY LABEL DROP EVENT TRIGGER always_start, always_end, always_drop, always_rewrite; DROP VIEW dummy_seclabel_view1; DROP TABLE dummy_seclabel_tbl1, dummy_seclabel_tbl2; -DROP SUBSCRIPTION dummy_sub NODROP SLOT; +DROP SUBSCRIPTION dummy_sub; DROP PUBLICATION dummy_pub; DROP ROLE regress_dummy_seclabel_user1; DROP ROLE regress_dummy_seclabel_user2; diff --git a/src/test/modules/dummy_seclabel/sql/dummy_seclabel.sql b/src/test/modules/dummy_seclabel/sql/dummy_seclabel.sql index d7795bd..8c347b6 100644 --- a/src/test/modules/dummy_seclabel/sql/dummy_seclabel.sql +++ b/src/test/modules/dummy_seclabel/sql/dummy_seclabel.sql @@ -73,7 +73,7 @@ SECURITY LABEL ON SCHEMA dummy_seclabel_test IS 'unclassified'; -- OK SET client_min_messages = error; CREATE PUBLICATION dummy_pub; -CREATE SUBSCRIPTION dummy_sub CONNECTION '' PUBLICATION foo WITH (connect = false); +CREATE SUBSCRIPTION dummy_sub CONNECTION '' PUBLICATION foo WITH (connect = false, slot_name = NONE); RESET client_min_messages; SECURITY LABEL ON PUBLICATION dummy_pub IS 'classified'; SECURITY LABEL ON SUBSCRIPTION dummy_sub IS 'classified'; @@ -108,7 +108,7 @@ DROP EVENT TRIGGER always_start, always_end, always_drop, always_rewrite; DROP VIEW dummy_seclabel_view1; DROP TABLE dummy_seclabel_tbl1, dummy_seclabel_tbl2; -DROP SUBSCRIPTION dummy_sub NODROP SLOT; +DROP SUBSCRIPTION dummy_sub; DROP PUBLICATION dummy_pub; DROP ROLE regress_dummy_seclabel_user1; diff --git a/src/test/regress/expected/object_address.out b/src/test/regress/expected/object_address.out index 0849715..700f261 100644 --- a/src/test/regress/expected/object_address.out +++ b/src/test/regress/expected/object_address.out @@ -37,7 +37,7 @@ CREATE TRANSFORM FOR int LANGUAGE SQL ( FROM SQL WITH FUNCTION varchar_transform(internal), TO SQL WITH FUNCTION int4recv(internal)); CREATE PUBLICATION addr_pub FOR TABLE addr_nsp.gentable; -CREATE SUBSCRIPTION addr_sub CONNECTION '' PUBLICATION bar WITH (connect = false); +CREATE SUBSCRIPTION addr_sub CONNECTION '' PUBLICATION bar WITH (connect = false, slot_name = NONE); WARNING: tables were not subscribed, you will have to run ALTER SUBSCRIPTION ... REFRESH PUBLICATION to subscribe the tables CREATE STATISTICS addr_nsp.gentable_stat ON (a,b) FROM addr_nsp.gentable; -- test some error cases @@ -477,7 +477,7 @@ SELECT (pg_identify_object(addr1.classid, addr1.objid, addr1.objsubid)).*, SET client_min_messages TO 'warning'; DROP FOREIGN DATA WRAPPER addr_fdw CASCADE; DROP PUBLICATION addr_pub; -DROP SUBSCRIPTION addr_sub NODROP SLOT; +DROP SUBSCRIPTION addr_sub; DROP SCHEMA addr_nsp CASCADE; DROP OWNED BY regress_addr_user; DROP USER regress_addr_user; diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out index 33d53de..fcf5646 100644 --- a/src/test/regress/expected/subscription.out +++ b/src/test/regress/expected/subscription.out @@ -109,17 +109,18 @@ HINT: The owner of a subscription must be a superuser. ALTER ROLE regress_subscription_user2 SUPERUSER; -- now it works ALTER SUBSCRIPTION testsub OWNER TO regress_subscription_user2; --- fail - cannot do DROP SUBSCRIPTION DROP SLOT inside transaction block +ALTER SUBSCRIPTION testsub WITH (slot_name = NONE); +-- fail - cannot do DROP SUBSCRIPTION CASCADE inside transaction block BEGIN; -DROP SUBSCRIPTION testsub DROP SLOT; -ERROR: DROP SUBSCRIPTION ... DROP SLOT cannot run inside a transaction block +DROP SUBSCRIPTION testsub CASCADE; +ERROR: DROP SUBSCRIPTION ... CASCADE cannot run inside a transaction block COMMIT; BEGIN; -DROP SUBSCRIPTION testsub NODROP SLOT; +DROP SUBSCRIPTION testsub; COMMIT; -DROP SUBSCRIPTION IF EXISTS testsub NODROP SLOT; +DROP SUBSCRIPTION IF EXISTS testsub; NOTICE: subscription "testsub" does not exist, skipping -DROP SUBSCRIPTION testsub NODROP SLOT; -- fail +DROP SUBSCRIPTION testsub; -- fail ERROR: subscription "testsub" does not exist RESET SESSION AUTHORIZATION; DROP ROLE regress_subscription_user; diff --git a/src/test/regress/sql/object_address.sql b/src/test/regress/sql/object_address.sql index c698a63..8a738e2 100644 --- a/src/test/regress/sql/object_address.sql +++ b/src/test/regress/sql/object_address.sql @@ -40,7 +40,7 @@ CREATE TRANSFORM FOR int LANGUAGE SQL ( FROM SQL WITH FUNCTION varchar_transform(internal), TO SQL WITH FUNCTION int4recv(internal)); CREATE PUBLICATION addr_pub FOR TABLE addr_nsp.gentable; -CREATE SUBSCRIPTION addr_sub CONNECTION '' PUBLICATION bar WITH (connect = false); +CREATE SUBSCRIPTION addr_sub CONNECTION '' PUBLICATION bar WITH (connect = false, slot_name = NONE); CREATE STATISTICS addr_nsp.gentable_stat ON (a,b) FROM addr_nsp.gentable; -- test some error cases @@ -205,7 +205,7 @@ SET client_min_messages TO 'warning'; DROP FOREIGN DATA WRAPPER addr_fdw CASCADE; DROP PUBLICATION addr_pub; -DROP SUBSCRIPTION addr_sub NODROP SLOT; +DROP SUBSCRIPTION addr_sub; DROP SCHEMA addr_nsp CASCADE; diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql index 5454420..c27fee5 100644 --- a/src/test/regress/sql/subscription.sql +++ b/src/test/regress/sql/subscription.sql @@ -80,17 +80,19 @@ ALTER ROLE regress_subscription_user2 SUPERUSER; -- now it works ALTER SUBSCRIPTION testsub OWNER TO regress_subscription_user2; --- fail - cannot do DROP SUBSCRIPTION DROP SLOT inside transaction block +ALTER SUBSCRIPTION testsub WITH (slot_name = NONE); + +-- fail - cannot do DROP SUBSCRIPTION CASCADE inside transaction block BEGIN; -DROP SUBSCRIPTION testsub DROP SLOT; +DROP SUBSCRIPTION testsub CASCADE; COMMIT; BEGIN; -DROP SUBSCRIPTION testsub NODROP SLOT; +DROP SUBSCRIPTION testsub; COMMIT; -DROP SUBSCRIPTION IF EXISTS testsub NODROP SLOT; -DROP SUBSCRIPTION testsub NODROP SLOT; -- fail +DROP SUBSCRIPTION IF EXISTS testsub; +DROP SUBSCRIPTION testsub; -- fail RESET SESSION AUTHORIZATION; DROP ROLE regress_subscription_user; diff --git a/src/test/subscription/t/001_rep_changes.pl b/src/test/subscription/t/001_rep_changes.pl index b4c5539..be55279 100644 --- a/src/test/subscription/t/001_rep_changes.pl +++ b/src/test/subscription/t/001_rep_changes.pl @@ -191,7 +191,7 @@ $node_publisher->poll_query_until('postgres', or die "Timed out while waiting for apply to restart"; # check all the cleanup -$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_renamed DROP SLOT"); +$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_renamed CASCADE"); $result = $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM pg_subscription"); diff --git a/src/test/subscription/t/004_sync.pl b/src/test/subscription/t/004_sync.pl index 8ece7dd..779c668 100644 --- a/src/test/subscription/t/004_sync.pl +++ b/src/test/subscription/t/004_sync.pl @@ -52,7 +52,7 @@ my $result = is($result, qq(10), 'initial data synced for first sub'); # drop subscription so that there is unreplicated data -$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub"); +$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub CASCADE"); $node_publisher->safe_psql('postgres', "INSERT INTO tab_rep SELECT generate_series(11,20)"); @@ -89,8 +89,8 @@ $node_subscriber->poll_query_until('postgres', "SELECT pid IS NOT NULL FROM pg_s or die "Timed out while waiting for subscriber to start"; # and drop both subscriptions -$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub"); -$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub2"); +$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub CASCADE"); +$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub2 CASCADE"); # check subscriptions are removed $result = @@ -154,7 +154,7 @@ $result = $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_rep_next"); is($result, qq(20), 'changes for table added after subscription initialized replicated'); -$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub"); +$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub CASCADE"); $node_subscriber->stop('fast'); $node_publisher->stop('fast'); -- 2.7.4 --------------06EC69927D8D5233C2F60AB9 Content-Type: text/plain Content-Disposition: inline Content-Transfer-Encoding: 8bit MIME-Version: 1.0 -- Sent via pgsql-hackers mailing list ([email protected]) To make changes to your subscription: http://www.postgresql.org/mailpref/pgsql-hackers --------------06EC69927D8D5233C2F60AB9-- ^ permalink raw reply [nested|flat] 27+ messages in thread
* [PATCH 2/2] doc review @ 2021-03-03 22:36 Justin Pryzby <[email protected]> 0 siblings, 0 replies; 27+ messages in thread From: Justin Pryzby @ 2021-03-03 22:36 UTC (permalink / raw) --- doc/src/sgml/libpq.sgml | 67 +++++++++++++------------- src/interfaces/libpq/fe-exec.c | 2 +- src/test/modules/test_libpq/pipeline.c | 6 +-- 3 files changed, 38 insertions(+), 37 deletions(-) diff --git a/doc/src/sgml/libpq.sgml b/doc/src/sgml/libpq.sgml index a34ecbb144..e2865a218b 100644 --- a/doc/src/sgml/libpq.sgml +++ b/doc/src/sgml/libpq.sgml @@ -3130,8 +3130,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_SYNC</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a point in a - pipeline where a synchronization point has been established. + The <structname>PGresult</structname> represents a + synchronization point in pipeline mode, requested by + <xref linkend="libpq-PQsendPipeline"/>. This status occurs only when pipeline mode has been selected. </para> </listitem> @@ -3141,9 +3142,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_ABORTED</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a pipeline that's + The <structname>PGresult</structname> represents a pipeline that has received an error from the server. <function>PQgetResult</function> - must be called repeatedly, and it will return this status code, + must be called repeatedly, and each time it will return this status code until the end of the current pipeline, at which point it will return <literal>PGRES_PIPELINE_SYNC</literal> and normal processing can resume. @@ -4953,7 +4954,7 @@ int PQflush(PGconn *conn); <title>Using Pipeline Mode</title> <para> - To issue pipelines the application must switch a connection into pipeline mode. + To issue pipelines, the application must switch a connection into pipeline mode. Enter pipeline mode with <xref linkend="libpq-PQenterPipelineMode"/> or test whether pipeline mode is active with <xref linkend="libpq-PQpipelineStatus"/>. @@ -4975,7 +4976,7 @@ int PQflush(PGconn *conn); server will block trying to send results to the client from queries it has already processed. This only occurs when the client sends enough queries to fill both its output buffer and the server's receive - buffer before switching to processing input from the server, + buffer before it switches to processing input from the server, but it's hard to predict exactly when that will happen. </para> </footnote> @@ -5025,7 +5026,7 @@ int PQflush(PGconn *conn); <title>Processing Results</title> <para> - To process the result of one pipeline query, the application calls + To process the result of one query in a pipeline, the application calls <function>PQgetResult</function> repeatedly and handles each result until <function>PQgetResult</function> returns null. The result from the next query in the pipeline may then be retrieved using @@ -5057,10 +5058,10 @@ int PQflush(PGconn *conn); <type>PGresult</type> types <literal>PGRES_PIPELINE_SYNC</literal> and <literal>PGRES_PIPELINE_ABORTED</literal>. <literal>PGRES_PIPELINE_SYNC</literal> is reported exactly once for each - <function>PQsendPipeline</function> call at the corresponding point in - the result stream. + <function>PQsendPipeline</function> after retrieving results for all + queries in the pipeline. <literal>PGRES_PIPELINE_ABORTED</literal> is emitted in place of a normal - result stream result for the first error and all subsequent results + stream result for the first error and all subsequent results except <literal>PGRES_PIPELINE_SYNC</literal> and null; see <xref linkend="libpq-pipeline-errors"/>. </para> @@ -5075,7 +5076,8 @@ int PQflush(PGconn *conn); application about the query currently being processed (except that <function>PQgetResult</function> returns null to indicate that we start returning the results of next query). The application must keep track - of the order in which it sent queries and the expected results. + of the order in which it sent queries, to associate them with their + corresponding results. Applications will typically use a state machine or a FIFO queue for this. </para> @@ -5091,10 +5093,10 @@ int PQflush(PGconn *conn); </para> <para> - From the client perspective, after the client gets a - <literal>PGRES_FATAL_ERROR</literal> return from - <function>PQresultStatus</function> the pipeline is flagged as aborted. - <application>libpq</application> will report + From the client perspective, after <function>PQresultStatus</function> + returns <literal>PGRES_FATAL_ERROR</literal>, + the pipeline is flagged as aborted. + <function>PQresultStatus</function>, will report a <literal>PGRES_PIPELINE_ABORTED</literal> result for each remaining queued operation in an aborted pipeline. The result for <function>PQsendPipeline</function> is reported as @@ -5108,8 +5110,8 @@ int PQflush(PGconn *conn); </para> <para> - If the pipeline used an implicit transaction then operations that have - already executed are rolled back and operations that were queued for after + If the pipeline used an implicit transaction, then operations that have + already executed are rolled back and operations that were queued to follow the failed operation are skipped entirely. The same behaviour holds if the pipeline starts and commits a single explicit transaction (i.e. the first statement is <literal>BEGIN</literal> and the last is @@ -5145,11 +5147,11 @@ int PQflush(PGconn *conn); <para> The client application should generally maintain a queue of work - still to be dispatched and a queue of work that has been dispatched + remaining to be dispatched and a queue of work that has been dispatched but not yet had its results processed. When the socket is writable it should dispatch more work. When the socket is readable it should read results and process them, matching them up to the next entry in - its expected results queue. Based on available memory, results from + its expected results queue. Based on available memory, results from the socket should be read frequently: there's no need to wait until the pipeline end to read the results. Pipelines should be scoped to logical units of work, usually (but not necessarily) one transaction per pipeline. @@ -5191,8 +5193,8 @@ int PQflush(PGconn *conn); <listitem> <para> - Returns current pipeline mode status of the <application>libpq</application> - connection. + Returns the current pipeline mode status of the + <application>libpq</application> connection. <synopsis> PGpipelineStatus PQpipelineStatus(const PGconn *conn); </synopsis> @@ -5233,11 +5235,10 @@ PGpipelineStatus PQpipelineStatus(const PGconn *conn); <listitem> <para> The <application>libpq</application> connection is in pipeline - mode and an error has occurred while processing the current - pipeline. - The aborted flag is cleared as soon as the result - of the <function>PQsendPipeline</function> at the end of the aborted - pipeline is processed. Clients don't usually need this function to + mode and an error occurred while processing the current pipeline. + The aborted flag is cleared when <function>PQresultStatus</function> + returns PGRES_PIPELINE_SYNC at the end of the pipeline. + Clients don't usually need this function to verify aborted status, as they can tell that the pipeline is aborted from the <literal>PGRES_PIPELINE_ABORTED</literal> result code. </para> @@ -5328,7 +5329,7 @@ int PQsendPipeline(PGconn *conn); Returns 1 for success. Returns 0 if the connection is not in pipeline mode or sending a <link linkend="protocol-flow-ext-query">sync message</link> - is failed. + failed. </para> </listitem> </varlistentry> @@ -5349,7 +5350,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is most useful when the server is distant, i.e., network latency (<quote>ping time</quote>) is high, and also when many small operations - are being performed in rapid sequence. There is usually less benefit + are being performed in rapid succession. There is usually less benefit in using pipelined commands when each query takes many multiples of the client/server round-trip time to execute. A 100-statement operation run on a server 300ms round-trip-time away would take 30 seconds in network latency alone @@ -5367,7 +5368,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is not useful when information from one operation is required by the client to produce the next operation. In such cases, the client - must introduce a synchronization point and wait for a full client/server + would have to introduce a synchronization point and wait for a full client/server round-trip to get the results it needs. However, it's often possible to adjust the client design to exchange the required information server-side. Read-modify-write cycles are especially good candidates; for example: @@ -5392,10 +5393,10 @@ UPDATE mytable SET x = x + 1 WHERE id = 42; <note> <para> - The pipeline API was introduced in PostgreSQL 14, but clients using - the PostgreSQL 14 version of <application>libpq</application> can use - pipelines on server versions 7.4 and newer. Pipeline mode works on any server - that supports the v3 extended query protocol. + The pipeline API was introduced in <productname>PostgreSQL</productname> 14. + Pipeline mode is a client-side feature which doesn't require server + support, and works on any server that supports the v3 extended query + protocol. </para> </note> </sect2> diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c index 7ae7c14948..d7b036a35c 100644 --- a/src/interfaces/libpq/fe-exec.c +++ b/src/interfaces/libpq/fe-exec.c @@ -3803,7 +3803,7 @@ pqPipelineFlush(PGconn *conn) { if ((conn->pipelineStatus == PQ_PIPELINE_OFF) || (conn->outCount >= OUTBUFFER_THRESHOLD)) - return (pqFlush(conn)); + return pqFlush(conn); return 0; } diff --git a/src/test/modules/test_libpq/pipeline.c b/src/test/modules/test_libpq/pipeline.c index f4a2bdec57..01d5a9a8ff 100644 --- a/src/test/modules/test_libpq/pipeline.c +++ b/src/test/modules/test_libpq/pipeline.c @@ -126,7 +126,7 @@ test_simple_pipeline(PGconn *conn) * process the results of as they come in. * * For a simple case we should be able to do this without interim - * processing of results since our out buffer will give us enough slush to + * processing of results since our output buffer will give us enough slush to * work with and we won't block on sending. So blocking mode is fine. */ if (PQisnonblocking(conn)) @@ -603,7 +603,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) /* * Now we start inserting. We'll be sending enough data that we could fill - * our out buffer, so to avoid deadlocking we need to enter nonblocking + * our output buffer, so to avoid deadlocking we need to enter nonblocking * mode and consume input while we send more output. As results of each * query are processed we should pop them to allow processing of the next * query. There's no need to finish the pipeline before processing @@ -635,7 +635,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) } /* - * Process any results, so we keep the server's out buffer free + * Process any results, so we keep the server's output buffer free * flowing and it can continue to process input */ if (FD_ISSET(sock, &input_mask)) -- 2.17.0 --oC1+HKm2/end4ao3-- ^ permalink raw reply [nested|flat] 27+ messages in thread
* [PATCH 2/2] doc review @ 2021-03-03 22:36 Justin Pryzby <[email protected]> 0 siblings, 0 replies; 27+ messages in thread From: Justin Pryzby @ 2021-03-03 22:36 UTC (permalink / raw) --- doc/src/sgml/libpq.sgml | 67 +++++++++++++------------- src/interfaces/libpq/fe-exec.c | 2 +- src/test/modules/test_libpq/pipeline.c | 6 +-- 3 files changed, 38 insertions(+), 37 deletions(-) diff --git a/doc/src/sgml/libpq.sgml b/doc/src/sgml/libpq.sgml index a34ecbb144..e2865a218b 100644 --- a/doc/src/sgml/libpq.sgml +++ b/doc/src/sgml/libpq.sgml @@ -3130,8 +3130,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_SYNC</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a point in a - pipeline where a synchronization point has been established. + The <structname>PGresult</structname> represents a + synchronization point in pipeline mode, requested by + <xref linkend="libpq-PQsendPipeline"/>. This status occurs only when pipeline mode has been selected. </para> </listitem> @@ -3141,9 +3142,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_ABORTED</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a pipeline that's + The <structname>PGresult</structname> represents a pipeline that has received an error from the server. <function>PQgetResult</function> - must be called repeatedly, and it will return this status code, + must be called repeatedly, and each time it will return this status code until the end of the current pipeline, at which point it will return <literal>PGRES_PIPELINE_SYNC</literal> and normal processing can resume. @@ -4953,7 +4954,7 @@ int PQflush(PGconn *conn); <title>Using Pipeline Mode</title> <para> - To issue pipelines the application must switch a connection into pipeline mode. + To issue pipelines, the application must switch a connection into pipeline mode. Enter pipeline mode with <xref linkend="libpq-PQenterPipelineMode"/> or test whether pipeline mode is active with <xref linkend="libpq-PQpipelineStatus"/>. @@ -4975,7 +4976,7 @@ int PQflush(PGconn *conn); server will block trying to send results to the client from queries it has already processed. This only occurs when the client sends enough queries to fill both its output buffer and the server's receive - buffer before switching to processing input from the server, + buffer before it switches to processing input from the server, but it's hard to predict exactly when that will happen. </para> </footnote> @@ -5025,7 +5026,7 @@ int PQflush(PGconn *conn); <title>Processing Results</title> <para> - To process the result of one pipeline query, the application calls + To process the result of one query in a pipeline, the application calls <function>PQgetResult</function> repeatedly and handles each result until <function>PQgetResult</function> returns null. The result from the next query in the pipeline may then be retrieved using @@ -5057,10 +5058,10 @@ int PQflush(PGconn *conn); <type>PGresult</type> types <literal>PGRES_PIPELINE_SYNC</literal> and <literal>PGRES_PIPELINE_ABORTED</literal>. <literal>PGRES_PIPELINE_SYNC</literal> is reported exactly once for each - <function>PQsendPipeline</function> call at the corresponding point in - the result stream. + <function>PQsendPipeline</function> after retrieving results for all + queries in the pipeline. <literal>PGRES_PIPELINE_ABORTED</literal> is emitted in place of a normal - result stream result for the first error and all subsequent results + stream result for the first error and all subsequent results except <literal>PGRES_PIPELINE_SYNC</literal> and null; see <xref linkend="libpq-pipeline-errors"/>. </para> @@ -5075,7 +5076,8 @@ int PQflush(PGconn *conn); application about the query currently being processed (except that <function>PQgetResult</function> returns null to indicate that we start returning the results of next query). The application must keep track - of the order in which it sent queries and the expected results. + of the order in which it sent queries, to associate them with their + corresponding results. Applications will typically use a state machine or a FIFO queue for this. </para> @@ -5091,10 +5093,10 @@ int PQflush(PGconn *conn); </para> <para> - From the client perspective, after the client gets a - <literal>PGRES_FATAL_ERROR</literal> return from - <function>PQresultStatus</function> the pipeline is flagged as aborted. - <application>libpq</application> will report + From the client perspective, after <function>PQresultStatus</function> + returns <literal>PGRES_FATAL_ERROR</literal>, + the pipeline is flagged as aborted. + <function>PQresultStatus</function>, will report a <literal>PGRES_PIPELINE_ABORTED</literal> result for each remaining queued operation in an aborted pipeline. The result for <function>PQsendPipeline</function> is reported as @@ -5108,8 +5110,8 @@ int PQflush(PGconn *conn); </para> <para> - If the pipeline used an implicit transaction then operations that have - already executed are rolled back and operations that were queued for after + If the pipeline used an implicit transaction, then operations that have + already executed are rolled back and operations that were queued to follow the failed operation are skipped entirely. The same behaviour holds if the pipeline starts and commits a single explicit transaction (i.e. the first statement is <literal>BEGIN</literal> and the last is @@ -5145,11 +5147,11 @@ int PQflush(PGconn *conn); <para> The client application should generally maintain a queue of work - still to be dispatched and a queue of work that has been dispatched + remaining to be dispatched and a queue of work that has been dispatched but not yet had its results processed. When the socket is writable it should dispatch more work. When the socket is readable it should read results and process them, matching them up to the next entry in - its expected results queue. Based on available memory, results from + its expected results queue. Based on available memory, results from the socket should be read frequently: there's no need to wait until the pipeline end to read the results. Pipelines should be scoped to logical units of work, usually (but not necessarily) one transaction per pipeline. @@ -5191,8 +5193,8 @@ int PQflush(PGconn *conn); <listitem> <para> - Returns current pipeline mode status of the <application>libpq</application> - connection. + Returns the current pipeline mode status of the + <application>libpq</application> connection. <synopsis> PGpipelineStatus PQpipelineStatus(const PGconn *conn); </synopsis> @@ -5233,11 +5235,10 @@ PGpipelineStatus PQpipelineStatus(const PGconn *conn); <listitem> <para> The <application>libpq</application> connection is in pipeline - mode and an error has occurred while processing the current - pipeline. - The aborted flag is cleared as soon as the result - of the <function>PQsendPipeline</function> at the end of the aborted - pipeline is processed. Clients don't usually need this function to + mode and an error occurred while processing the current pipeline. + The aborted flag is cleared when <function>PQresultStatus</function> + returns PGRES_PIPELINE_SYNC at the end of the pipeline. + Clients don't usually need this function to verify aborted status, as they can tell that the pipeline is aborted from the <literal>PGRES_PIPELINE_ABORTED</literal> result code. </para> @@ -5328,7 +5329,7 @@ int PQsendPipeline(PGconn *conn); Returns 1 for success. Returns 0 if the connection is not in pipeline mode or sending a <link linkend="protocol-flow-ext-query">sync message</link> - is failed. + failed. </para> </listitem> </varlistentry> @@ -5349,7 +5350,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is most useful when the server is distant, i.e., network latency (<quote>ping time</quote>) is high, and also when many small operations - are being performed in rapid sequence. There is usually less benefit + are being performed in rapid succession. There is usually less benefit in using pipelined commands when each query takes many multiples of the client/server round-trip time to execute. A 100-statement operation run on a server 300ms round-trip-time away would take 30 seconds in network latency alone @@ -5367,7 +5368,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is not useful when information from one operation is required by the client to produce the next operation. In such cases, the client - must introduce a synchronization point and wait for a full client/server + would have to introduce a synchronization point and wait for a full client/server round-trip to get the results it needs. However, it's often possible to adjust the client design to exchange the required information server-side. Read-modify-write cycles are especially good candidates; for example: @@ -5392,10 +5393,10 @@ UPDATE mytable SET x = x + 1 WHERE id = 42; <note> <para> - The pipeline API was introduced in PostgreSQL 14, but clients using - the PostgreSQL 14 version of <application>libpq</application> can use - pipelines on server versions 7.4 and newer. Pipeline mode works on any server - that supports the v3 extended query protocol. + The pipeline API was introduced in <productname>PostgreSQL</productname> 14. + Pipeline mode is a client-side feature which doesn't require server + support, and works on any server that supports the v3 extended query + protocol. </para> </note> </sect2> diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c index 7ae7c14948..d7b036a35c 100644 --- a/src/interfaces/libpq/fe-exec.c +++ b/src/interfaces/libpq/fe-exec.c @@ -3803,7 +3803,7 @@ pqPipelineFlush(PGconn *conn) { if ((conn->pipelineStatus == PQ_PIPELINE_OFF) || (conn->outCount >= OUTBUFFER_THRESHOLD)) - return (pqFlush(conn)); + return pqFlush(conn); return 0; } diff --git a/src/test/modules/test_libpq/pipeline.c b/src/test/modules/test_libpq/pipeline.c index f4a2bdec57..01d5a9a8ff 100644 --- a/src/test/modules/test_libpq/pipeline.c +++ b/src/test/modules/test_libpq/pipeline.c @@ -126,7 +126,7 @@ test_simple_pipeline(PGconn *conn) * process the results of as they come in. * * For a simple case we should be able to do this without interim - * processing of results since our out buffer will give us enough slush to + * processing of results since our output buffer will give us enough slush to * work with and we won't block on sending. So blocking mode is fine. */ if (PQisnonblocking(conn)) @@ -603,7 +603,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) /* * Now we start inserting. We'll be sending enough data that we could fill - * our out buffer, so to avoid deadlocking we need to enter nonblocking + * our output buffer, so to avoid deadlocking we need to enter nonblocking * mode and consume input while we send more output. As results of each * query are processed we should pop them to allow processing of the next * query. There's no need to finish the pipeline before processing @@ -635,7 +635,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) } /* - * Process any results, so we keep the server's out buffer free + * Process any results, so we keep the server's output buffer free * flowing and it can continue to process input */ if (FD_ISSET(sock, &input_mask)) -- 2.17.0 --oC1+HKm2/end4ao3-- ^ permalink raw reply [nested|flat] 27+ messages in thread
* [PATCH 2/2] doc review @ 2021-03-03 22:36 Justin Pryzby <[email protected]> 0 siblings, 0 replies; 27+ messages in thread From: Justin Pryzby @ 2021-03-03 22:36 UTC (permalink / raw) --- doc/src/sgml/libpq.sgml | 67 +++++++++++++------------- src/interfaces/libpq/fe-exec.c | 2 +- src/test/modules/test_libpq/pipeline.c | 6 +-- 3 files changed, 38 insertions(+), 37 deletions(-) diff --git a/doc/src/sgml/libpq.sgml b/doc/src/sgml/libpq.sgml index a34ecbb144..e2865a218b 100644 --- a/doc/src/sgml/libpq.sgml +++ b/doc/src/sgml/libpq.sgml @@ -3130,8 +3130,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_SYNC</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a point in a - pipeline where a synchronization point has been established. + The <structname>PGresult</structname> represents a + synchronization point in pipeline mode, requested by + <xref linkend="libpq-PQsendPipeline"/>. This status occurs only when pipeline mode has been selected. </para> </listitem> @@ -3141,9 +3142,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_ABORTED</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a pipeline that's + The <structname>PGresult</structname> represents a pipeline that has received an error from the server. <function>PQgetResult</function> - must be called repeatedly, and it will return this status code, + must be called repeatedly, and each time it will return this status code until the end of the current pipeline, at which point it will return <literal>PGRES_PIPELINE_SYNC</literal> and normal processing can resume. @@ -4953,7 +4954,7 @@ int PQflush(PGconn *conn); <title>Using Pipeline Mode</title> <para> - To issue pipelines the application must switch a connection into pipeline mode. + To issue pipelines, the application must switch a connection into pipeline mode. Enter pipeline mode with <xref linkend="libpq-PQenterPipelineMode"/> or test whether pipeline mode is active with <xref linkend="libpq-PQpipelineStatus"/>. @@ -4975,7 +4976,7 @@ int PQflush(PGconn *conn); server will block trying to send results to the client from queries it has already processed. This only occurs when the client sends enough queries to fill both its output buffer and the server's receive - buffer before switching to processing input from the server, + buffer before it switches to processing input from the server, but it's hard to predict exactly when that will happen. </para> </footnote> @@ -5025,7 +5026,7 @@ int PQflush(PGconn *conn); <title>Processing Results</title> <para> - To process the result of one pipeline query, the application calls + To process the result of one query in a pipeline, the application calls <function>PQgetResult</function> repeatedly and handles each result until <function>PQgetResult</function> returns null. The result from the next query in the pipeline may then be retrieved using @@ -5057,10 +5058,10 @@ int PQflush(PGconn *conn); <type>PGresult</type> types <literal>PGRES_PIPELINE_SYNC</literal> and <literal>PGRES_PIPELINE_ABORTED</literal>. <literal>PGRES_PIPELINE_SYNC</literal> is reported exactly once for each - <function>PQsendPipeline</function> call at the corresponding point in - the result stream. + <function>PQsendPipeline</function> after retrieving results for all + queries in the pipeline. <literal>PGRES_PIPELINE_ABORTED</literal> is emitted in place of a normal - result stream result for the first error and all subsequent results + stream result for the first error and all subsequent results except <literal>PGRES_PIPELINE_SYNC</literal> and null; see <xref linkend="libpq-pipeline-errors"/>. </para> @@ -5075,7 +5076,8 @@ int PQflush(PGconn *conn); application about the query currently being processed (except that <function>PQgetResult</function> returns null to indicate that we start returning the results of next query). The application must keep track - of the order in which it sent queries and the expected results. + of the order in which it sent queries, to associate them with their + corresponding results. Applications will typically use a state machine or a FIFO queue for this. </para> @@ -5091,10 +5093,10 @@ int PQflush(PGconn *conn); </para> <para> - From the client perspective, after the client gets a - <literal>PGRES_FATAL_ERROR</literal> return from - <function>PQresultStatus</function> the pipeline is flagged as aborted. - <application>libpq</application> will report + From the client perspective, after <function>PQresultStatus</function> + returns <literal>PGRES_FATAL_ERROR</literal>, + the pipeline is flagged as aborted. + <function>PQresultStatus</function>, will report a <literal>PGRES_PIPELINE_ABORTED</literal> result for each remaining queued operation in an aborted pipeline. The result for <function>PQsendPipeline</function> is reported as @@ -5108,8 +5110,8 @@ int PQflush(PGconn *conn); </para> <para> - If the pipeline used an implicit transaction then operations that have - already executed are rolled back and operations that were queued for after + If the pipeline used an implicit transaction, then operations that have + already executed are rolled back and operations that were queued to follow the failed operation are skipped entirely. The same behaviour holds if the pipeline starts and commits a single explicit transaction (i.e. the first statement is <literal>BEGIN</literal> and the last is @@ -5145,11 +5147,11 @@ int PQflush(PGconn *conn); <para> The client application should generally maintain a queue of work - still to be dispatched and a queue of work that has been dispatched + remaining to be dispatched and a queue of work that has been dispatched but not yet had its results processed. When the socket is writable it should dispatch more work. When the socket is readable it should read results and process them, matching them up to the next entry in - its expected results queue. Based on available memory, results from + its expected results queue. Based on available memory, results from the socket should be read frequently: there's no need to wait until the pipeline end to read the results. Pipelines should be scoped to logical units of work, usually (but not necessarily) one transaction per pipeline. @@ -5191,8 +5193,8 @@ int PQflush(PGconn *conn); <listitem> <para> - Returns current pipeline mode status of the <application>libpq</application> - connection. + Returns the current pipeline mode status of the + <application>libpq</application> connection. <synopsis> PGpipelineStatus PQpipelineStatus(const PGconn *conn); </synopsis> @@ -5233,11 +5235,10 @@ PGpipelineStatus PQpipelineStatus(const PGconn *conn); <listitem> <para> The <application>libpq</application> connection is in pipeline - mode and an error has occurred while processing the current - pipeline. - The aborted flag is cleared as soon as the result - of the <function>PQsendPipeline</function> at the end of the aborted - pipeline is processed. Clients don't usually need this function to + mode and an error occurred while processing the current pipeline. + The aborted flag is cleared when <function>PQresultStatus</function> + returns PGRES_PIPELINE_SYNC at the end of the pipeline. + Clients don't usually need this function to verify aborted status, as they can tell that the pipeline is aborted from the <literal>PGRES_PIPELINE_ABORTED</literal> result code. </para> @@ -5328,7 +5329,7 @@ int PQsendPipeline(PGconn *conn); Returns 1 for success. Returns 0 if the connection is not in pipeline mode or sending a <link linkend="protocol-flow-ext-query">sync message</link> - is failed. + failed. </para> </listitem> </varlistentry> @@ -5349,7 +5350,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is most useful when the server is distant, i.e., network latency (<quote>ping time</quote>) is high, and also when many small operations - are being performed in rapid sequence. There is usually less benefit + are being performed in rapid succession. There is usually less benefit in using pipelined commands when each query takes many multiples of the client/server round-trip time to execute. A 100-statement operation run on a server 300ms round-trip-time away would take 30 seconds in network latency alone @@ -5367,7 +5368,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is not useful when information from one operation is required by the client to produce the next operation. In such cases, the client - must introduce a synchronization point and wait for a full client/server + would have to introduce a synchronization point and wait for a full client/server round-trip to get the results it needs. However, it's often possible to adjust the client design to exchange the required information server-side. Read-modify-write cycles are especially good candidates; for example: @@ -5392,10 +5393,10 @@ UPDATE mytable SET x = x + 1 WHERE id = 42; <note> <para> - The pipeline API was introduced in PostgreSQL 14, but clients using - the PostgreSQL 14 version of <application>libpq</application> can use - pipelines on server versions 7.4 and newer. Pipeline mode works on any server - that supports the v3 extended query protocol. + The pipeline API was introduced in <productname>PostgreSQL</productname> 14. + Pipeline mode is a client-side feature which doesn't require server + support, and works on any server that supports the v3 extended query + protocol. </para> </note> </sect2> diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c index 7ae7c14948..d7b036a35c 100644 --- a/src/interfaces/libpq/fe-exec.c +++ b/src/interfaces/libpq/fe-exec.c @@ -3803,7 +3803,7 @@ pqPipelineFlush(PGconn *conn) { if ((conn->pipelineStatus == PQ_PIPELINE_OFF) || (conn->outCount >= OUTBUFFER_THRESHOLD)) - return (pqFlush(conn)); + return pqFlush(conn); return 0; } diff --git a/src/test/modules/test_libpq/pipeline.c b/src/test/modules/test_libpq/pipeline.c index f4a2bdec57..01d5a9a8ff 100644 --- a/src/test/modules/test_libpq/pipeline.c +++ b/src/test/modules/test_libpq/pipeline.c @@ -126,7 +126,7 @@ test_simple_pipeline(PGconn *conn) * process the results of as they come in. * * For a simple case we should be able to do this without interim - * processing of results since our out buffer will give us enough slush to + * processing of results since our output buffer will give us enough slush to * work with and we won't block on sending. So blocking mode is fine. */ if (PQisnonblocking(conn)) @@ -603,7 +603,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) /* * Now we start inserting. We'll be sending enough data that we could fill - * our out buffer, so to avoid deadlocking we need to enter nonblocking + * our output buffer, so to avoid deadlocking we need to enter nonblocking * mode and consume input while we send more output. As results of each * query are processed we should pop them to allow processing of the next * query. There's no need to finish the pipeline before processing @@ -635,7 +635,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) } /* - * Process any results, so we keep the server's out buffer free + * Process any results, so we keep the server's output buffer free * flowing and it can continue to process input */ if (FD_ISSET(sock, &input_mask)) -- 2.17.0 --oC1+HKm2/end4ao3-- ^ permalink raw reply [nested|flat] 27+ messages in thread
* [PATCH 2/2] doc review @ 2021-03-03 22:36 Justin Pryzby <[email protected]> 0 siblings, 0 replies; 27+ messages in thread From: Justin Pryzby @ 2021-03-03 22:36 UTC (permalink / raw) --- doc/src/sgml/libpq.sgml | 67 +++++++++++++------------- src/interfaces/libpq/fe-exec.c | 2 +- src/test/modules/test_libpq/pipeline.c | 6 +-- 3 files changed, 38 insertions(+), 37 deletions(-) diff --git a/doc/src/sgml/libpq.sgml b/doc/src/sgml/libpq.sgml index a34ecbb144..e2865a218b 100644 --- a/doc/src/sgml/libpq.sgml +++ b/doc/src/sgml/libpq.sgml @@ -3130,8 +3130,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_SYNC</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a point in a - pipeline where a synchronization point has been established. + The <structname>PGresult</structname> represents a + synchronization point in pipeline mode, requested by + <xref linkend="libpq-PQsendPipeline"/>. This status occurs only when pipeline mode has been selected. </para> </listitem> @@ -3141,9 +3142,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_ABORTED</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a pipeline that's + The <structname>PGresult</structname> represents a pipeline that has received an error from the server. <function>PQgetResult</function> - must be called repeatedly, and it will return this status code, + must be called repeatedly, and each time it will return this status code until the end of the current pipeline, at which point it will return <literal>PGRES_PIPELINE_SYNC</literal> and normal processing can resume. @@ -4953,7 +4954,7 @@ int PQflush(PGconn *conn); <title>Using Pipeline Mode</title> <para> - To issue pipelines the application must switch a connection into pipeline mode. + To issue pipelines, the application must switch a connection into pipeline mode. Enter pipeline mode with <xref linkend="libpq-PQenterPipelineMode"/> or test whether pipeline mode is active with <xref linkend="libpq-PQpipelineStatus"/>. @@ -4975,7 +4976,7 @@ int PQflush(PGconn *conn); server will block trying to send results to the client from queries it has already processed. This only occurs when the client sends enough queries to fill both its output buffer and the server's receive - buffer before switching to processing input from the server, + buffer before it switches to processing input from the server, but it's hard to predict exactly when that will happen. </para> </footnote> @@ -5025,7 +5026,7 @@ int PQflush(PGconn *conn); <title>Processing Results</title> <para> - To process the result of one pipeline query, the application calls + To process the result of one query in a pipeline, the application calls <function>PQgetResult</function> repeatedly and handles each result until <function>PQgetResult</function> returns null. The result from the next query in the pipeline may then be retrieved using @@ -5057,10 +5058,10 @@ int PQflush(PGconn *conn); <type>PGresult</type> types <literal>PGRES_PIPELINE_SYNC</literal> and <literal>PGRES_PIPELINE_ABORTED</literal>. <literal>PGRES_PIPELINE_SYNC</literal> is reported exactly once for each - <function>PQsendPipeline</function> call at the corresponding point in - the result stream. + <function>PQsendPipeline</function> after retrieving results for all + queries in the pipeline. <literal>PGRES_PIPELINE_ABORTED</literal> is emitted in place of a normal - result stream result for the first error and all subsequent results + stream result for the first error and all subsequent results except <literal>PGRES_PIPELINE_SYNC</literal> and null; see <xref linkend="libpq-pipeline-errors"/>. </para> @@ -5075,7 +5076,8 @@ int PQflush(PGconn *conn); application about the query currently being processed (except that <function>PQgetResult</function> returns null to indicate that we start returning the results of next query). The application must keep track - of the order in which it sent queries and the expected results. + of the order in which it sent queries, to associate them with their + corresponding results. Applications will typically use a state machine or a FIFO queue for this. </para> @@ -5091,10 +5093,10 @@ int PQflush(PGconn *conn); </para> <para> - From the client perspective, after the client gets a - <literal>PGRES_FATAL_ERROR</literal> return from - <function>PQresultStatus</function> the pipeline is flagged as aborted. - <application>libpq</application> will report + From the client perspective, after <function>PQresultStatus</function> + returns <literal>PGRES_FATAL_ERROR</literal>, + the pipeline is flagged as aborted. + <function>PQresultStatus</function>, will report a <literal>PGRES_PIPELINE_ABORTED</literal> result for each remaining queued operation in an aborted pipeline. The result for <function>PQsendPipeline</function> is reported as @@ -5108,8 +5110,8 @@ int PQflush(PGconn *conn); </para> <para> - If the pipeline used an implicit transaction then operations that have - already executed are rolled back and operations that were queued for after + If the pipeline used an implicit transaction, then operations that have + already executed are rolled back and operations that were queued to follow the failed operation are skipped entirely. The same behaviour holds if the pipeline starts and commits a single explicit transaction (i.e. the first statement is <literal>BEGIN</literal> and the last is @@ -5145,11 +5147,11 @@ int PQflush(PGconn *conn); <para> The client application should generally maintain a queue of work - still to be dispatched and a queue of work that has been dispatched + remaining to be dispatched and a queue of work that has been dispatched but not yet had its results processed. When the socket is writable it should dispatch more work. When the socket is readable it should read results and process them, matching them up to the next entry in - its expected results queue. Based on available memory, results from + its expected results queue. Based on available memory, results from the socket should be read frequently: there's no need to wait until the pipeline end to read the results. Pipelines should be scoped to logical units of work, usually (but not necessarily) one transaction per pipeline. @@ -5191,8 +5193,8 @@ int PQflush(PGconn *conn); <listitem> <para> - Returns current pipeline mode status of the <application>libpq</application> - connection. + Returns the current pipeline mode status of the + <application>libpq</application> connection. <synopsis> PGpipelineStatus PQpipelineStatus(const PGconn *conn); </synopsis> @@ -5233,11 +5235,10 @@ PGpipelineStatus PQpipelineStatus(const PGconn *conn); <listitem> <para> The <application>libpq</application> connection is in pipeline - mode and an error has occurred while processing the current - pipeline. - The aborted flag is cleared as soon as the result - of the <function>PQsendPipeline</function> at the end of the aborted - pipeline is processed. Clients don't usually need this function to + mode and an error occurred while processing the current pipeline. + The aborted flag is cleared when <function>PQresultStatus</function> + returns PGRES_PIPELINE_SYNC at the end of the pipeline. + Clients don't usually need this function to verify aborted status, as they can tell that the pipeline is aborted from the <literal>PGRES_PIPELINE_ABORTED</literal> result code. </para> @@ -5328,7 +5329,7 @@ int PQsendPipeline(PGconn *conn); Returns 1 for success. Returns 0 if the connection is not in pipeline mode or sending a <link linkend="protocol-flow-ext-query">sync message</link> - is failed. + failed. </para> </listitem> </varlistentry> @@ -5349,7 +5350,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is most useful when the server is distant, i.e., network latency (<quote>ping time</quote>) is high, and also when many small operations - are being performed in rapid sequence. There is usually less benefit + are being performed in rapid succession. There is usually less benefit in using pipelined commands when each query takes many multiples of the client/server round-trip time to execute. A 100-statement operation run on a server 300ms round-trip-time away would take 30 seconds in network latency alone @@ -5367,7 +5368,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is not useful when information from one operation is required by the client to produce the next operation. In such cases, the client - must introduce a synchronization point and wait for a full client/server + would have to introduce a synchronization point and wait for a full client/server round-trip to get the results it needs. However, it's often possible to adjust the client design to exchange the required information server-side. Read-modify-write cycles are especially good candidates; for example: @@ -5392,10 +5393,10 @@ UPDATE mytable SET x = x + 1 WHERE id = 42; <note> <para> - The pipeline API was introduced in PostgreSQL 14, but clients using - the PostgreSQL 14 version of <application>libpq</application> can use - pipelines on server versions 7.4 and newer. Pipeline mode works on any server - that supports the v3 extended query protocol. + The pipeline API was introduced in <productname>PostgreSQL</productname> 14. + Pipeline mode is a client-side feature which doesn't require server + support, and works on any server that supports the v3 extended query + protocol. </para> </note> </sect2> diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c index 7ae7c14948..d7b036a35c 100644 --- a/src/interfaces/libpq/fe-exec.c +++ b/src/interfaces/libpq/fe-exec.c @@ -3803,7 +3803,7 @@ pqPipelineFlush(PGconn *conn) { if ((conn->pipelineStatus == PQ_PIPELINE_OFF) || (conn->outCount >= OUTBUFFER_THRESHOLD)) - return (pqFlush(conn)); + return pqFlush(conn); return 0; } diff --git a/src/test/modules/test_libpq/pipeline.c b/src/test/modules/test_libpq/pipeline.c index f4a2bdec57..01d5a9a8ff 100644 --- a/src/test/modules/test_libpq/pipeline.c +++ b/src/test/modules/test_libpq/pipeline.c @@ -126,7 +126,7 @@ test_simple_pipeline(PGconn *conn) * process the results of as they come in. * * For a simple case we should be able to do this without interim - * processing of results since our out buffer will give us enough slush to + * processing of results since our output buffer will give us enough slush to * work with and we won't block on sending. So blocking mode is fine. */ if (PQisnonblocking(conn)) @@ -603,7 +603,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) /* * Now we start inserting. We'll be sending enough data that we could fill - * our out buffer, so to avoid deadlocking we need to enter nonblocking + * our output buffer, so to avoid deadlocking we need to enter nonblocking * mode and consume input while we send more output. As results of each * query are processed we should pop them to allow processing of the next * query. There's no need to finish the pipeline before processing @@ -635,7 +635,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) } /* - * Process any results, so we keep the server's out buffer free + * Process any results, so we keep the server's output buffer free * flowing and it can continue to process input */ if (FD_ISSET(sock, &input_mask)) -- 2.17.0 --oC1+HKm2/end4ao3-- ^ permalink raw reply [nested|flat] 27+ messages in thread
* [PATCH 2/2] doc review @ 2021-03-03 22:36 Justin Pryzby <[email protected]> 0 siblings, 0 replies; 27+ messages in thread From: Justin Pryzby @ 2021-03-03 22:36 UTC (permalink / raw) --- doc/src/sgml/libpq.sgml | 67 +++++++++++++------------- src/interfaces/libpq/fe-exec.c | 2 +- src/test/modules/test_libpq/pipeline.c | 6 +-- 3 files changed, 38 insertions(+), 37 deletions(-) diff --git a/doc/src/sgml/libpq.sgml b/doc/src/sgml/libpq.sgml index a34ecbb144..e2865a218b 100644 --- a/doc/src/sgml/libpq.sgml +++ b/doc/src/sgml/libpq.sgml @@ -3130,8 +3130,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_SYNC</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a point in a - pipeline where a synchronization point has been established. + The <structname>PGresult</structname> represents a + synchronization point in pipeline mode, requested by + <xref linkend="libpq-PQsendPipeline"/>. This status occurs only when pipeline mode has been selected. </para> </listitem> @@ -3141,9 +3142,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_ABORTED</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a pipeline that's + The <structname>PGresult</structname> represents a pipeline that has received an error from the server. <function>PQgetResult</function> - must be called repeatedly, and it will return this status code, + must be called repeatedly, and each time it will return this status code until the end of the current pipeline, at which point it will return <literal>PGRES_PIPELINE_SYNC</literal> and normal processing can resume. @@ -4953,7 +4954,7 @@ int PQflush(PGconn *conn); <title>Using Pipeline Mode</title> <para> - To issue pipelines the application must switch a connection into pipeline mode. + To issue pipelines, the application must switch a connection into pipeline mode. Enter pipeline mode with <xref linkend="libpq-PQenterPipelineMode"/> or test whether pipeline mode is active with <xref linkend="libpq-PQpipelineStatus"/>. @@ -4975,7 +4976,7 @@ int PQflush(PGconn *conn); server will block trying to send results to the client from queries it has already processed. This only occurs when the client sends enough queries to fill both its output buffer and the server's receive - buffer before switching to processing input from the server, + buffer before it switches to processing input from the server, but it's hard to predict exactly when that will happen. </para> </footnote> @@ -5025,7 +5026,7 @@ int PQflush(PGconn *conn); <title>Processing Results</title> <para> - To process the result of one pipeline query, the application calls + To process the result of one query in a pipeline, the application calls <function>PQgetResult</function> repeatedly and handles each result until <function>PQgetResult</function> returns null. The result from the next query in the pipeline may then be retrieved using @@ -5057,10 +5058,10 @@ int PQflush(PGconn *conn); <type>PGresult</type> types <literal>PGRES_PIPELINE_SYNC</literal> and <literal>PGRES_PIPELINE_ABORTED</literal>. <literal>PGRES_PIPELINE_SYNC</literal> is reported exactly once for each - <function>PQsendPipeline</function> call at the corresponding point in - the result stream. + <function>PQsendPipeline</function> after retrieving results for all + queries in the pipeline. <literal>PGRES_PIPELINE_ABORTED</literal> is emitted in place of a normal - result stream result for the first error and all subsequent results + stream result for the first error and all subsequent results except <literal>PGRES_PIPELINE_SYNC</literal> and null; see <xref linkend="libpq-pipeline-errors"/>. </para> @@ -5075,7 +5076,8 @@ int PQflush(PGconn *conn); application about the query currently being processed (except that <function>PQgetResult</function> returns null to indicate that we start returning the results of next query). The application must keep track - of the order in which it sent queries and the expected results. + of the order in which it sent queries, to associate them with their + corresponding results. Applications will typically use a state machine or a FIFO queue for this. </para> @@ -5091,10 +5093,10 @@ int PQflush(PGconn *conn); </para> <para> - From the client perspective, after the client gets a - <literal>PGRES_FATAL_ERROR</literal> return from - <function>PQresultStatus</function> the pipeline is flagged as aborted. - <application>libpq</application> will report + From the client perspective, after <function>PQresultStatus</function> + returns <literal>PGRES_FATAL_ERROR</literal>, + the pipeline is flagged as aborted. + <function>PQresultStatus</function>, will report a <literal>PGRES_PIPELINE_ABORTED</literal> result for each remaining queued operation in an aborted pipeline. The result for <function>PQsendPipeline</function> is reported as @@ -5108,8 +5110,8 @@ int PQflush(PGconn *conn); </para> <para> - If the pipeline used an implicit transaction then operations that have - already executed are rolled back and operations that were queued for after + If the pipeline used an implicit transaction, then operations that have + already executed are rolled back and operations that were queued to follow the failed operation are skipped entirely. The same behaviour holds if the pipeline starts and commits a single explicit transaction (i.e. the first statement is <literal>BEGIN</literal> and the last is @@ -5145,11 +5147,11 @@ int PQflush(PGconn *conn); <para> The client application should generally maintain a queue of work - still to be dispatched and a queue of work that has been dispatched + remaining to be dispatched and a queue of work that has been dispatched but not yet had its results processed. When the socket is writable it should dispatch more work. When the socket is readable it should read results and process them, matching them up to the next entry in - its expected results queue. Based on available memory, results from + its expected results queue. Based on available memory, results from the socket should be read frequently: there's no need to wait until the pipeline end to read the results. Pipelines should be scoped to logical units of work, usually (but not necessarily) one transaction per pipeline. @@ -5191,8 +5193,8 @@ int PQflush(PGconn *conn); <listitem> <para> - Returns current pipeline mode status of the <application>libpq</application> - connection. + Returns the current pipeline mode status of the + <application>libpq</application> connection. <synopsis> PGpipelineStatus PQpipelineStatus(const PGconn *conn); </synopsis> @@ -5233,11 +5235,10 @@ PGpipelineStatus PQpipelineStatus(const PGconn *conn); <listitem> <para> The <application>libpq</application> connection is in pipeline - mode and an error has occurred while processing the current - pipeline. - The aborted flag is cleared as soon as the result - of the <function>PQsendPipeline</function> at the end of the aborted - pipeline is processed. Clients don't usually need this function to + mode and an error occurred while processing the current pipeline. + The aborted flag is cleared when <function>PQresultStatus</function> + returns PGRES_PIPELINE_SYNC at the end of the pipeline. + Clients don't usually need this function to verify aborted status, as they can tell that the pipeline is aborted from the <literal>PGRES_PIPELINE_ABORTED</literal> result code. </para> @@ -5328,7 +5329,7 @@ int PQsendPipeline(PGconn *conn); Returns 1 for success. Returns 0 if the connection is not in pipeline mode or sending a <link linkend="protocol-flow-ext-query">sync message</link> - is failed. + failed. </para> </listitem> </varlistentry> @@ -5349,7 +5350,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is most useful when the server is distant, i.e., network latency (<quote>ping time</quote>) is high, and also when many small operations - are being performed in rapid sequence. There is usually less benefit + are being performed in rapid succession. There is usually less benefit in using pipelined commands when each query takes many multiples of the client/server round-trip time to execute. A 100-statement operation run on a server 300ms round-trip-time away would take 30 seconds in network latency alone @@ -5367,7 +5368,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is not useful when information from one operation is required by the client to produce the next operation. In such cases, the client - must introduce a synchronization point and wait for a full client/server + would have to introduce a synchronization point and wait for a full client/server round-trip to get the results it needs. However, it's often possible to adjust the client design to exchange the required information server-side. Read-modify-write cycles are especially good candidates; for example: @@ -5392,10 +5393,10 @@ UPDATE mytable SET x = x + 1 WHERE id = 42; <note> <para> - The pipeline API was introduced in PostgreSQL 14, but clients using - the PostgreSQL 14 version of <application>libpq</application> can use - pipelines on server versions 7.4 and newer. Pipeline mode works on any server - that supports the v3 extended query protocol. + The pipeline API was introduced in <productname>PostgreSQL</productname> 14. + Pipeline mode is a client-side feature which doesn't require server + support, and works on any server that supports the v3 extended query + protocol. </para> </note> </sect2> diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c index 7ae7c14948..d7b036a35c 100644 --- a/src/interfaces/libpq/fe-exec.c +++ b/src/interfaces/libpq/fe-exec.c @@ -3803,7 +3803,7 @@ pqPipelineFlush(PGconn *conn) { if ((conn->pipelineStatus == PQ_PIPELINE_OFF) || (conn->outCount >= OUTBUFFER_THRESHOLD)) - return (pqFlush(conn)); + return pqFlush(conn); return 0; } diff --git a/src/test/modules/test_libpq/pipeline.c b/src/test/modules/test_libpq/pipeline.c index f4a2bdec57..01d5a9a8ff 100644 --- a/src/test/modules/test_libpq/pipeline.c +++ b/src/test/modules/test_libpq/pipeline.c @@ -126,7 +126,7 @@ test_simple_pipeline(PGconn *conn) * process the results of as they come in. * * For a simple case we should be able to do this without interim - * processing of results since our out buffer will give us enough slush to + * processing of results since our output buffer will give us enough slush to * work with and we won't block on sending. So blocking mode is fine. */ if (PQisnonblocking(conn)) @@ -603,7 +603,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) /* * Now we start inserting. We'll be sending enough data that we could fill - * our out buffer, so to avoid deadlocking we need to enter nonblocking + * our output buffer, so to avoid deadlocking we need to enter nonblocking * mode and consume input while we send more output. As results of each * query are processed we should pop them to allow processing of the next * query. There's no need to finish the pipeline before processing @@ -635,7 +635,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) } /* - * Process any results, so we keep the server's out buffer free + * Process any results, so we keep the server's output buffer free * flowing and it can continue to process input */ if (FD_ISSET(sock, &input_mask)) -- 2.17.0 --oC1+HKm2/end4ao3-- ^ permalink raw reply [nested|flat] 27+ messages in thread
* [PATCH 2/2] doc review @ 2021-03-03 22:36 Justin Pryzby <[email protected]> 0 siblings, 0 replies; 27+ messages in thread From: Justin Pryzby @ 2021-03-03 22:36 UTC (permalink / raw) --- doc/src/sgml/libpq.sgml | 67 +++++++++++++------------- src/interfaces/libpq/fe-exec.c | 2 +- src/test/modules/test_libpq/pipeline.c | 6 +-- 3 files changed, 38 insertions(+), 37 deletions(-) diff --git a/doc/src/sgml/libpq.sgml b/doc/src/sgml/libpq.sgml index a34ecbb144..e2865a218b 100644 --- a/doc/src/sgml/libpq.sgml +++ b/doc/src/sgml/libpq.sgml @@ -3130,8 +3130,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_SYNC</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a point in a - pipeline where a synchronization point has been established. + The <structname>PGresult</structname> represents a + synchronization point in pipeline mode, requested by + <xref linkend="libpq-PQsendPipeline"/>. This status occurs only when pipeline mode has been selected. </para> </listitem> @@ -3141,9 +3142,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_ABORTED</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a pipeline that's + The <structname>PGresult</structname> represents a pipeline that has received an error from the server. <function>PQgetResult</function> - must be called repeatedly, and it will return this status code, + must be called repeatedly, and each time it will return this status code until the end of the current pipeline, at which point it will return <literal>PGRES_PIPELINE_SYNC</literal> and normal processing can resume. @@ -4953,7 +4954,7 @@ int PQflush(PGconn *conn); <title>Using Pipeline Mode</title> <para> - To issue pipelines the application must switch a connection into pipeline mode. + To issue pipelines, the application must switch a connection into pipeline mode. Enter pipeline mode with <xref linkend="libpq-PQenterPipelineMode"/> or test whether pipeline mode is active with <xref linkend="libpq-PQpipelineStatus"/>. @@ -4975,7 +4976,7 @@ int PQflush(PGconn *conn); server will block trying to send results to the client from queries it has already processed. This only occurs when the client sends enough queries to fill both its output buffer and the server's receive - buffer before switching to processing input from the server, + buffer before it switches to processing input from the server, but it's hard to predict exactly when that will happen. </para> </footnote> @@ -5025,7 +5026,7 @@ int PQflush(PGconn *conn); <title>Processing Results</title> <para> - To process the result of one pipeline query, the application calls + To process the result of one query in a pipeline, the application calls <function>PQgetResult</function> repeatedly and handles each result until <function>PQgetResult</function> returns null. The result from the next query in the pipeline may then be retrieved using @@ -5057,10 +5058,10 @@ int PQflush(PGconn *conn); <type>PGresult</type> types <literal>PGRES_PIPELINE_SYNC</literal> and <literal>PGRES_PIPELINE_ABORTED</literal>. <literal>PGRES_PIPELINE_SYNC</literal> is reported exactly once for each - <function>PQsendPipeline</function> call at the corresponding point in - the result stream. + <function>PQsendPipeline</function> after retrieving results for all + queries in the pipeline. <literal>PGRES_PIPELINE_ABORTED</literal> is emitted in place of a normal - result stream result for the first error and all subsequent results + stream result for the first error and all subsequent results except <literal>PGRES_PIPELINE_SYNC</literal> and null; see <xref linkend="libpq-pipeline-errors"/>. </para> @@ -5075,7 +5076,8 @@ int PQflush(PGconn *conn); application about the query currently being processed (except that <function>PQgetResult</function> returns null to indicate that we start returning the results of next query). The application must keep track - of the order in which it sent queries and the expected results. + of the order in which it sent queries, to associate them with their + corresponding results. Applications will typically use a state machine or a FIFO queue for this. </para> @@ -5091,10 +5093,10 @@ int PQflush(PGconn *conn); </para> <para> - From the client perspective, after the client gets a - <literal>PGRES_FATAL_ERROR</literal> return from - <function>PQresultStatus</function> the pipeline is flagged as aborted. - <application>libpq</application> will report + From the client perspective, after <function>PQresultStatus</function> + returns <literal>PGRES_FATAL_ERROR</literal>, + the pipeline is flagged as aborted. + <function>PQresultStatus</function>, will report a <literal>PGRES_PIPELINE_ABORTED</literal> result for each remaining queued operation in an aborted pipeline. The result for <function>PQsendPipeline</function> is reported as @@ -5108,8 +5110,8 @@ int PQflush(PGconn *conn); </para> <para> - If the pipeline used an implicit transaction then operations that have - already executed are rolled back and operations that were queued for after + If the pipeline used an implicit transaction, then operations that have + already executed are rolled back and operations that were queued to follow the failed operation are skipped entirely. The same behaviour holds if the pipeline starts and commits a single explicit transaction (i.e. the first statement is <literal>BEGIN</literal> and the last is @@ -5145,11 +5147,11 @@ int PQflush(PGconn *conn); <para> The client application should generally maintain a queue of work - still to be dispatched and a queue of work that has been dispatched + remaining to be dispatched and a queue of work that has been dispatched but not yet had its results processed. When the socket is writable it should dispatch more work. When the socket is readable it should read results and process them, matching them up to the next entry in - its expected results queue. Based on available memory, results from + its expected results queue. Based on available memory, results from the socket should be read frequently: there's no need to wait until the pipeline end to read the results. Pipelines should be scoped to logical units of work, usually (but not necessarily) one transaction per pipeline. @@ -5191,8 +5193,8 @@ int PQflush(PGconn *conn); <listitem> <para> - Returns current pipeline mode status of the <application>libpq</application> - connection. + Returns the current pipeline mode status of the + <application>libpq</application> connection. <synopsis> PGpipelineStatus PQpipelineStatus(const PGconn *conn); </synopsis> @@ -5233,11 +5235,10 @@ PGpipelineStatus PQpipelineStatus(const PGconn *conn); <listitem> <para> The <application>libpq</application> connection is in pipeline - mode and an error has occurred while processing the current - pipeline. - The aborted flag is cleared as soon as the result - of the <function>PQsendPipeline</function> at the end of the aborted - pipeline is processed. Clients don't usually need this function to + mode and an error occurred while processing the current pipeline. + The aborted flag is cleared when <function>PQresultStatus</function> + returns PGRES_PIPELINE_SYNC at the end of the pipeline. + Clients don't usually need this function to verify aborted status, as they can tell that the pipeline is aborted from the <literal>PGRES_PIPELINE_ABORTED</literal> result code. </para> @@ -5328,7 +5329,7 @@ int PQsendPipeline(PGconn *conn); Returns 1 for success. Returns 0 if the connection is not in pipeline mode or sending a <link linkend="protocol-flow-ext-query">sync message</link> - is failed. + failed. </para> </listitem> </varlistentry> @@ -5349,7 +5350,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is most useful when the server is distant, i.e., network latency (<quote>ping time</quote>) is high, and also when many small operations - are being performed in rapid sequence. There is usually less benefit + are being performed in rapid succession. There is usually less benefit in using pipelined commands when each query takes many multiples of the client/server round-trip time to execute. A 100-statement operation run on a server 300ms round-trip-time away would take 30 seconds in network latency alone @@ -5367,7 +5368,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is not useful when information from one operation is required by the client to produce the next operation. In such cases, the client - must introduce a synchronization point and wait for a full client/server + would have to introduce a synchronization point and wait for a full client/server round-trip to get the results it needs. However, it's often possible to adjust the client design to exchange the required information server-side. Read-modify-write cycles are especially good candidates; for example: @@ -5392,10 +5393,10 @@ UPDATE mytable SET x = x + 1 WHERE id = 42; <note> <para> - The pipeline API was introduced in PostgreSQL 14, but clients using - the PostgreSQL 14 version of <application>libpq</application> can use - pipelines on server versions 7.4 and newer. Pipeline mode works on any server - that supports the v3 extended query protocol. + The pipeline API was introduced in <productname>PostgreSQL</productname> 14. + Pipeline mode is a client-side feature which doesn't require server + support, and works on any server that supports the v3 extended query + protocol. </para> </note> </sect2> diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c index 7ae7c14948..d7b036a35c 100644 --- a/src/interfaces/libpq/fe-exec.c +++ b/src/interfaces/libpq/fe-exec.c @@ -3803,7 +3803,7 @@ pqPipelineFlush(PGconn *conn) { if ((conn->pipelineStatus == PQ_PIPELINE_OFF) || (conn->outCount >= OUTBUFFER_THRESHOLD)) - return (pqFlush(conn)); + return pqFlush(conn); return 0; } diff --git a/src/test/modules/test_libpq/pipeline.c b/src/test/modules/test_libpq/pipeline.c index f4a2bdec57..01d5a9a8ff 100644 --- a/src/test/modules/test_libpq/pipeline.c +++ b/src/test/modules/test_libpq/pipeline.c @@ -126,7 +126,7 @@ test_simple_pipeline(PGconn *conn) * process the results of as they come in. * * For a simple case we should be able to do this without interim - * processing of results since our out buffer will give us enough slush to + * processing of results since our output buffer will give us enough slush to * work with and we won't block on sending. So blocking mode is fine. */ if (PQisnonblocking(conn)) @@ -603,7 +603,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) /* * Now we start inserting. We'll be sending enough data that we could fill - * our out buffer, so to avoid deadlocking we need to enter nonblocking + * our output buffer, so to avoid deadlocking we need to enter nonblocking * mode and consume input while we send more output. As results of each * query are processed we should pop them to allow processing of the next * query. There's no need to finish the pipeline before processing @@ -635,7 +635,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) } /* - * Process any results, so we keep the server's out buffer free + * Process any results, so we keep the server's output buffer free * flowing and it can continue to process input */ if (FD_ISSET(sock, &input_mask)) -- 2.17.0 --oC1+HKm2/end4ao3-- ^ permalink raw reply [nested|flat] 27+ messages in thread
* [PATCH 2/2] doc review @ 2021-03-03 22:36 Justin Pryzby <[email protected]> 0 siblings, 0 replies; 27+ messages in thread From: Justin Pryzby @ 2021-03-03 22:36 UTC (permalink / raw) --- doc/src/sgml/libpq.sgml | 67 +++++++++++++------------- src/interfaces/libpq/fe-exec.c | 2 +- src/test/modules/test_libpq/pipeline.c | 6 +-- 3 files changed, 38 insertions(+), 37 deletions(-) diff --git a/doc/src/sgml/libpq.sgml b/doc/src/sgml/libpq.sgml index a34ecbb144..e2865a218b 100644 --- a/doc/src/sgml/libpq.sgml +++ b/doc/src/sgml/libpq.sgml @@ -3130,8 +3130,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_SYNC</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a point in a - pipeline where a synchronization point has been established. + The <structname>PGresult</structname> represents a + synchronization point in pipeline mode, requested by + <xref linkend="libpq-PQsendPipeline"/>. This status occurs only when pipeline mode has been selected. </para> </listitem> @@ -3141,9 +3142,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_ABORTED</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a pipeline that's + The <structname>PGresult</structname> represents a pipeline that has received an error from the server. <function>PQgetResult</function> - must be called repeatedly, and it will return this status code, + must be called repeatedly, and each time it will return this status code until the end of the current pipeline, at which point it will return <literal>PGRES_PIPELINE_SYNC</literal> and normal processing can resume. @@ -4953,7 +4954,7 @@ int PQflush(PGconn *conn); <title>Using Pipeline Mode</title> <para> - To issue pipelines the application must switch a connection into pipeline mode. + To issue pipelines, the application must switch a connection into pipeline mode. Enter pipeline mode with <xref linkend="libpq-PQenterPipelineMode"/> or test whether pipeline mode is active with <xref linkend="libpq-PQpipelineStatus"/>. @@ -4975,7 +4976,7 @@ int PQflush(PGconn *conn); server will block trying to send results to the client from queries it has already processed. This only occurs when the client sends enough queries to fill both its output buffer and the server's receive - buffer before switching to processing input from the server, + buffer before it switches to processing input from the server, but it's hard to predict exactly when that will happen. </para> </footnote> @@ -5025,7 +5026,7 @@ int PQflush(PGconn *conn); <title>Processing Results</title> <para> - To process the result of one pipeline query, the application calls + To process the result of one query in a pipeline, the application calls <function>PQgetResult</function> repeatedly and handles each result until <function>PQgetResult</function> returns null. The result from the next query in the pipeline may then be retrieved using @@ -5057,10 +5058,10 @@ int PQflush(PGconn *conn); <type>PGresult</type> types <literal>PGRES_PIPELINE_SYNC</literal> and <literal>PGRES_PIPELINE_ABORTED</literal>. <literal>PGRES_PIPELINE_SYNC</literal> is reported exactly once for each - <function>PQsendPipeline</function> call at the corresponding point in - the result stream. + <function>PQsendPipeline</function> after retrieving results for all + queries in the pipeline. <literal>PGRES_PIPELINE_ABORTED</literal> is emitted in place of a normal - result stream result for the first error and all subsequent results + stream result for the first error and all subsequent results except <literal>PGRES_PIPELINE_SYNC</literal> and null; see <xref linkend="libpq-pipeline-errors"/>. </para> @@ -5075,7 +5076,8 @@ int PQflush(PGconn *conn); application about the query currently being processed (except that <function>PQgetResult</function> returns null to indicate that we start returning the results of next query). The application must keep track - of the order in which it sent queries and the expected results. + of the order in which it sent queries, to associate them with their + corresponding results. Applications will typically use a state machine or a FIFO queue for this. </para> @@ -5091,10 +5093,10 @@ int PQflush(PGconn *conn); </para> <para> - From the client perspective, after the client gets a - <literal>PGRES_FATAL_ERROR</literal> return from - <function>PQresultStatus</function> the pipeline is flagged as aborted. - <application>libpq</application> will report + From the client perspective, after <function>PQresultStatus</function> + returns <literal>PGRES_FATAL_ERROR</literal>, + the pipeline is flagged as aborted. + <function>PQresultStatus</function>, will report a <literal>PGRES_PIPELINE_ABORTED</literal> result for each remaining queued operation in an aborted pipeline. The result for <function>PQsendPipeline</function> is reported as @@ -5108,8 +5110,8 @@ int PQflush(PGconn *conn); </para> <para> - If the pipeline used an implicit transaction then operations that have - already executed are rolled back and operations that were queued for after + If the pipeline used an implicit transaction, then operations that have + already executed are rolled back and operations that were queued to follow the failed operation are skipped entirely. The same behaviour holds if the pipeline starts and commits a single explicit transaction (i.e. the first statement is <literal>BEGIN</literal> and the last is @@ -5145,11 +5147,11 @@ int PQflush(PGconn *conn); <para> The client application should generally maintain a queue of work - still to be dispatched and a queue of work that has been dispatched + remaining to be dispatched and a queue of work that has been dispatched but not yet had its results processed. When the socket is writable it should dispatch more work. When the socket is readable it should read results and process them, matching them up to the next entry in - its expected results queue. Based on available memory, results from + its expected results queue. Based on available memory, results from the socket should be read frequently: there's no need to wait until the pipeline end to read the results. Pipelines should be scoped to logical units of work, usually (but not necessarily) one transaction per pipeline. @@ -5191,8 +5193,8 @@ int PQflush(PGconn *conn); <listitem> <para> - Returns current pipeline mode status of the <application>libpq</application> - connection. + Returns the current pipeline mode status of the + <application>libpq</application> connection. <synopsis> PGpipelineStatus PQpipelineStatus(const PGconn *conn); </synopsis> @@ -5233,11 +5235,10 @@ PGpipelineStatus PQpipelineStatus(const PGconn *conn); <listitem> <para> The <application>libpq</application> connection is in pipeline - mode and an error has occurred while processing the current - pipeline. - The aborted flag is cleared as soon as the result - of the <function>PQsendPipeline</function> at the end of the aborted - pipeline is processed. Clients don't usually need this function to + mode and an error occurred while processing the current pipeline. + The aborted flag is cleared when <function>PQresultStatus</function> + returns PGRES_PIPELINE_SYNC at the end of the pipeline. + Clients don't usually need this function to verify aborted status, as they can tell that the pipeline is aborted from the <literal>PGRES_PIPELINE_ABORTED</literal> result code. </para> @@ -5328,7 +5329,7 @@ int PQsendPipeline(PGconn *conn); Returns 1 for success. Returns 0 if the connection is not in pipeline mode or sending a <link linkend="protocol-flow-ext-query">sync message</link> - is failed. + failed. </para> </listitem> </varlistentry> @@ -5349,7 +5350,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is most useful when the server is distant, i.e., network latency (<quote>ping time</quote>) is high, and also when many small operations - are being performed in rapid sequence. There is usually less benefit + are being performed in rapid succession. There is usually less benefit in using pipelined commands when each query takes many multiples of the client/server round-trip time to execute. A 100-statement operation run on a server 300ms round-trip-time away would take 30 seconds in network latency alone @@ -5367,7 +5368,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is not useful when information from one operation is required by the client to produce the next operation. In such cases, the client - must introduce a synchronization point and wait for a full client/server + would have to introduce a synchronization point and wait for a full client/server round-trip to get the results it needs. However, it's often possible to adjust the client design to exchange the required information server-side. Read-modify-write cycles are especially good candidates; for example: @@ -5392,10 +5393,10 @@ UPDATE mytable SET x = x + 1 WHERE id = 42; <note> <para> - The pipeline API was introduced in PostgreSQL 14, but clients using - the PostgreSQL 14 version of <application>libpq</application> can use - pipelines on server versions 7.4 and newer. Pipeline mode works on any server - that supports the v3 extended query protocol. + The pipeline API was introduced in <productname>PostgreSQL</productname> 14. + Pipeline mode is a client-side feature which doesn't require server + support, and works on any server that supports the v3 extended query + protocol. </para> </note> </sect2> diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c index 7ae7c14948..d7b036a35c 100644 --- a/src/interfaces/libpq/fe-exec.c +++ b/src/interfaces/libpq/fe-exec.c @@ -3803,7 +3803,7 @@ pqPipelineFlush(PGconn *conn) { if ((conn->pipelineStatus == PQ_PIPELINE_OFF) || (conn->outCount >= OUTBUFFER_THRESHOLD)) - return (pqFlush(conn)); + return pqFlush(conn); return 0; } diff --git a/src/test/modules/test_libpq/pipeline.c b/src/test/modules/test_libpq/pipeline.c index f4a2bdec57..01d5a9a8ff 100644 --- a/src/test/modules/test_libpq/pipeline.c +++ b/src/test/modules/test_libpq/pipeline.c @@ -126,7 +126,7 @@ test_simple_pipeline(PGconn *conn) * process the results of as they come in. * * For a simple case we should be able to do this without interim - * processing of results since our out buffer will give us enough slush to + * processing of results since our output buffer will give us enough slush to * work with and we won't block on sending. So blocking mode is fine. */ if (PQisnonblocking(conn)) @@ -603,7 +603,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) /* * Now we start inserting. We'll be sending enough data that we could fill - * our out buffer, so to avoid deadlocking we need to enter nonblocking + * our output buffer, so to avoid deadlocking we need to enter nonblocking * mode and consume input while we send more output. As results of each * query are processed we should pop them to allow processing of the next * query. There's no need to finish the pipeline before processing @@ -635,7 +635,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) } /* - * Process any results, so we keep the server's out buffer free + * Process any results, so we keep the server's output buffer free * flowing and it can continue to process input */ if (FD_ISSET(sock, &input_mask)) -- 2.17.0 --oC1+HKm2/end4ao3-- ^ permalink raw reply [nested|flat] 27+ messages in thread
* [PATCH 2/2] doc review @ 2021-03-03 22:36 Justin Pryzby <[email protected]> 0 siblings, 0 replies; 27+ messages in thread From: Justin Pryzby @ 2021-03-03 22:36 UTC (permalink / raw) --- doc/src/sgml/libpq.sgml | 67 +++++++++++++------------- src/interfaces/libpq/fe-exec.c | 2 +- src/test/modules/test_libpq/pipeline.c | 6 +-- 3 files changed, 38 insertions(+), 37 deletions(-) diff --git a/doc/src/sgml/libpq.sgml b/doc/src/sgml/libpq.sgml index a34ecbb144..e2865a218b 100644 --- a/doc/src/sgml/libpq.sgml +++ b/doc/src/sgml/libpq.sgml @@ -3130,8 +3130,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_SYNC</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a point in a - pipeline where a synchronization point has been established. + The <structname>PGresult</structname> represents a + synchronization point in pipeline mode, requested by + <xref linkend="libpq-PQsendPipeline"/>. This status occurs only when pipeline mode has been selected. </para> </listitem> @@ -3141,9 +3142,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_ABORTED</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a pipeline that's + The <structname>PGresult</structname> represents a pipeline that has received an error from the server. <function>PQgetResult</function> - must be called repeatedly, and it will return this status code, + must be called repeatedly, and each time it will return this status code until the end of the current pipeline, at which point it will return <literal>PGRES_PIPELINE_SYNC</literal> and normal processing can resume. @@ -4953,7 +4954,7 @@ int PQflush(PGconn *conn); <title>Using Pipeline Mode</title> <para> - To issue pipelines the application must switch a connection into pipeline mode. + To issue pipelines, the application must switch a connection into pipeline mode. Enter pipeline mode with <xref linkend="libpq-PQenterPipelineMode"/> or test whether pipeline mode is active with <xref linkend="libpq-PQpipelineStatus"/>. @@ -4975,7 +4976,7 @@ int PQflush(PGconn *conn); server will block trying to send results to the client from queries it has already processed. This only occurs when the client sends enough queries to fill both its output buffer and the server's receive - buffer before switching to processing input from the server, + buffer before it switches to processing input from the server, but it's hard to predict exactly when that will happen. </para> </footnote> @@ -5025,7 +5026,7 @@ int PQflush(PGconn *conn); <title>Processing Results</title> <para> - To process the result of one pipeline query, the application calls + To process the result of one query in a pipeline, the application calls <function>PQgetResult</function> repeatedly and handles each result until <function>PQgetResult</function> returns null. The result from the next query in the pipeline may then be retrieved using @@ -5057,10 +5058,10 @@ int PQflush(PGconn *conn); <type>PGresult</type> types <literal>PGRES_PIPELINE_SYNC</literal> and <literal>PGRES_PIPELINE_ABORTED</literal>. <literal>PGRES_PIPELINE_SYNC</literal> is reported exactly once for each - <function>PQsendPipeline</function> call at the corresponding point in - the result stream. + <function>PQsendPipeline</function> after retrieving results for all + queries in the pipeline. <literal>PGRES_PIPELINE_ABORTED</literal> is emitted in place of a normal - result stream result for the first error and all subsequent results + stream result for the first error and all subsequent results except <literal>PGRES_PIPELINE_SYNC</literal> and null; see <xref linkend="libpq-pipeline-errors"/>. </para> @@ -5075,7 +5076,8 @@ int PQflush(PGconn *conn); application about the query currently being processed (except that <function>PQgetResult</function> returns null to indicate that we start returning the results of next query). The application must keep track - of the order in which it sent queries and the expected results. + of the order in which it sent queries, to associate them with their + corresponding results. Applications will typically use a state machine or a FIFO queue for this. </para> @@ -5091,10 +5093,10 @@ int PQflush(PGconn *conn); </para> <para> - From the client perspective, after the client gets a - <literal>PGRES_FATAL_ERROR</literal> return from - <function>PQresultStatus</function> the pipeline is flagged as aborted. - <application>libpq</application> will report + From the client perspective, after <function>PQresultStatus</function> + returns <literal>PGRES_FATAL_ERROR</literal>, + the pipeline is flagged as aborted. + <function>PQresultStatus</function>, will report a <literal>PGRES_PIPELINE_ABORTED</literal> result for each remaining queued operation in an aborted pipeline. The result for <function>PQsendPipeline</function> is reported as @@ -5108,8 +5110,8 @@ int PQflush(PGconn *conn); </para> <para> - If the pipeline used an implicit transaction then operations that have - already executed are rolled back and operations that were queued for after + If the pipeline used an implicit transaction, then operations that have + already executed are rolled back and operations that were queued to follow the failed operation are skipped entirely. The same behaviour holds if the pipeline starts and commits a single explicit transaction (i.e. the first statement is <literal>BEGIN</literal> and the last is @@ -5145,11 +5147,11 @@ int PQflush(PGconn *conn); <para> The client application should generally maintain a queue of work - still to be dispatched and a queue of work that has been dispatched + remaining to be dispatched and a queue of work that has been dispatched but not yet had its results processed. When the socket is writable it should dispatch more work. When the socket is readable it should read results and process them, matching them up to the next entry in - its expected results queue. Based on available memory, results from + its expected results queue. Based on available memory, results from the socket should be read frequently: there's no need to wait until the pipeline end to read the results. Pipelines should be scoped to logical units of work, usually (but not necessarily) one transaction per pipeline. @@ -5191,8 +5193,8 @@ int PQflush(PGconn *conn); <listitem> <para> - Returns current pipeline mode status of the <application>libpq</application> - connection. + Returns the current pipeline mode status of the + <application>libpq</application> connection. <synopsis> PGpipelineStatus PQpipelineStatus(const PGconn *conn); </synopsis> @@ -5233,11 +5235,10 @@ PGpipelineStatus PQpipelineStatus(const PGconn *conn); <listitem> <para> The <application>libpq</application> connection is in pipeline - mode and an error has occurred while processing the current - pipeline. - The aborted flag is cleared as soon as the result - of the <function>PQsendPipeline</function> at the end of the aborted - pipeline is processed. Clients don't usually need this function to + mode and an error occurred while processing the current pipeline. + The aborted flag is cleared when <function>PQresultStatus</function> + returns PGRES_PIPELINE_SYNC at the end of the pipeline. + Clients don't usually need this function to verify aborted status, as they can tell that the pipeline is aborted from the <literal>PGRES_PIPELINE_ABORTED</literal> result code. </para> @@ -5328,7 +5329,7 @@ int PQsendPipeline(PGconn *conn); Returns 1 for success. Returns 0 if the connection is not in pipeline mode or sending a <link linkend="protocol-flow-ext-query">sync message</link> - is failed. + failed. </para> </listitem> </varlistentry> @@ -5349,7 +5350,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is most useful when the server is distant, i.e., network latency (<quote>ping time</quote>) is high, and also when many small operations - are being performed in rapid sequence. There is usually less benefit + are being performed in rapid succession. There is usually less benefit in using pipelined commands when each query takes many multiples of the client/server round-trip time to execute. A 100-statement operation run on a server 300ms round-trip-time away would take 30 seconds in network latency alone @@ -5367,7 +5368,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is not useful when information from one operation is required by the client to produce the next operation. In such cases, the client - must introduce a synchronization point and wait for a full client/server + would have to introduce a synchronization point and wait for a full client/server round-trip to get the results it needs. However, it's often possible to adjust the client design to exchange the required information server-side. Read-modify-write cycles are especially good candidates; for example: @@ -5392,10 +5393,10 @@ UPDATE mytable SET x = x + 1 WHERE id = 42; <note> <para> - The pipeline API was introduced in PostgreSQL 14, but clients using - the PostgreSQL 14 version of <application>libpq</application> can use - pipelines on server versions 7.4 and newer. Pipeline mode works on any server - that supports the v3 extended query protocol. + The pipeline API was introduced in <productname>PostgreSQL</productname> 14. + Pipeline mode is a client-side feature which doesn't require server + support, and works on any server that supports the v3 extended query + protocol. </para> </note> </sect2> diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c index 7ae7c14948..d7b036a35c 100644 --- a/src/interfaces/libpq/fe-exec.c +++ b/src/interfaces/libpq/fe-exec.c @@ -3803,7 +3803,7 @@ pqPipelineFlush(PGconn *conn) { if ((conn->pipelineStatus == PQ_PIPELINE_OFF) || (conn->outCount >= OUTBUFFER_THRESHOLD)) - return (pqFlush(conn)); + return pqFlush(conn); return 0; } diff --git a/src/test/modules/test_libpq/pipeline.c b/src/test/modules/test_libpq/pipeline.c index f4a2bdec57..01d5a9a8ff 100644 --- a/src/test/modules/test_libpq/pipeline.c +++ b/src/test/modules/test_libpq/pipeline.c @@ -126,7 +126,7 @@ test_simple_pipeline(PGconn *conn) * process the results of as they come in. * * For a simple case we should be able to do this without interim - * processing of results since our out buffer will give us enough slush to + * processing of results since our output buffer will give us enough slush to * work with and we won't block on sending. So blocking mode is fine. */ if (PQisnonblocking(conn)) @@ -603,7 +603,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) /* * Now we start inserting. We'll be sending enough data that we could fill - * our out buffer, so to avoid deadlocking we need to enter nonblocking + * our output buffer, so to avoid deadlocking we need to enter nonblocking * mode and consume input while we send more output. As results of each * query are processed we should pop them to allow processing of the next * query. There's no need to finish the pipeline before processing @@ -635,7 +635,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) } /* - * Process any results, so we keep the server's out buffer free + * Process any results, so we keep the server's output buffer free * flowing and it can continue to process input */ if (FD_ISSET(sock, &input_mask)) -- 2.17.0 --oC1+HKm2/end4ao3-- ^ permalink raw reply [nested|flat] 27+ messages in thread
* [PATCH 2/2] doc review @ 2021-03-03 22:36 Justin Pryzby <[email protected]> 0 siblings, 0 replies; 27+ messages in thread From: Justin Pryzby @ 2021-03-03 22:36 UTC (permalink / raw) --- doc/src/sgml/libpq.sgml | 67 +++++++++++++------------- src/interfaces/libpq/fe-exec.c | 2 +- src/test/modules/test_libpq/pipeline.c | 6 +-- 3 files changed, 38 insertions(+), 37 deletions(-) diff --git a/doc/src/sgml/libpq.sgml b/doc/src/sgml/libpq.sgml index a34ecbb144..e2865a218b 100644 --- a/doc/src/sgml/libpq.sgml +++ b/doc/src/sgml/libpq.sgml @@ -3130,8 +3130,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_SYNC</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a point in a - pipeline where a synchronization point has been established. + The <structname>PGresult</structname> represents a + synchronization point in pipeline mode, requested by + <xref linkend="libpq-PQsendPipeline"/>. This status occurs only when pipeline mode has been selected. </para> </listitem> @@ -3141,9 +3142,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_ABORTED</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a pipeline that's + The <structname>PGresult</structname> represents a pipeline that has received an error from the server. <function>PQgetResult</function> - must be called repeatedly, and it will return this status code, + must be called repeatedly, and each time it will return this status code until the end of the current pipeline, at which point it will return <literal>PGRES_PIPELINE_SYNC</literal> and normal processing can resume. @@ -4953,7 +4954,7 @@ int PQflush(PGconn *conn); <title>Using Pipeline Mode</title> <para> - To issue pipelines the application must switch a connection into pipeline mode. + To issue pipelines, the application must switch a connection into pipeline mode. Enter pipeline mode with <xref linkend="libpq-PQenterPipelineMode"/> or test whether pipeline mode is active with <xref linkend="libpq-PQpipelineStatus"/>. @@ -4975,7 +4976,7 @@ int PQflush(PGconn *conn); server will block trying to send results to the client from queries it has already processed. This only occurs when the client sends enough queries to fill both its output buffer and the server's receive - buffer before switching to processing input from the server, + buffer before it switches to processing input from the server, but it's hard to predict exactly when that will happen. </para> </footnote> @@ -5025,7 +5026,7 @@ int PQflush(PGconn *conn); <title>Processing Results</title> <para> - To process the result of one pipeline query, the application calls + To process the result of one query in a pipeline, the application calls <function>PQgetResult</function> repeatedly and handles each result until <function>PQgetResult</function> returns null. The result from the next query in the pipeline may then be retrieved using @@ -5057,10 +5058,10 @@ int PQflush(PGconn *conn); <type>PGresult</type> types <literal>PGRES_PIPELINE_SYNC</literal> and <literal>PGRES_PIPELINE_ABORTED</literal>. <literal>PGRES_PIPELINE_SYNC</literal> is reported exactly once for each - <function>PQsendPipeline</function> call at the corresponding point in - the result stream. + <function>PQsendPipeline</function> after retrieving results for all + queries in the pipeline. <literal>PGRES_PIPELINE_ABORTED</literal> is emitted in place of a normal - result stream result for the first error and all subsequent results + stream result for the first error and all subsequent results except <literal>PGRES_PIPELINE_SYNC</literal> and null; see <xref linkend="libpq-pipeline-errors"/>. </para> @@ -5075,7 +5076,8 @@ int PQflush(PGconn *conn); application about the query currently being processed (except that <function>PQgetResult</function> returns null to indicate that we start returning the results of next query). The application must keep track - of the order in which it sent queries and the expected results. + of the order in which it sent queries, to associate them with their + corresponding results. Applications will typically use a state machine or a FIFO queue for this. </para> @@ -5091,10 +5093,10 @@ int PQflush(PGconn *conn); </para> <para> - From the client perspective, after the client gets a - <literal>PGRES_FATAL_ERROR</literal> return from - <function>PQresultStatus</function> the pipeline is flagged as aborted. - <application>libpq</application> will report + From the client perspective, after <function>PQresultStatus</function> + returns <literal>PGRES_FATAL_ERROR</literal>, + the pipeline is flagged as aborted. + <function>PQresultStatus</function>, will report a <literal>PGRES_PIPELINE_ABORTED</literal> result for each remaining queued operation in an aborted pipeline. The result for <function>PQsendPipeline</function> is reported as @@ -5108,8 +5110,8 @@ int PQflush(PGconn *conn); </para> <para> - If the pipeline used an implicit transaction then operations that have - already executed are rolled back and operations that were queued for after + If the pipeline used an implicit transaction, then operations that have + already executed are rolled back and operations that were queued to follow the failed operation are skipped entirely. The same behaviour holds if the pipeline starts and commits a single explicit transaction (i.e. the first statement is <literal>BEGIN</literal> and the last is @@ -5145,11 +5147,11 @@ int PQflush(PGconn *conn); <para> The client application should generally maintain a queue of work - still to be dispatched and a queue of work that has been dispatched + remaining to be dispatched and a queue of work that has been dispatched but not yet had its results processed. When the socket is writable it should dispatch more work. When the socket is readable it should read results and process them, matching them up to the next entry in - its expected results queue. Based on available memory, results from + its expected results queue. Based on available memory, results from the socket should be read frequently: there's no need to wait until the pipeline end to read the results. Pipelines should be scoped to logical units of work, usually (but not necessarily) one transaction per pipeline. @@ -5191,8 +5193,8 @@ int PQflush(PGconn *conn); <listitem> <para> - Returns current pipeline mode status of the <application>libpq</application> - connection. + Returns the current pipeline mode status of the + <application>libpq</application> connection. <synopsis> PGpipelineStatus PQpipelineStatus(const PGconn *conn); </synopsis> @@ -5233,11 +5235,10 @@ PGpipelineStatus PQpipelineStatus(const PGconn *conn); <listitem> <para> The <application>libpq</application> connection is in pipeline - mode and an error has occurred while processing the current - pipeline. - The aborted flag is cleared as soon as the result - of the <function>PQsendPipeline</function> at the end of the aborted - pipeline is processed. Clients don't usually need this function to + mode and an error occurred while processing the current pipeline. + The aborted flag is cleared when <function>PQresultStatus</function> + returns PGRES_PIPELINE_SYNC at the end of the pipeline. + Clients don't usually need this function to verify aborted status, as they can tell that the pipeline is aborted from the <literal>PGRES_PIPELINE_ABORTED</literal> result code. </para> @@ -5328,7 +5329,7 @@ int PQsendPipeline(PGconn *conn); Returns 1 for success. Returns 0 if the connection is not in pipeline mode or sending a <link linkend="protocol-flow-ext-query">sync message</link> - is failed. + failed. </para> </listitem> </varlistentry> @@ -5349,7 +5350,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is most useful when the server is distant, i.e., network latency (<quote>ping time</quote>) is high, and also when many small operations - are being performed in rapid sequence. There is usually less benefit + are being performed in rapid succession. There is usually less benefit in using pipelined commands when each query takes many multiples of the client/server round-trip time to execute. A 100-statement operation run on a server 300ms round-trip-time away would take 30 seconds in network latency alone @@ -5367,7 +5368,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is not useful when information from one operation is required by the client to produce the next operation. In such cases, the client - must introduce a synchronization point and wait for a full client/server + would have to introduce a synchronization point and wait for a full client/server round-trip to get the results it needs. However, it's often possible to adjust the client design to exchange the required information server-side. Read-modify-write cycles are especially good candidates; for example: @@ -5392,10 +5393,10 @@ UPDATE mytable SET x = x + 1 WHERE id = 42; <note> <para> - The pipeline API was introduced in PostgreSQL 14, but clients using - the PostgreSQL 14 version of <application>libpq</application> can use - pipelines on server versions 7.4 and newer. Pipeline mode works on any server - that supports the v3 extended query protocol. + The pipeline API was introduced in <productname>PostgreSQL</productname> 14. + Pipeline mode is a client-side feature which doesn't require server + support, and works on any server that supports the v3 extended query + protocol. </para> </note> </sect2> diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c index 7ae7c14948..d7b036a35c 100644 --- a/src/interfaces/libpq/fe-exec.c +++ b/src/interfaces/libpq/fe-exec.c @@ -3803,7 +3803,7 @@ pqPipelineFlush(PGconn *conn) { if ((conn->pipelineStatus == PQ_PIPELINE_OFF) || (conn->outCount >= OUTBUFFER_THRESHOLD)) - return (pqFlush(conn)); + return pqFlush(conn); return 0; } diff --git a/src/test/modules/test_libpq/pipeline.c b/src/test/modules/test_libpq/pipeline.c index f4a2bdec57..01d5a9a8ff 100644 --- a/src/test/modules/test_libpq/pipeline.c +++ b/src/test/modules/test_libpq/pipeline.c @@ -126,7 +126,7 @@ test_simple_pipeline(PGconn *conn) * process the results of as they come in. * * For a simple case we should be able to do this without interim - * processing of results since our out buffer will give us enough slush to + * processing of results since our output buffer will give us enough slush to * work with and we won't block on sending. So blocking mode is fine. */ if (PQisnonblocking(conn)) @@ -603,7 +603,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) /* * Now we start inserting. We'll be sending enough data that we could fill - * our out buffer, so to avoid deadlocking we need to enter nonblocking + * our output buffer, so to avoid deadlocking we need to enter nonblocking * mode and consume input while we send more output. As results of each * query are processed we should pop them to allow processing of the next * query. There's no need to finish the pipeline before processing @@ -635,7 +635,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) } /* - * Process any results, so we keep the server's out buffer free + * Process any results, so we keep the server's output buffer free * flowing and it can continue to process input */ if (FD_ISSET(sock, &input_mask)) -- 2.17.0 --oC1+HKm2/end4ao3-- ^ permalink raw reply [nested|flat] 27+ messages in thread
* [PATCH 2/2] doc review @ 2021-03-03 22:36 Justin Pryzby <[email protected]> 0 siblings, 0 replies; 27+ messages in thread From: Justin Pryzby @ 2021-03-03 22:36 UTC (permalink / raw) --- doc/src/sgml/libpq.sgml | 67 +++++++++++++------------- src/interfaces/libpq/fe-exec.c | 2 +- src/test/modules/test_libpq/pipeline.c | 6 +-- 3 files changed, 38 insertions(+), 37 deletions(-) diff --git a/doc/src/sgml/libpq.sgml b/doc/src/sgml/libpq.sgml index a34ecbb144..e2865a218b 100644 --- a/doc/src/sgml/libpq.sgml +++ b/doc/src/sgml/libpq.sgml @@ -3130,8 +3130,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_SYNC</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a point in a - pipeline where a synchronization point has been established. + The <structname>PGresult</structname> represents a + synchronization point in pipeline mode, requested by + <xref linkend="libpq-PQsendPipeline"/>. This status occurs only when pipeline mode has been selected. </para> </listitem> @@ -3141,9 +3142,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_ABORTED</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a pipeline that's + The <structname>PGresult</structname> represents a pipeline that has received an error from the server. <function>PQgetResult</function> - must be called repeatedly, and it will return this status code, + must be called repeatedly, and each time it will return this status code until the end of the current pipeline, at which point it will return <literal>PGRES_PIPELINE_SYNC</literal> and normal processing can resume. @@ -4953,7 +4954,7 @@ int PQflush(PGconn *conn); <title>Using Pipeline Mode</title> <para> - To issue pipelines the application must switch a connection into pipeline mode. + To issue pipelines, the application must switch a connection into pipeline mode. Enter pipeline mode with <xref linkend="libpq-PQenterPipelineMode"/> or test whether pipeline mode is active with <xref linkend="libpq-PQpipelineStatus"/>. @@ -4975,7 +4976,7 @@ int PQflush(PGconn *conn); server will block trying to send results to the client from queries it has already processed. This only occurs when the client sends enough queries to fill both its output buffer and the server's receive - buffer before switching to processing input from the server, + buffer before it switches to processing input from the server, but it's hard to predict exactly when that will happen. </para> </footnote> @@ -5025,7 +5026,7 @@ int PQflush(PGconn *conn); <title>Processing Results</title> <para> - To process the result of one pipeline query, the application calls + To process the result of one query in a pipeline, the application calls <function>PQgetResult</function> repeatedly and handles each result until <function>PQgetResult</function> returns null. The result from the next query in the pipeline may then be retrieved using @@ -5057,10 +5058,10 @@ int PQflush(PGconn *conn); <type>PGresult</type> types <literal>PGRES_PIPELINE_SYNC</literal> and <literal>PGRES_PIPELINE_ABORTED</literal>. <literal>PGRES_PIPELINE_SYNC</literal> is reported exactly once for each - <function>PQsendPipeline</function> call at the corresponding point in - the result stream. + <function>PQsendPipeline</function> after retrieving results for all + queries in the pipeline. <literal>PGRES_PIPELINE_ABORTED</literal> is emitted in place of a normal - result stream result for the first error and all subsequent results + stream result for the first error and all subsequent results except <literal>PGRES_PIPELINE_SYNC</literal> and null; see <xref linkend="libpq-pipeline-errors"/>. </para> @@ -5075,7 +5076,8 @@ int PQflush(PGconn *conn); application about the query currently being processed (except that <function>PQgetResult</function> returns null to indicate that we start returning the results of next query). The application must keep track - of the order in which it sent queries and the expected results. + of the order in which it sent queries, to associate them with their + corresponding results. Applications will typically use a state machine or a FIFO queue for this. </para> @@ -5091,10 +5093,10 @@ int PQflush(PGconn *conn); </para> <para> - From the client perspective, after the client gets a - <literal>PGRES_FATAL_ERROR</literal> return from - <function>PQresultStatus</function> the pipeline is flagged as aborted. - <application>libpq</application> will report + From the client perspective, after <function>PQresultStatus</function> + returns <literal>PGRES_FATAL_ERROR</literal>, + the pipeline is flagged as aborted. + <function>PQresultStatus</function>, will report a <literal>PGRES_PIPELINE_ABORTED</literal> result for each remaining queued operation in an aborted pipeline. The result for <function>PQsendPipeline</function> is reported as @@ -5108,8 +5110,8 @@ int PQflush(PGconn *conn); </para> <para> - If the pipeline used an implicit transaction then operations that have - already executed are rolled back and operations that were queued for after + If the pipeline used an implicit transaction, then operations that have + already executed are rolled back and operations that were queued to follow the failed operation are skipped entirely. The same behaviour holds if the pipeline starts and commits a single explicit transaction (i.e. the first statement is <literal>BEGIN</literal> and the last is @@ -5145,11 +5147,11 @@ int PQflush(PGconn *conn); <para> The client application should generally maintain a queue of work - still to be dispatched and a queue of work that has been dispatched + remaining to be dispatched and a queue of work that has been dispatched but not yet had its results processed. When the socket is writable it should dispatch more work. When the socket is readable it should read results and process them, matching them up to the next entry in - its expected results queue. Based on available memory, results from + its expected results queue. Based on available memory, results from the socket should be read frequently: there's no need to wait until the pipeline end to read the results. Pipelines should be scoped to logical units of work, usually (but not necessarily) one transaction per pipeline. @@ -5191,8 +5193,8 @@ int PQflush(PGconn *conn); <listitem> <para> - Returns current pipeline mode status of the <application>libpq</application> - connection. + Returns the current pipeline mode status of the + <application>libpq</application> connection. <synopsis> PGpipelineStatus PQpipelineStatus(const PGconn *conn); </synopsis> @@ -5233,11 +5235,10 @@ PGpipelineStatus PQpipelineStatus(const PGconn *conn); <listitem> <para> The <application>libpq</application> connection is in pipeline - mode and an error has occurred while processing the current - pipeline. - The aborted flag is cleared as soon as the result - of the <function>PQsendPipeline</function> at the end of the aborted - pipeline is processed. Clients don't usually need this function to + mode and an error occurred while processing the current pipeline. + The aborted flag is cleared when <function>PQresultStatus</function> + returns PGRES_PIPELINE_SYNC at the end of the pipeline. + Clients don't usually need this function to verify aborted status, as they can tell that the pipeline is aborted from the <literal>PGRES_PIPELINE_ABORTED</literal> result code. </para> @@ -5328,7 +5329,7 @@ int PQsendPipeline(PGconn *conn); Returns 1 for success. Returns 0 if the connection is not in pipeline mode or sending a <link linkend="protocol-flow-ext-query">sync message</link> - is failed. + failed. </para> </listitem> </varlistentry> @@ -5349,7 +5350,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is most useful when the server is distant, i.e., network latency (<quote>ping time</quote>) is high, and also when many small operations - are being performed in rapid sequence. There is usually less benefit + are being performed in rapid succession. There is usually less benefit in using pipelined commands when each query takes many multiples of the client/server round-trip time to execute. A 100-statement operation run on a server 300ms round-trip-time away would take 30 seconds in network latency alone @@ -5367,7 +5368,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is not useful when information from one operation is required by the client to produce the next operation. In such cases, the client - must introduce a synchronization point and wait for a full client/server + would have to introduce a synchronization point and wait for a full client/server round-trip to get the results it needs. However, it's often possible to adjust the client design to exchange the required information server-side. Read-modify-write cycles are especially good candidates; for example: @@ -5392,10 +5393,10 @@ UPDATE mytable SET x = x + 1 WHERE id = 42; <note> <para> - The pipeline API was introduced in PostgreSQL 14, but clients using - the PostgreSQL 14 version of <application>libpq</application> can use - pipelines on server versions 7.4 and newer. Pipeline mode works on any server - that supports the v3 extended query protocol. + The pipeline API was introduced in <productname>PostgreSQL</productname> 14. + Pipeline mode is a client-side feature which doesn't require server + support, and works on any server that supports the v3 extended query + protocol. </para> </note> </sect2> diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c index 7ae7c14948..d7b036a35c 100644 --- a/src/interfaces/libpq/fe-exec.c +++ b/src/interfaces/libpq/fe-exec.c @@ -3803,7 +3803,7 @@ pqPipelineFlush(PGconn *conn) { if ((conn->pipelineStatus == PQ_PIPELINE_OFF) || (conn->outCount >= OUTBUFFER_THRESHOLD)) - return (pqFlush(conn)); + return pqFlush(conn); return 0; } diff --git a/src/test/modules/test_libpq/pipeline.c b/src/test/modules/test_libpq/pipeline.c index f4a2bdec57..01d5a9a8ff 100644 --- a/src/test/modules/test_libpq/pipeline.c +++ b/src/test/modules/test_libpq/pipeline.c @@ -126,7 +126,7 @@ test_simple_pipeline(PGconn *conn) * process the results of as they come in. * * For a simple case we should be able to do this without interim - * processing of results since our out buffer will give us enough slush to + * processing of results since our output buffer will give us enough slush to * work with and we won't block on sending. So blocking mode is fine. */ if (PQisnonblocking(conn)) @@ -603,7 +603,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) /* * Now we start inserting. We'll be sending enough data that we could fill - * our out buffer, so to avoid deadlocking we need to enter nonblocking + * our output buffer, so to avoid deadlocking we need to enter nonblocking * mode and consume input while we send more output. As results of each * query are processed we should pop them to allow processing of the next * query. There's no need to finish the pipeline before processing @@ -635,7 +635,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) } /* - * Process any results, so we keep the server's out buffer free + * Process any results, so we keep the server's output buffer free * flowing and it can continue to process input */ if (FD_ISSET(sock, &input_mask)) -- 2.17.0 --oC1+HKm2/end4ao3-- ^ permalink raw reply [nested|flat] 27+ messages in thread
* [PATCH 2/2] doc review @ 2021-03-03 22:36 Justin Pryzby <[email protected]> 0 siblings, 0 replies; 27+ messages in thread From: Justin Pryzby @ 2021-03-03 22:36 UTC (permalink / raw) --- doc/src/sgml/libpq.sgml | 67 +++++++++++++------------- src/interfaces/libpq/fe-exec.c | 2 +- src/test/modules/test_libpq/pipeline.c | 6 +-- 3 files changed, 38 insertions(+), 37 deletions(-) diff --git a/doc/src/sgml/libpq.sgml b/doc/src/sgml/libpq.sgml index a34ecbb144..e2865a218b 100644 --- a/doc/src/sgml/libpq.sgml +++ b/doc/src/sgml/libpq.sgml @@ -3130,8 +3130,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_SYNC</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a point in a - pipeline where a synchronization point has been established. + The <structname>PGresult</structname> represents a + synchronization point in pipeline mode, requested by + <xref linkend="libpq-PQsendPipeline"/>. This status occurs only when pipeline mode has been selected. </para> </listitem> @@ -3141,9 +3142,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_ABORTED</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a pipeline that's + The <structname>PGresult</structname> represents a pipeline that has received an error from the server. <function>PQgetResult</function> - must be called repeatedly, and it will return this status code, + must be called repeatedly, and each time it will return this status code until the end of the current pipeline, at which point it will return <literal>PGRES_PIPELINE_SYNC</literal> and normal processing can resume. @@ -4953,7 +4954,7 @@ int PQflush(PGconn *conn); <title>Using Pipeline Mode</title> <para> - To issue pipelines the application must switch a connection into pipeline mode. + To issue pipelines, the application must switch a connection into pipeline mode. Enter pipeline mode with <xref linkend="libpq-PQenterPipelineMode"/> or test whether pipeline mode is active with <xref linkend="libpq-PQpipelineStatus"/>. @@ -4975,7 +4976,7 @@ int PQflush(PGconn *conn); server will block trying to send results to the client from queries it has already processed. This only occurs when the client sends enough queries to fill both its output buffer and the server's receive - buffer before switching to processing input from the server, + buffer before it switches to processing input from the server, but it's hard to predict exactly when that will happen. </para> </footnote> @@ -5025,7 +5026,7 @@ int PQflush(PGconn *conn); <title>Processing Results</title> <para> - To process the result of one pipeline query, the application calls + To process the result of one query in a pipeline, the application calls <function>PQgetResult</function> repeatedly and handles each result until <function>PQgetResult</function> returns null. The result from the next query in the pipeline may then be retrieved using @@ -5057,10 +5058,10 @@ int PQflush(PGconn *conn); <type>PGresult</type> types <literal>PGRES_PIPELINE_SYNC</literal> and <literal>PGRES_PIPELINE_ABORTED</literal>. <literal>PGRES_PIPELINE_SYNC</literal> is reported exactly once for each - <function>PQsendPipeline</function> call at the corresponding point in - the result stream. + <function>PQsendPipeline</function> after retrieving results for all + queries in the pipeline. <literal>PGRES_PIPELINE_ABORTED</literal> is emitted in place of a normal - result stream result for the first error and all subsequent results + stream result for the first error and all subsequent results except <literal>PGRES_PIPELINE_SYNC</literal> and null; see <xref linkend="libpq-pipeline-errors"/>. </para> @@ -5075,7 +5076,8 @@ int PQflush(PGconn *conn); application about the query currently being processed (except that <function>PQgetResult</function> returns null to indicate that we start returning the results of next query). The application must keep track - of the order in which it sent queries and the expected results. + of the order in which it sent queries, to associate them with their + corresponding results. Applications will typically use a state machine or a FIFO queue for this. </para> @@ -5091,10 +5093,10 @@ int PQflush(PGconn *conn); </para> <para> - From the client perspective, after the client gets a - <literal>PGRES_FATAL_ERROR</literal> return from - <function>PQresultStatus</function> the pipeline is flagged as aborted. - <application>libpq</application> will report + From the client perspective, after <function>PQresultStatus</function> + returns <literal>PGRES_FATAL_ERROR</literal>, + the pipeline is flagged as aborted. + <function>PQresultStatus</function>, will report a <literal>PGRES_PIPELINE_ABORTED</literal> result for each remaining queued operation in an aborted pipeline. The result for <function>PQsendPipeline</function> is reported as @@ -5108,8 +5110,8 @@ int PQflush(PGconn *conn); </para> <para> - If the pipeline used an implicit transaction then operations that have - already executed are rolled back and operations that were queued for after + If the pipeline used an implicit transaction, then operations that have + already executed are rolled back and operations that were queued to follow the failed operation are skipped entirely. The same behaviour holds if the pipeline starts and commits a single explicit transaction (i.e. the first statement is <literal>BEGIN</literal> and the last is @@ -5145,11 +5147,11 @@ int PQflush(PGconn *conn); <para> The client application should generally maintain a queue of work - still to be dispatched and a queue of work that has been dispatched + remaining to be dispatched and a queue of work that has been dispatched but not yet had its results processed. When the socket is writable it should dispatch more work. When the socket is readable it should read results and process them, matching them up to the next entry in - its expected results queue. Based on available memory, results from + its expected results queue. Based on available memory, results from the socket should be read frequently: there's no need to wait until the pipeline end to read the results. Pipelines should be scoped to logical units of work, usually (but not necessarily) one transaction per pipeline. @@ -5191,8 +5193,8 @@ int PQflush(PGconn *conn); <listitem> <para> - Returns current pipeline mode status of the <application>libpq</application> - connection. + Returns the current pipeline mode status of the + <application>libpq</application> connection. <synopsis> PGpipelineStatus PQpipelineStatus(const PGconn *conn); </synopsis> @@ -5233,11 +5235,10 @@ PGpipelineStatus PQpipelineStatus(const PGconn *conn); <listitem> <para> The <application>libpq</application> connection is in pipeline - mode and an error has occurred while processing the current - pipeline. - The aborted flag is cleared as soon as the result - of the <function>PQsendPipeline</function> at the end of the aborted - pipeline is processed. Clients don't usually need this function to + mode and an error occurred while processing the current pipeline. + The aborted flag is cleared when <function>PQresultStatus</function> + returns PGRES_PIPELINE_SYNC at the end of the pipeline. + Clients don't usually need this function to verify aborted status, as they can tell that the pipeline is aborted from the <literal>PGRES_PIPELINE_ABORTED</literal> result code. </para> @@ -5328,7 +5329,7 @@ int PQsendPipeline(PGconn *conn); Returns 1 for success. Returns 0 if the connection is not in pipeline mode or sending a <link linkend="protocol-flow-ext-query">sync message</link> - is failed. + failed. </para> </listitem> </varlistentry> @@ -5349,7 +5350,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is most useful when the server is distant, i.e., network latency (<quote>ping time</quote>) is high, and also when many small operations - are being performed in rapid sequence. There is usually less benefit + are being performed in rapid succession. There is usually less benefit in using pipelined commands when each query takes many multiples of the client/server round-trip time to execute. A 100-statement operation run on a server 300ms round-trip-time away would take 30 seconds in network latency alone @@ -5367,7 +5368,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is not useful when information from one operation is required by the client to produce the next operation. In such cases, the client - must introduce a synchronization point and wait for a full client/server + would have to introduce a synchronization point and wait for a full client/server round-trip to get the results it needs. However, it's often possible to adjust the client design to exchange the required information server-side. Read-modify-write cycles are especially good candidates; for example: @@ -5392,10 +5393,10 @@ UPDATE mytable SET x = x + 1 WHERE id = 42; <note> <para> - The pipeline API was introduced in PostgreSQL 14, but clients using - the PostgreSQL 14 version of <application>libpq</application> can use - pipelines on server versions 7.4 and newer. Pipeline mode works on any server - that supports the v3 extended query protocol. + The pipeline API was introduced in <productname>PostgreSQL</productname> 14. + Pipeline mode is a client-side feature which doesn't require server + support, and works on any server that supports the v3 extended query + protocol. </para> </note> </sect2> diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c index 7ae7c14948..d7b036a35c 100644 --- a/src/interfaces/libpq/fe-exec.c +++ b/src/interfaces/libpq/fe-exec.c @@ -3803,7 +3803,7 @@ pqPipelineFlush(PGconn *conn) { if ((conn->pipelineStatus == PQ_PIPELINE_OFF) || (conn->outCount >= OUTBUFFER_THRESHOLD)) - return (pqFlush(conn)); + return pqFlush(conn); return 0; } diff --git a/src/test/modules/test_libpq/pipeline.c b/src/test/modules/test_libpq/pipeline.c index f4a2bdec57..01d5a9a8ff 100644 --- a/src/test/modules/test_libpq/pipeline.c +++ b/src/test/modules/test_libpq/pipeline.c @@ -126,7 +126,7 @@ test_simple_pipeline(PGconn *conn) * process the results of as they come in. * * For a simple case we should be able to do this without interim - * processing of results since our out buffer will give us enough slush to + * processing of results since our output buffer will give us enough slush to * work with and we won't block on sending. So blocking mode is fine. */ if (PQisnonblocking(conn)) @@ -603,7 +603,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) /* * Now we start inserting. We'll be sending enough data that we could fill - * our out buffer, so to avoid deadlocking we need to enter nonblocking + * our output buffer, so to avoid deadlocking we need to enter nonblocking * mode and consume input while we send more output. As results of each * query are processed we should pop them to allow processing of the next * query. There's no need to finish the pipeline before processing @@ -635,7 +635,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) } /* - * Process any results, so we keep the server's out buffer free + * Process any results, so we keep the server's output buffer free * flowing and it can continue to process input */ if (FD_ISSET(sock, &input_mask)) -- 2.17.0 --oC1+HKm2/end4ao3-- ^ permalink raw reply [nested|flat] 27+ messages in thread
* [PATCH 2/2] doc review @ 2021-03-03 22:36 Justin Pryzby <[email protected]> 0 siblings, 0 replies; 27+ messages in thread From: Justin Pryzby @ 2021-03-03 22:36 UTC (permalink / raw) --- doc/src/sgml/libpq.sgml | 67 +++++++++++++------------- src/interfaces/libpq/fe-exec.c | 2 +- src/test/modules/test_libpq/pipeline.c | 6 +-- 3 files changed, 38 insertions(+), 37 deletions(-) diff --git a/doc/src/sgml/libpq.sgml b/doc/src/sgml/libpq.sgml index a34ecbb144..e2865a218b 100644 --- a/doc/src/sgml/libpq.sgml +++ b/doc/src/sgml/libpq.sgml @@ -3130,8 +3130,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_SYNC</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a point in a - pipeline where a synchronization point has been established. + The <structname>PGresult</structname> represents a + synchronization point in pipeline mode, requested by + <xref linkend="libpq-PQsendPipeline"/>. This status occurs only when pipeline mode has been selected. </para> </listitem> @@ -3141,9 +3142,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_ABORTED</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a pipeline that's + The <structname>PGresult</structname> represents a pipeline that has received an error from the server. <function>PQgetResult</function> - must be called repeatedly, and it will return this status code, + must be called repeatedly, and each time it will return this status code until the end of the current pipeline, at which point it will return <literal>PGRES_PIPELINE_SYNC</literal> and normal processing can resume. @@ -4953,7 +4954,7 @@ int PQflush(PGconn *conn); <title>Using Pipeline Mode</title> <para> - To issue pipelines the application must switch a connection into pipeline mode. + To issue pipelines, the application must switch a connection into pipeline mode. Enter pipeline mode with <xref linkend="libpq-PQenterPipelineMode"/> or test whether pipeline mode is active with <xref linkend="libpq-PQpipelineStatus"/>. @@ -4975,7 +4976,7 @@ int PQflush(PGconn *conn); server will block trying to send results to the client from queries it has already processed. This only occurs when the client sends enough queries to fill both its output buffer and the server's receive - buffer before switching to processing input from the server, + buffer before it switches to processing input from the server, but it's hard to predict exactly when that will happen. </para> </footnote> @@ -5025,7 +5026,7 @@ int PQflush(PGconn *conn); <title>Processing Results</title> <para> - To process the result of one pipeline query, the application calls + To process the result of one query in a pipeline, the application calls <function>PQgetResult</function> repeatedly and handles each result until <function>PQgetResult</function> returns null. The result from the next query in the pipeline may then be retrieved using @@ -5057,10 +5058,10 @@ int PQflush(PGconn *conn); <type>PGresult</type> types <literal>PGRES_PIPELINE_SYNC</literal> and <literal>PGRES_PIPELINE_ABORTED</literal>. <literal>PGRES_PIPELINE_SYNC</literal> is reported exactly once for each - <function>PQsendPipeline</function> call at the corresponding point in - the result stream. + <function>PQsendPipeline</function> after retrieving results for all + queries in the pipeline. <literal>PGRES_PIPELINE_ABORTED</literal> is emitted in place of a normal - result stream result for the first error and all subsequent results + stream result for the first error and all subsequent results except <literal>PGRES_PIPELINE_SYNC</literal> and null; see <xref linkend="libpq-pipeline-errors"/>. </para> @@ -5075,7 +5076,8 @@ int PQflush(PGconn *conn); application about the query currently being processed (except that <function>PQgetResult</function> returns null to indicate that we start returning the results of next query). The application must keep track - of the order in which it sent queries and the expected results. + of the order in which it sent queries, to associate them with their + corresponding results. Applications will typically use a state machine or a FIFO queue for this. </para> @@ -5091,10 +5093,10 @@ int PQflush(PGconn *conn); </para> <para> - From the client perspective, after the client gets a - <literal>PGRES_FATAL_ERROR</literal> return from - <function>PQresultStatus</function> the pipeline is flagged as aborted. - <application>libpq</application> will report + From the client perspective, after <function>PQresultStatus</function> + returns <literal>PGRES_FATAL_ERROR</literal>, + the pipeline is flagged as aborted. + <function>PQresultStatus</function>, will report a <literal>PGRES_PIPELINE_ABORTED</literal> result for each remaining queued operation in an aborted pipeline. The result for <function>PQsendPipeline</function> is reported as @@ -5108,8 +5110,8 @@ int PQflush(PGconn *conn); </para> <para> - If the pipeline used an implicit transaction then operations that have - already executed are rolled back and operations that were queued for after + If the pipeline used an implicit transaction, then operations that have + already executed are rolled back and operations that were queued to follow the failed operation are skipped entirely. The same behaviour holds if the pipeline starts and commits a single explicit transaction (i.e. the first statement is <literal>BEGIN</literal> and the last is @@ -5145,11 +5147,11 @@ int PQflush(PGconn *conn); <para> The client application should generally maintain a queue of work - still to be dispatched and a queue of work that has been dispatched + remaining to be dispatched and a queue of work that has been dispatched but not yet had its results processed. When the socket is writable it should dispatch more work. When the socket is readable it should read results and process them, matching them up to the next entry in - its expected results queue. Based on available memory, results from + its expected results queue. Based on available memory, results from the socket should be read frequently: there's no need to wait until the pipeline end to read the results. Pipelines should be scoped to logical units of work, usually (but not necessarily) one transaction per pipeline. @@ -5191,8 +5193,8 @@ int PQflush(PGconn *conn); <listitem> <para> - Returns current pipeline mode status of the <application>libpq</application> - connection. + Returns the current pipeline mode status of the + <application>libpq</application> connection. <synopsis> PGpipelineStatus PQpipelineStatus(const PGconn *conn); </synopsis> @@ -5233,11 +5235,10 @@ PGpipelineStatus PQpipelineStatus(const PGconn *conn); <listitem> <para> The <application>libpq</application> connection is in pipeline - mode and an error has occurred while processing the current - pipeline. - The aborted flag is cleared as soon as the result - of the <function>PQsendPipeline</function> at the end of the aborted - pipeline is processed. Clients don't usually need this function to + mode and an error occurred while processing the current pipeline. + The aborted flag is cleared when <function>PQresultStatus</function> + returns PGRES_PIPELINE_SYNC at the end of the pipeline. + Clients don't usually need this function to verify aborted status, as they can tell that the pipeline is aborted from the <literal>PGRES_PIPELINE_ABORTED</literal> result code. </para> @@ -5328,7 +5329,7 @@ int PQsendPipeline(PGconn *conn); Returns 1 for success. Returns 0 if the connection is not in pipeline mode or sending a <link linkend="protocol-flow-ext-query">sync message</link> - is failed. + failed. </para> </listitem> </varlistentry> @@ -5349,7 +5350,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is most useful when the server is distant, i.e., network latency (<quote>ping time</quote>) is high, and also when many small operations - are being performed in rapid sequence. There is usually less benefit + are being performed in rapid succession. There is usually less benefit in using pipelined commands when each query takes many multiples of the client/server round-trip time to execute. A 100-statement operation run on a server 300ms round-trip-time away would take 30 seconds in network latency alone @@ -5367,7 +5368,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is not useful when information from one operation is required by the client to produce the next operation. In such cases, the client - must introduce a synchronization point and wait for a full client/server + would have to introduce a synchronization point and wait for a full client/server round-trip to get the results it needs. However, it's often possible to adjust the client design to exchange the required information server-side. Read-modify-write cycles are especially good candidates; for example: @@ -5392,10 +5393,10 @@ UPDATE mytable SET x = x + 1 WHERE id = 42; <note> <para> - The pipeline API was introduced in PostgreSQL 14, but clients using - the PostgreSQL 14 version of <application>libpq</application> can use - pipelines on server versions 7.4 and newer. Pipeline mode works on any server - that supports the v3 extended query protocol. + The pipeline API was introduced in <productname>PostgreSQL</productname> 14. + Pipeline mode is a client-side feature which doesn't require server + support, and works on any server that supports the v3 extended query + protocol. </para> </note> </sect2> diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c index 7ae7c14948..d7b036a35c 100644 --- a/src/interfaces/libpq/fe-exec.c +++ b/src/interfaces/libpq/fe-exec.c @@ -3803,7 +3803,7 @@ pqPipelineFlush(PGconn *conn) { if ((conn->pipelineStatus == PQ_PIPELINE_OFF) || (conn->outCount >= OUTBUFFER_THRESHOLD)) - return (pqFlush(conn)); + return pqFlush(conn); return 0; } diff --git a/src/test/modules/test_libpq/pipeline.c b/src/test/modules/test_libpq/pipeline.c index f4a2bdec57..01d5a9a8ff 100644 --- a/src/test/modules/test_libpq/pipeline.c +++ b/src/test/modules/test_libpq/pipeline.c @@ -126,7 +126,7 @@ test_simple_pipeline(PGconn *conn) * process the results of as they come in. * * For a simple case we should be able to do this without interim - * processing of results since our out buffer will give us enough slush to + * processing of results since our output buffer will give us enough slush to * work with and we won't block on sending. So blocking mode is fine. */ if (PQisnonblocking(conn)) @@ -603,7 +603,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) /* * Now we start inserting. We'll be sending enough data that we could fill - * our out buffer, so to avoid deadlocking we need to enter nonblocking + * our output buffer, so to avoid deadlocking we need to enter nonblocking * mode and consume input while we send more output. As results of each * query are processed we should pop them to allow processing of the next * query. There's no need to finish the pipeline before processing @@ -635,7 +635,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) } /* - * Process any results, so we keep the server's out buffer free + * Process any results, so we keep the server's output buffer free * flowing and it can continue to process input */ if (FD_ISSET(sock, &input_mask)) -- 2.17.0 --oC1+HKm2/end4ao3-- ^ permalink raw reply [nested|flat] 27+ messages in thread
* [PATCH 2/2] doc review @ 2021-03-03 22:36 Justin Pryzby <[email protected]> 0 siblings, 0 replies; 27+ messages in thread From: Justin Pryzby @ 2021-03-03 22:36 UTC (permalink / raw) --- doc/src/sgml/libpq.sgml | 67 +++++++++++++------------- src/interfaces/libpq/fe-exec.c | 2 +- src/test/modules/test_libpq/pipeline.c | 6 +-- 3 files changed, 38 insertions(+), 37 deletions(-) diff --git a/doc/src/sgml/libpq.sgml b/doc/src/sgml/libpq.sgml index a34ecbb144..e2865a218b 100644 --- a/doc/src/sgml/libpq.sgml +++ b/doc/src/sgml/libpq.sgml @@ -3130,8 +3130,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_SYNC</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a point in a - pipeline where a synchronization point has been established. + The <structname>PGresult</structname> represents a + synchronization point in pipeline mode, requested by + <xref linkend="libpq-PQsendPipeline"/>. This status occurs only when pipeline mode has been selected. </para> </listitem> @@ -3141,9 +3142,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_ABORTED</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a pipeline that's + The <structname>PGresult</structname> represents a pipeline that has received an error from the server. <function>PQgetResult</function> - must be called repeatedly, and it will return this status code, + must be called repeatedly, and each time it will return this status code until the end of the current pipeline, at which point it will return <literal>PGRES_PIPELINE_SYNC</literal> and normal processing can resume. @@ -4953,7 +4954,7 @@ int PQflush(PGconn *conn); <title>Using Pipeline Mode</title> <para> - To issue pipelines the application must switch a connection into pipeline mode. + To issue pipelines, the application must switch a connection into pipeline mode. Enter pipeline mode with <xref linkend="libpq-PQenterPipelineMode"/> or test whether pipeline mode is active with <xref linkend="libpq-PQpipelineStatus"/>. @@ -4975,7 +4976,7 @@ int PQflush(PGconn *conn); server will block trying to send results to the client from queries it has already processed. This only occurs when the client sends enough queries to fill both its output buffer and the server's receive - buffer before switching to processing input from the server, + buffer before it switches to processing input from the server, but it's hard to predict exactly when that will happen. </para> </footnote> @@ -5025,7 +5026,7 @@ int PQflush(PGconn *conn); <title>Processing Results</title> <para> - To process the result of one pipeline query, the application calls + To process the result of one query in a pipeline, the application calls <function>PQgetResult</function> repeatedly and handles each result until <function>PQgetResult</function> returns null. The result from the next query in the pipeline may then be retrieved using @@ -5057,10 +5058,10 @@ int PQflush(PGconn *conn); <type>PGresult</type> types <literal>PGRES_PIPELINE_SYNC</literal> and <literal>PGRES_PIPELINE_ABORTED</literal>. <literal>PGRES_PIPELINE_SYNC</literal> is reported exactly once for each - <function>PQsendPipeline</function> call at the corresponding point in - the result stream. + <function>PQsendPipeline</function> after retrieving results for all + queries in the pipeline. <literal>PGRES_PIPELINE_ABORTED</literal> is emitted in place of a normal - result stream result for the first error and all subsequent results + stream result for the first error and all subsequent results except <literal>PGRES_PIPELINE_SYNC</literal> and null; see <xref linkend="libpq-pipeline-errors"/>. </para> @@ -5075,7 +5076,8 @@ int PQflush(PGconn *conn); application about the query currently being processed (except that <function>PQgetResult</function> returns null to indicate that we start returning the results of next query). The application must keep track - of the order in which it sent queries and the expected results. + of the order in which it sent queries, to associate them with their + corresponding results. Applications will typically use a state machine or a FIFO queue for this. </para> @@ -5091,10 +5093,10 @@ int PQflush(PGconn *conn); </para> <para> - From the client perspective, after the client gets a - <literal>PGRES_FATAL_ERROR</literal> return from - <function>PQresultStatus</function> the pipeline is flagged as aborted. - <application>libpq</application> will report + From the client perspective, after <function>PQresultStatus</function> + returns <literal>PGRES_FATAL_ERROR</literal>, + the pipeline is flagged as aborted. + <function>PQresultStatus</function>, will report a <literal>PGRES_PIPELINE_ABORTED</literal> result for each remaining queued operation in an aborted pipeline. The result for <function>PQsendPipeline</function> is reported as @@ -5108,8 +5110,8 @@ int PQflush(PGconn *conn); </para> <para> - If the pipeline used an implicit transaction then operations that have - already executed are rolled back and operations that were queued for after + If the pipeline used an implicit transaction, then operations that have + already executed are rolled back and operations that were queued to follow the failed operation are skipped entirely. The same behaviour holds if the pipeline starts and commits a single explicit transaction (i.e. the first statement is <literal>BEGIN</literal> and the last is @@ -5145,11 +5147,11 @@ int PQflush(PGconn *conn); <para> The client application should generally maintain a queue of work - still to be dispatched and a queue of work that has been dispatched + remaining to be dispatched and a queue of work that has been dispatched but not yet had its results processed. When the socket is writable it should dispatch more work. When the socket is readable it should read results and process them, matching them up to the next entry in - its expected results queue. Based on available memory, results from + its expected results queue. Based on available memory, results from the socket should be read frequently: there's no need to wait until the pipeline end to read the results. Pipelines should be scoped to logical units of work, usually (but not necessarily) one transaction per pipeline. @@ -5191,8 +5193,8 @@ int PQflush(PGconn *conn); <listitem> <para> - Returns current pipeline mode status of the <application>libpq</application> - connection. + Returns the current pipeline mode status of the + <application>libpq</application> connection. <synopsis> PGpipelineStatus PQpipelineStatus(const PGconn *conn); </synopsis> @@ -5233,11 +5235,10 @@ PGpipelineStatus PQpipelineStatus(const PGconn *conn); <listitem> <para> The <application>libpq</application> connection is in pipeline - mode and an error has occurred while processing the current - pipeline. - The aborted flag is cleared as soon as the result - of the <function>PQsendPipeline</function> at the end of the aborted - pipeline is processed. Clients don't usually need this function to + mode and an error occurred while processing the current pipeline. + The aborted flag is cleared when <function>PQresultStatus</function> + returns PGRES_PIPELINE_SYNC at the end of the pipeline. + Clients don't usually need this function to verify aborted status, as they can tell that the pipeline is aborted from the <literal>PGRES_PIPELINE_ABORTED</literal> result code. </para> @@ -5328,7 +5329,7 @@ int PQsendPipeline(PGconn *conn); Returns 1 for success. Returns 0 if the connection is not in pipeline mode or sending a <link linkend="protocol-flow-ext-query">sync message</link> - is failed. + failed. </para> </listitem> </varlistentry> @@ -5349,7 +5350,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is most useful when the server is distant, i.e., network latency (<quote>ping time</quote>) is high, and also when many small operations - are being performed in rapid sequence. There is usually less benefit + are being performed in rapid succession. There is usually less benefit in using pipelined commands when each query takes many multiples of the client/server round-trip time to execute. A 100-statement operation run on a server 300ms round-trip-time away would take 30 seconds in network latency alone @@ -5367,7 +5368,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is not useful when information from one operation is required by the client to produce the next operation. In such cases, the client - must introduce a synchronization point and wait for a full client/server + would have to introduce a synchronization point and wait for a full client/server round-trip to get the results it needs. However, it's often possible to adjust the client design to exchange the required information server-side. Read-modify-write cycles are especially good candidates; for example: @@ -5392,10 +5393,10 @@ UPDATE mytable SET x = x + 1 WHERE id = 42; <note> <para> - The pipeline API was introduced in PostgreSQL 14, but clients using - the PostgreSQL 14 version of <application>libpq</application> can use - pipelines on server versions 7.4 and newer. Pipeline mode works on any server - that supports the v3 extended query protocol. + The pipeline API was introduced in <productname>PostgreSQL</productname> 14. + Pipeline mode is a client-side feature which doesn't require server + support, and works on any server that supports the v3 extended query + protocol. </para> </note> </sect2> diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c index 7ae7c14948..d7b036a35c 100644 --- a/src/interfaces/libpq/fe-exec.c +++ b/src/interfaces/libpq/fe-exec.c @@ -3803,7 +3803,7 @@ pqPipelineFlush(PGconn *conn) { if ((conn->pipelineStatus == PQ_PIPELINE_OFF) || (conn->outCount >= OUTBUFFER_THRESHOLD)) - return (pqFlush(conn)); + return pqFlush(conn); return 0; } diff --git a/src/test/modules/test_libpq/pipeline.c b/src/test/modules/test_libpq/pipeline.c index f4a2bdec57..01d5a9a8ff 100644 --- a/src/test/modules/test_libpq/pipeline.c +++ b/src/test/modules/test_libpq/pipeline.c @@ -126,7 +126,7 @@ test_simple_pipeline(PGconn *conn) * process the results of as they come in. * * For a simple case we should be able to do this without interim - * processing of results since our out buffer will give us enough slush to + * processing of results since our output buffer will give us enough slush to * work with and we won't block on sending. So blocking mode is fine. */ if (PQisnonblocking(conn)) @@ -603,7 +603,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) /* * Now we start inserting. We'll be sending enough data that we could fill - * our out buffer, so to avoid deadlocking we need to enter nonblocking + * our output buffer, so to avoid deadlocking we need to enter nonblocking * mode and consume input while we send more output. As results of each * query are processed we should pop them to allow processing of the next * query. There's no need to finish the pipeline before processing @@ -635,7 +635,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) } /* - * Process any results, so we keep the server's out buffer free + * Process any results, so we keep the server's output buffer free * flowing and it can continue to process input */ if (FD_ISSET(sock, &input_mask)) -- 2.17.0 --oC1+HKm2/end4ao3-- ^ permalink raw reply [nested|flat] 27+ messages in thread
* [PATCH 2/2] doc review @ 2021-03-03 22:36 Justin Pryzby <[email protected]> 0 siblings, 0 replies; 27+ messages in thread From: Justin Pryzby @ 2021-03-03 22:36 UTC (permalink / raw) --- doc/src/sgml/libpq.sgml | 67 +++++++++++++------------- src/interfaces/libpq/fe-exec.c | 2 +- src/test/modules/test_libpq/pipeline.c | 6 +-- 3 files changed, 38 insertions(+), 37 deletions(-) diff --git a/doc/src/sgml/libpq.sgml b/doc/src/sgml/libpq.sgml index a34ecbb144..e2865a218b 100644 --- a/doc/src/sgml/libpq.sgml +++ b/doc/src/sgml/libpq.sgml @@ -3130,8 +3130,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_SYNC</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a point in a - pipeline where a synchronization point has been established. + The <structname>PGresult</structname> represents a + synchronization point in pipeline mode, requested by + <xref linkend="libpq-PQsendPipeline"/>. This status occurs only when pipeline mode has been selected. </para> </listitem> @@ -3141,9 +3142,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_ABORTED</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a pipeline that's + The <structname>PGresult</structname> represents a pipeline that has received an error from the server. <function>PQgetResult</function> - must be called repeatedly, and it will return this status code, + must be called repeatedly, and each time it will return this status code until the end of the current pipeline, at which point it will return <literal>PGRES_PIPELINE_SYNC</literal> and normal processing can resume. @@ -4953,7 +4954,7 @@ int PQflush(PGconn *conn); <title>Using Pipeline Mode</title> <para> - To issue pipelines the application must switch a connection into pipeline mode. + To issue pipelines, the application must switch a connection into pipeline mode. Enter pipeline mode with <xref linkend="libpq-PQenterPipelineMode"/> or test whether pipeline mode is active with <xref linkend="libpq-PQpipelineStatus"/>. @@ -4975,7 +4976,7 @@ int PQflush(PGconn *conn); server will block trying to send results to the client from queries it has already processed. This only occurs when the client sends enough queries to fill both its output buffer and the server's receive - buffer before switching to processing input from the server, + buffer before it switches to processing input from the server, but it's hard to predict exactly when that will happen. </para> </footnote> @@ -5025,7 +5026,7 @@ int PQflush(PGconn *conn); <title>Processing Results</title> <para> - To process the result of one pipeline query, the application calls + To process the result of one query in a pipeline, the application calls <function>PQgetResult</function> repeatedly and handles each result until <function>PQgetResult</function> returns null. The result from the next query in the pipeline may then be retrieved using @@ -5057,10 +5058,10 @@ int PQflush(PGconn *conn); <type>PGresult</type> types <literal>PGRES_PIPELINE_SYNC</literal> and <literal>PGRES_PIPELINE_ABORTED</literal>. <literal>PGRES_PIPELINE_SYNC</literal> is reported exactly once for each - <function>PQsendPipeline</function> call at the corresponding point in - the result stream. + <function>PQsendPipeline</function> after retrieving results for all + queries in the pipeline. <literal>PGRES_PIPELINE_ABORTED</literal> is emitted in place of a normal - result stream result for the first error and all subsequent results + stream result for the first error and all subsequent results except <literal>PGRES_PIPELINE_SYNC</literal> and null; see <xref linkend="libpq-pipeline-errors"/>. </para> @@ -5075,7 +5076,8 @@ int PQflush(PGconn *conn); application about the query currently being processed (except that <function>PQgetResult</function> returns null to indicate that we start returning the results of next query). The application must keep track - of the order in which it sent queries and the expected results. + of the order in which it sent queries, to associate them with their + corresponding results. Applications will typically use a state machine or a FIFO queue for this. </para> @@ -5091,10 +5093,10 @@ int PQflush(PGconn *conn); </para> <para> - From the client perspective, after the client gets a - <literal>PGRES_FATAL_ERROR</literal> return from - <function>PQresultStatus</function> the pipeline is flagged as aborted. - <application>libpq</application> will report + From the client perspective, after <function>PQresultStatus</function> + returns <literal>PGRES_FATAL_ERROR</literal>, + the pipeline is flagged as aborted. + <function>PQresultStatus</function>, will report a <literal>PGRES_PIPELINE_ABORTED</literal> result for each remaining queued operation in an aborted pipeline. The result for <function>PQsendPipeline</function> is reported as @@ -5108,8 +5110,8 @@ int PQflush(PGconn *conn); </para> <para> - If the pipeline used an implicit transaction then operations that have - already executed are rolled back and operations that were queued for after + If the pipeline used an implicit transaction, then operations that have + already executed are rolled back and operations that were queued to follow the failed operation are skipped entirely. The same behaviour holds if the pipeline starts and commits a single explicit transaction (i.e. the first statement is <literal>BEGIN</literal> and the last is @@ -5145,11 +5147,11 @@ int PQflush(PGconn *conn); <para> The client application should generally maintain a queue of work - still to be dispatched and a queue of work that has been dispatched + remaining to be dispatched and a queue of work that has been dispatched but not yet had its results processed. When the socket is writable it should dispatch more work. When the socket is readable it should read results and process them, matching them up to the next entry in - its expected results queue. Based on available memory, results from + its expected results queue. Based on available memory, results from the socket should be read frequently: there's no need to wait until the pipeline end to read the results. Pipelines should be scoped to logical units of work, usually (but not necessarily) one transaction per pipeline. @@ -5191,8 +5193,8 @@ int PQflush(PGconn *conn); <listitem> <para> - Returns current pipeline mode status of the <application>libpq</application> - connection. + Returns the current pipeline mode status of the + <application>libpq</application> connection. <synopsis> PGpipelineStatus PQpipelineStatus(const PGconn *conn); </synopsis> @@ -5233,11 +5235,10 @@ PGpipelineStatus PQpipelineStatus(const PGconn *conn); <listitem> <para> The <application>libpq</application> connection is in pipeline - mode and an error has occurred while processing the current - pipeline. - The aborted flag is cleared as soon as the result - of the <function>PQsendPipeline</function> at the end of the aborted - pipeline is processed. Clients don't usually need this function to + mode and an error occurred while processing the current pipeline. + The aborted flag is cleared when <function>PQresultStatus</function> + returns PGRES_PIPELINE_SYNC at the end of the pipeline. + Clients don't usually need this function to verify aborted status, as they can tell that the pipeline is aborted from the <literal>PGRES_PIPELINE_ABORTED</literal> result code. </para> @@ -5328,7 +5329,7 @@ int PQsendPipeline(PGconn *conn); Returns 1 for success. Returns 0 if the connection is not in pipeline mode or sending a <link linkend="protocol-flow-ext-query">sync message</link> - is failed. + failed. </para> </listitem> </varlistentry> @@ -5349,7 +5350,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is most useful when the server is distant, i.e., network latency (<quote>ping time</quote>) is high, and also when many small operations - are being performed in rapid sequence. There is usually less benefit + are being performed in rapid succession. There is usually less benefit in using pipelined commands when each query takes many multiples of the client/server round-trip time to execute. A 100-statement operation run on a server 300ms round-trip-time away would take 30 seconds in network latency alone @@ -5367,7 +5368,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is not useful when information from one operation is required by the client to produce the next operation. In such cases, the client - must introduce a synchronization point and wait for a full client/server + would have to introduce a synchronization point and wait for a full client/server round-trip to get the results it needs. However, it's often possible to adjust the client design to exchange the required information server-side. Read-modify-write cycles are especially good candidates; for example: @@ -5392,10 +5393,10 @@ UPDATE mytable SET x = x + 1 WHERE id = 42; <note> <para> - The pipeline API was introduced in PostgreSQL 14, but clients using - the PostgreSQL 14 version of <application>libpq</application> can use - pipelines on server versions 7.4 and newer. Pipeline mode works on any server - that supports the v3 extended query protocol. + The pipeline API was introduced in <productname>PostgreSQL</productname> 14. + Pipeline mode is a client-side feature which doesn't require server + support, and works on any server that supports the v3 extended query + protocol. </para> </note> </sect2> diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c index 7ae7c14948..d7b036a35c 100644 --- a/src/interfaces/libpq/fe-exec.c +++ b/src/interfaces/libpq/fe-exec.c @@ -3803,7 +3803,7 @@ pqPipelineFlush(PGconn *conn) { if ((conn->pipelineStatus == PQ_PIPELINE_OFF) || (conn->outCount >= OUTBUFFER_THRESHOLD)) - return (pqFlush(conn)); + return pqFlush(conn); return 0; } diff --git a/src/test/modules/test_libpq/pipeline.c b/src/test/modules/test_libpq/pipeline.c index f4a2bdec57..01d5a9a8ff 100644 --- a/src/test/modules/test_libpq/pipeline.c +++ b/src/test/modules/test_libpq/pipeline.c @@ -126,7 +126,7 @@ test_simple_pipeline(PGconn *conn) * process the results of as they come in. * * For a simple case we should be able to do this without interim - * processing of results since our out buffer will give us enough slush to + * processing of results since our output buffer will give us enough slush to * work with and we won't block on sending. So blocking mode is fine. */ if (PQisnonblocking(conn)) @@ -603,7 +603,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) /* * Now we start inserting. We'll be sending enough data that we could fill - * our out buffer, so to avoid deadlocking we need to enter nonblocking + * our output buffer, so to avoid deadlocking we need to enter nonblocking * mode and consume input while we send more output. As results of each * query are processed we should pop them to allow processing of the next * query. There's no need to finish the pipeline before processing @@ -635,7 +635,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) } /* - * Process any results, so we keep the server's out buffer free + * Process any results, so we keep the server's output buffer free * flowing and it can continue to process input */ if (FD_ISSET(sock, &input_mask)) -- 2.17.0 --oC1+HKm2/end4ao3-- ^ permalink raw reply [nested|flat] 27+ messages in thread
* [PATCH 2/2] doc review @ 2021-03-03 22:36 Justin Pryzby <[email protected]> 0 siblings, 0 replies; 27+ messages in thread From: Justin Pryzby @ 2021-03-03 22:36 UTC (permalink / raw) --- doc/src/sgml/libpq.sgml | 67 +++++++++++++------------- src/interfaces/libpq/fe-exec.c | 2 +- src/test/modules/test_libpq/pipeline.c | 6 +-- 3 files changed, 38 insertions(+), 37 deletions(-) diff --git a/doc/src/sgml/libpq.sgml b/doc/src/sgml/libpq.sgml index a34ecbb144..e2865a218b 100644 --- a/doc/src/sgml/libpq.sgml +++ b/doc/src/sgml/libpq.sgml @@ -3130,8 +3130,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_SYNC</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a point in a - pipeline where a synchronization point has been established. + The <structname>PGresult</structname> represents a + synchronization point in pipeline mode, requested by + <xref linkend="libpq-PQsendPipeline"/>. This status occurs only when pipeline mode has been selected. </para> </listitem> @@ -3141,9 +3142,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_ABORTED</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a pipeline that's + The <structname>PGresult</structname> represents a pipeline that has received an error from the server. <function>PQgetResult</function> - must be called repeatedly, and it will return this status code, + must be called repeatedly, and each time it will return this status code until the end of the current pipeline, at which point it will return <literal>PGRES_PIPELINE_SYNC</literal> and normal processing can resume. @@ -4953,7 +4954,7 @@ int PQflush(PGconn *conn); <title>Using Pipeline Mode</title> <para> - To issue pipelines the application must switch a connection into pipeline mode. + To issue pipelines, the application must switch a connection into pipeline mode. Enter pipeline mode with <xref linkend="libpq-PQenterPipelineMode"/> or test whether pipeline mode is active with <xref linkend="libpq-PQpipelineStatus"/>. @@ -4975,7 +4976,7 @@ int PQflush(PGconn *conn); server will block trying to send results to the client from queries it has already processed. This only occurs when the client sends enough queries to fill both its output buffer and the server's receive - buffer before switching to processing input from the server, + buffer before it switches to processing input from the server, but it's hard to predict exactly when that will happen. </para> </footnote> @@ -5025,7 +5026,7 @@ int PQflush(PGconn *conn); <title>Processing Results</title> <para> - To process the result of one pipeline query, the application calls + To process the result of one query in a pipeline, the application calls <function>PQgetResult</function> repeatedly and handles each result until <function>PQgetResult</function> returns null. The result from the next query in the pipeline may then be retrieved using @@ -5057,10 +5058,10 @@ int PQflush(PGconn *conn); <type>PGresult</type> types <literal>PGRES_PIPELINE_SYNC</literal> and <literal>PGRES_PIPELINE_ABORTED</literal>. <literal>PGRES_PIPELINE_SYNC</literal> is reported exactly once for each - <function>PQsendPipeline</function> call at the corresponding point in - the result stream. + <function>PQsendPipeline</function> after retrieving results for all + queries in the pipeline. <literal>PGRES_PIPELINE_ABORTED</literal> is emitted in place of a normal - result stream result for the first error and all subsequent results + stream result for the first error and all subsequent results except <literal>PGRES_PIPELINE_SYNC</literal> and null; see <xref linkend="libpq-pipeline-errors"/>. </para> @@ -5075,7 +5076,8 @@ int PQflush(PGconn *conn); application about the query currently being processed (except that <function>PQgetResult</function> returns null to indicate that we start returning the results of next query). The application must keep track - of the order in which it sent queries and the expected results. + of the order in which it sent queries, to associate them with their + corresponding results. Applications will typically use a state machine or a FIFO queue for this. </para> @@ -5091,10 +5093,10 @@ int PQflush(PGconn *conn); </para> <para> - From the client perspective, after the client gets a - <literal>PGRES_FATAL_ERROR</literal> return from - <function>PQresultStatus</function> the pipeline is flagged as aborted. - <application>libpq</application> will report + From the client perspective, after <function>PQresultStatus</function> + returns <literal>PGRES_FATAL_ERROR</literal>, + the pipeline is flagged as aborted. + <function>PQresultStatus</function>, will report a <literal>PGRES_PIPELINE_ABORTED</literal> result for each remaining queued operation in an aborted pipeline. The result for <function>PQsendPipeline</function> is reported as @@ -5108,8 +5110,8 @@ int PQflush(PGconn *conn); </para> <para> - If the pipeline used an implicit transaction then operations that have - already executed are rolled back and operations that were queued for after + If the pipeline used an implicit transaction, then operations that have + already executed are rolled back and operations that were queued to follow the failed operation are skipped entirely. The same behaviour holds if the pipeline starts and commits a single explicit transaction (i.e. the first statement is <literal>BEGIN</literal> and the last is @@ -5145,11 +5147,11 @@ int PQflush(PGconn *conn); <para> The client application should generally maintain a queue of work - still to be dispatched and a queue of work that has been dispatched + remaining to be dispatched and a queue of work that has been dispatched but not yet had its results processed. When the socket is writable it should dispatch more work. When the socket is readable it should read results and process them, matching them up to the next entry in - its expected results queue. Based on available memory, results from + its expected results queue. Based on available memory, results from the socket should be read frequently: there's no need to wait until the pipeline end to read the results. Pipelines should be scoped to logical units of work, usually (but not necessarily) one transaction per pipeline. @@ -5191,8 +5193,8 @@ int PQflush(PGconn *conn); <listitem> <para> - Returns current pipeline mode status of the <application>libpq</application> - connection. + Returns the current pipeline mode status of the + <application>libpq</application> connection. <synopsis> PGpipelineStatus PQpipelineStatus(const PGconn *conn); </synopsis> @@ -5233,11 +5235,10 @@ PGpipelineStatus PQpipelineStatus(const PGconn *conn); <listitem> <para> The <application>libpq</application> connection is in pipeline - mode and an error has occurred while processing the current - pipeline. - The aborted flag is cleared as soon as the result - of the <function>PQsendPipeline</function> at the end of the aborted - pipeline is processed. Clients don't usually need this function to + mode and an error occurred while processing the current pipeline. + The aborted flag is cleared when <function>PQresultStatus</function> + returns PGRES_PIPELINE_SYNC at the end of the pipeline. + Clients don't usually need this function to verify aborted status, as they can tell that the pipeline is aborted from the <literal>PGRES_PIPELINE_ABORTED</literal> result code. </para> @@ -5328,7 +5329,7 @@ int PQsendPipeline(PGconn *conn); Returns 1 for success. Returns 0 if the connection is not in pipeline mode or sending a <link linkend="protocol-flow-ext-query">sync message</link> - is failed. + failed. </para> </listitem> </varlistentry> @@ -5349,7 +5350,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is most useful when the server is distant, i.e., network latency (<quote>ping time</quote>) is high, and also when many small operations - are being performed in rapid sequence. There is usually less benefit + are being performed in rapid succession. There is usually less benefit in using pipelined commands when each query takes many multiples of the client/server round-trip time to execute. A 100-statement operation run on a server 300ms round-trip-time away would take 30 seconds in network latency alone @@ -5367,7 +5368,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is not useful when information from one operation is required by the client to produce the next operation. In such cases, the client - must introduce a synchronization point and wait for a full client/server + would have to introduce a synchronization point and wait for a full client/server round-trip to get the results it needs. However, it's often possible to adjust the client design to exchange the required information server-side. Read-modify-write cycles are especially good candidates; for example: @@ -5392,10 +5393,10 @@ UPDATE mytable SET x = x + 1 WHERE id = 42; <note> <para> - The pipeline API was introduced in PostgreSQL 14, but clients using - the PostgreSQL 14 version of <application>libpq</application> can use - pipelines on server versions 7.4 and newer. Pipeline mode works on any server - that supports the v3 extended query protocol. + The pipeline API was introduced in <productname>PostgreSQL</productname> 14. + Pipeline mode is a client-side feature which doesn't require server + support, and works on any server that supports the v3 extended query + protocol. </para> </note> </sect2> diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c index 7ae7c14948..d7b036a35c 100644 --- a/src/interfaces/libpq/fe-exec.c +++ b/src/interfaces/libpq/fe-exec.c @@ -3803,7 +3803,7 @@ pqPipelineFlush(PGconn *conn) { if ((conn->pipelineStatus == PQ_PIPELINE_OFF) || (conn->outCount >= OUTBUFFER_THRESHOLD)) - return (pqFlush(conn)); + return pqFlush(conn); return 0; } diff --git a/src/test/modules/test_libpq/pipeline.c b/src/test/modules/test_libpq/pipeline.c index f4a2bdec57..01d5a9a8ff 100644 --- a/src/test/modules/test_libpq/pipeline.c +++ b/src/test/modules/test_libpq/pipeline.c @@ -126,7 +126,7 @@ test_simple_pipeline(PGconn *conn) * process the results of as they come in. * * For a simple case we should be able to do this without interim - * processing of results since our out buffer will give us enough slush to + * processing of results since our output buffer will give us enough slush to * work with and we won't block on sending. So blocking mode is fine. */ if (PQisnonblocking(conn)) @@ -603,7 +603,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) /* * Now we start inserting. We'll be sending enough data that we could fill - * our out buffer, so to avoid deadlocking we need to enter nonblocking + * our output buffer, so to avoid deadlocking we need to enter nonblocking * mode and consume input while we send more output. As results of each * query are processed we should pop them to allow processing of the next * query. There's no need to finish the pipeline before processing @@ -635,7 +635,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) } /* - * Process any results, so we keep the server's out buffer free + * Process any results, so we keep the server's output buffer free * flowing and it can continue to process input */ if (FD_ISSET(sock, &input_mask)) -- 2.17.0 --oC1+HKm2/end4ao3-- ^ permalink raw reply [nested|flat] 27+ messages in thread
* [PATCH 2/2] doc review @ 2021-03-03 22:36 Justin Pryzby <[email protected]> 0 siblings, 0 replies; 27+ messages in thread From: Justin Pryzby @ 2021-03-03 22:36 UTC (permalink / raw) --- doc/src/sgml/libpq.sgml | 67 +++++++++++++------------- src/interfaces/libpq/fe-exec.c | 2 +- src/test/modules/test_libpq/pipeline.c | 6 +-- 3 files changed, 38 insertions(+), 37 deletions(-) diff --git a/doc/src/sgml/libpq.sgml b/doc/src/sgml/libpq.sgml index a34ecbb144..e2865a218b 100644 --- a/doc/src/sgml/libpq.sgml +++ b/doc/src/sgml/libpq.sgml @@ -3130,8 +3130,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_SYNC</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a point in a - pipeline where a synchronization point has been established. + The <structname>PGresult</structname> represents a + synchronization point in pipeline mode, requested by + <xref linkend="libpq-PQsendPipeline"/>. This status occurs only when pipeline mode has been selected. </para> </listitem> @@ -3141,9 +3142,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_ABORTED</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a pipeline that's + The <structname>PGresult</structname> represents a pipeline that has received an error from the server. <function>PQgetResult</function> - must be called repeatedly, and it will return this status code, + must be called repeatedly, and each time it will return this status code until the end of the current pipeline, at which point it will return <literal>PGRES_PIPELINE_SYNC</literal> and normal processing can resume. @@ -4953,7 +4954,7 @@ int PQflush(PGconn *conn); <title>Using Pipeline Mode</title> <para> - To issue pipelines the application must switch a connection into pipeline mode. + To issue pipelines, the application must switch a connection into pipeline mode. Enter pipeline mode with <xref linkend="libpq-PQenterPipelineMode"/> or test whether pipeline mode is active with <xref linkend="libpq-PQpipelineStatus"/>. @@ -4975,7 +4976,7 @@ int PQflush(PGconn *conn); server will block trying to send results to the client from queries it has already processed. This only occurs when the client sends enough queries to fill both its output buffer and the server's receive - buffer before switching to processing input from the server, + buffer before it switches to processing input from the server, but it's hard to predict exactly when that will happen. </para> </footnote> @@ -5025,7 +5026,7 @@ int PQflush(PGconn *conn); <title>Processing Results</title> <para> - To process the result of one pipeline query, the application calls + To process the result of one query in a pipeline, the application calls <function>PQgetResult</function> repeatedly and handles each result until <function>PQgetResult</function> returns null. The result from the next query in the pipeline may then be retrieved using @@ -5057,10 +5058,10 @@ int PQflush(PGconn *conn); <type>PGresult</type> types <literal>PGRES_PIPELINE_SYNC</literal> and <literal>PGRES_PIPELINE_ABORTED</literal>. <literal>PGRES_PIPELINE_SYNC</literal> is reported exactly once for each - <function>PQsendPipeline</function> call at the corresponding point in - the result stream. + <function>PQsendPipeline</function> after retrieving results for all + queries in the pipeline. <literal>PGRES_PIPELINE_ABORTED</literal> is emitted in place of a normal - result stream result for the first error and all subsequent results + stream result for the first error and all subsequent results except <literal>PGRES_PIPELINE_SYNC</literal> and null; see <xref linkend="libpq-pipeline-errors"/>. </para> @@ -5075,7 +5076,8 @@ int PQflush(PGconn *conn); application about the query currently being processed (except that <function>PQgetResult</function> returns null to indicate that we start returning the results of next query). The application must keep track - of the order in which it sent queries and the expected results. + of the order in which it sent queries, to associate them with their + corresponding results. Applications will typically use a state machine or a FIFO queue for this. </para> @@ -5091,10 +5093,10 @@ int PQflush(PGconn *conn); </para> <para> - From the client perspective, after the client gets a - <literal>PGRES_FATAL_ERROR</literal> return from - <function>PQresultStatus</function> the pipeline is flagged as aborted. - <application>libpq</application> will report + From the client perspective, after <function>PQresultStatus</function> + returns <literal>PGRES_FATAL_ERROR</literal>, + the pipeline is flagged as aborted. + <function>PQresultStatus</function>, will report a <literal>PGRES_PIPELINE_ABORTED</literal> result for each remaining queued operation in an aborted pipeline. The result for <function>PQsendPipeline</function> is reported as @@ -5108,8 +5110,8 @@ int PQflush(PGconn *conn); </para> <para> - If the pipeline used an implicit transaction then operations that have - already executed are rolled back and operations that were queued for after + If the pipeline used an implicit transaction, then operations that have + already executed are rolled back and operations that were queued to follow the failed operation are skipped entirely. The same behaviour holds if the pipeline starts and commits a single explicit transaction (i.e. the first statement is <literal>BEGIN</literal> and the last is @@ -5145,11 +5147,11 @@ int PQflush(PGconn *conn); <para> The client application should generally maintain a queue of work - still to be dispatched and a queue of work that has been dispatched + remaining to be dispatched and a queue of work that has been dispatched but not yet had its results processed. When the socket is writable it should dispatch more work. When the socket is readable it should read results and process them, matching them up to the next entry in - its expected results queue. Based on available memory, results from + its expected results queue. Based on available memory, results from the socket should be read frequently: there's no need to wait until the pipeline end to read the results. Pipelines should be scoped to logical units of work, usually (but not necessarily) one transaction per pipeline. @@ -5191,8 +5193,8 @@ int PQflush(PGconn *conn); <listitem> <para> - Returns current pipeline mode status of the <application>libpq</application> - connection. + Returns the current pipeline mode status of the + <application>libpq</application> connection. <synopsis> PGpipelineStatus PQpipelineStatus(const PGconn *conn); </synopsis> @@ -5233,11 +5235,10 @@ PGpipelineStatus PQpipelineStatus(const PGconn *conn); <listitem> <para> The <application>libpq</application> connection is in pipeline - mode and an error has occurred while processing the current - pipeline. - The aborted flag is cleared as soon as the result - of the <function>PQsendPipeline</function> at the end of the aborted - pipeline is processed. Clients don't usually need this function to + mode and an error occurred while processing the current pipeline. + The aborted flag is cleared when <function>PQresultStatus</function> + returns PGRES_PIPELINE_SYNC at the end of the pipeline. + Clients don't usually need this function to verify aborted status, as they can tell that the pipeline is aborted from the <literal>PGRES_PIPELINE_ABORTED</literal> result code. </para> @@ -5328,7 +5329,7 @@ int PQsendPipeline(PGconn *conn); Returns 1 for success. Returns 0 if the connection is not in pipeline mode or sending a <link linkend="protocol-flow-ext-query">sync message</link> - is failed. + failed. </para> </listitem> </varlistentry> @@ -5349,7 +5350,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is most useful when the server is distant, i.e., network latency (<quote>ping time</quote>) is high, and also when many small operations - are being performed in rapid sequence. There is usually less benefit + are being performed in rapid succession. There is usually less benefit in using pipelined commands when each query takes many multiples of the client/server round-trip time to execute. A 100-statement operation run on a server 300ms round-trip-time away would take 30 seconds in network latency alone @@ -5367,7 +5368,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is not useful when information from one operation is required by the client to produce the next operation. In such cases, the client - must introduce a synchronization point and wait for a full client/server + would have to introduce a synchronization point and wait for a full client/server round-trip to get the results it needs. However, it's often possible to adjust the client design to exchange the required information server-side. Read-modify-write cycles are especially good candidates; for example: @@ -5392,10 +5393,10 @@ UPDATE mytable SET x = x + 1 WHERE id = 42; <note> <para> - The pipeline API was introduced in PostgreSQL 14, but clients using - the PostgreSQL 14 version of <application>libpq</application> can use - pipelines on server versions 7.4 and newer. Pipeline mode works on any server - that supports the v3 extended query protocol. + The pipeline API was introduced in <productname>PostgreSQL</productname> 14. + Pipeline mode is a client-side feature which doesn't require server + support, and works on any server that supports the v3 extended query + protocol. </para> </note> </sect2> diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c index 7ae7c14948..d7b036a35c 100644 --- a/src/interfaces/libpq/fe-exec.c +++ b/src/interfaces/libpq/fe-exec.c @@ -3803,7 +3803,7 @@ pqPipelineFlush(PGconn *conn) { if ((conn->pipelineStatus == PQ_PIPELINE_OFF) || (conn->outCount >= OUTBUFFER_THRESHOLD)) - return (pqFlush(conn)); + return pqFlush(conn); return 0; } diff --git a/src/test/modules/test_libpq/pipeline.c b/src/test/modules/test_libpq/pipeline.c index f4a2bdec57..01d5a9a8ff 100644 --- a/src/test/modules/test_libpq/pipeline.c +++ b/src/test/modules/test_libpq/pipeline.c @@ -126,7 +126,7 @@ test_simple_pipeline(PGconn *conn) * process the results of as they come in. * * For a simple case we should be able to do this without interim - * processing of results since our out buffer will give us enough slush to + * processing of results since our output buffer will give us enough slush to * work with and we won't block on sending. So blocking mode is fine. */ if (PQisnonblocking(conn)) @@ -603,7 +603,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) /* * Now we start inserting. We'll be sending enough data that we could fill - * our out buffer, so to avoid deadlocking we need to enter nonblocking + * our output buffer, so to avoid deadlocking we need to enter nonblocking * mode and consume input while we send more output. As results of each * query are processed we should pop them to allow processing of the next * query. There's no need to finish the pipeline before processing @@ -635,7 +635,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) } /* - * Process any results, so we keep the server's out buffer free + * Process any results, so we keep the server's output buffer free * flowing and it can continue to process input */ if (FD_ISSET(sock, &input_mask)) -- 2.17.0 --oC1+HKm2/end4ao3-- ^ permalink raw reply [nested|flat] 27+ messages in thread
* [PATCH 2/2] doc review @ 2021-03-03 22:36 Justin Pryzby <[email protected]> 0 siblings, 0 replies; 27+ messages in thread From: Justin Pryzby @ 2021-03-03 22:36 UTC (permalink / raw) --- doc/src/sgml/libpq.sgml | 67 +++++++++++++------------- src/interfaces/libpq/fe-exec.c | 2 +- src/test/modules/test_libpq/pipeline.c | 6 +-- 3 files changed, 38 insertions(+), 37 deletions(-) diff --git a/doc/src/sgml/libpq.sgml b/doc/src/sgml/libpq.sgml index a34ecbb144..e2865a218b 100644 --- a/doc/src/sgml/libpq.sgml +++ b/doc/src/sgml/libpq.sgml @@ -3130,8 +3130,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_SYNC</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a point in a - pipeline where a synchronization point has been established. + The <structname>PGresult</structname> represents a + synchronization point in pipeline mode, requested by + <xref linkend="libpq-PQsendPipeline"/>. This status occurs only when pipeline mode has been selected. </para> </listitem> @@ -3141,9 +3142,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_ABORTED</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a pipeline that's + The <structname>PGresult</structname> represents a pipeline that has received an error from the server. <function>PQgetResult</function> - must be called repeatedly, and it will return this status code, + must be called repeatedly, and each time it will return this status code until the end of the current pipeline, at which point it will return <literal>PGRES_PIPELINE_SYNC</literal> and normal processing can resume. @@ -4953,7 +4954,7 @@ int PQflush(PGconn *conn); <title>Using Pipeline Mode</title> <para> - To issue pipelines the application must switch a connection into pipeline mode. + To issue pipelines, the application must switch a connection into pipeline mode. Enter pipeline mode with <xref linkend="libpq-PQenterPipelineMode"/> or test whether pipeline mode is active with <xref linkend="libpq-PQpipelineStatus"/>. @@ -4975,7 +4976,7 @@ int PQflush(PGconn *conn); server will block trying to send results to the client from queries it has already processed. This only occurs when the client sends enough queries to fill both its output buffer and the server's receive - buffer before switching to processing input from the server, + buffer before it switches to processing input from the server, but it's hard to predict exactly when that will happen. </para> </footnote> @@ -5025,7 +5026,7 @@ int PQflush(PGconn *conn); <title>Processing Results</title> <para> - To process the result of one pipeline query, the application calls + To process the result of one query in a pipeline, the application calls <function>PQgetResult</function> repeatedly and handles each result until <function>PQgetResult</function> returns null. The result from the next query in the pipeline may then be retrieved using @@ -5057,10 +5058,10 @@ int PQflush(PGconn *conn); <type>PGresult</type> types <literal>PGRES_PIPELINE_SYNC</literal> and <literal>PGRES_PIPELINE_ABORTED</literal>. <literal>PGRES_PIPELINE_SYNC</literal> is reported exactly once for each - <function>PQsendPipeline</function> call at the corresponding point in - the result stream. + <function>PQsendPipeline</function> after retrieving results for all + queries in the pipeline. <literal>PGRES_PIPELINE_ABORTED</literal> is emitted in place of a normal - result stream result for the first error and all subsequent results + stream result for the first error and all subsequent results except <literal>PGRES_PIPELINE_SYNC</literal> and null; see <xref linkend="libpq-pipeline-errors"/>. </para> @@ -5075,7 +5076,8 @@ int PQflush(PGconn *conn); application about the query currently being processed (except that <function>PQgetResult</function> returns null to indicate that we start returning the results of next query). The application must keep track - of the order in which it sent queries and the expected results. + of the order in which it sent queries, to associate them with their + corresponding results. Applications will typically use a state machine or a FIFO queue for this. </para> @@ -5091,10 +5093,10 @@ int PQflush(PGconn *conn); </para> <para> - From the client perspective, after the client gets a - <literal>PGRES_FATAL_ERROR</literal> return from - <function>PQresultStatus</function> the pipeline is flagged as aborted. - <application>libpq</application> will report + From the client perspective, after <function>PQresultStatus</function> + returns <literal>PGRES_FATAL_ERROR</literal>, + the pipeline is flagged as aborted. + <function>PQresultStatus</function>, will report a <literal>PGRES_PIPELINE_ABORTED</literal> result for each remaining queued operation in an aborted pipeline. The result for <function>PQsendPipeline</function> is reported as @@ -5108,8 +5110,8 @@ int PQflush(PGconn *conn); </para> <para> - If the pipeline used an implicit transaction then operations that have - already executed are rolled back and operations that were queued for after + If the pipeline used an implicit transaction, then operations that have + already executed are rolled back and operations that were queued to follow the failed operation are skipped entirely. The same behaviour holds if the pipeline starts and commits a single explicit transaction (i.e. the first statement is <literal>BEGIN</literal> and the last is @@ -5145,11 +5147,11 @@ int PQflush(PGconn *conn); <para> The client application should generally maintain a queue of work - still to be dispatched and a queue of work that has been dispatched + remaining to be dispatched and a queue of work that has been dispatched but not yet had its results processed. When the socket is writable it should dispatch more work. When the socket is readable it should read results and process them, matching them up to the next entry in - its expected results queue. Based on available memory, results from + its expected results queue. Based on available memory, results from the socket should be read frequently: there's no need to wait until the pipeline end to read the results. Pipelines should be scoped to logical units of work, usually (but not necessarily) one transaction per pipeline. @@ -5191,8 +5193,8 @@ int PQflush(PGconn *conn); <listitem> <para> - Returns current pipeline mode status of the <application>libpq</application> - connection. + Returns the current pipeline mode status of the + <application>libpq</application> connection. <synopsis> PGpipelineStatus PQpipelineStatus(const PGconn *conn); </synopsis> @@ -5233,11 +5235,10 @@ PGpipelineStatus PQpipelineStatus(const PGconn *conn); <listitem> <para> The <application>libpq</application> connection is in pipeline - mode and an error has occurred while processing the current - pipeline. - The aborted flag is cleared as soon as the result - of the <function>PQsendPipeline</function> at the end of the aborted - pipeline is processed. Clients don't usually need this function to + mode and an error occurred while processing the current pipeline. + The aborted flag is cleared when <function>PQresultStatus</function> + returns PGRES_PIPELINE_SYNC at the end of the pipeline. + Clients don't usually need this function to verify aborted status, as they can tell that the pipeline is aborted from the <literal>PGRES_PIPELINE_ABORTED</literal> result code. </para> @@ -5328,7 +5329,7 @@ int PQsendPipeline(PGconn *conn); Returns 1 for success. Returns 0 if the connection is not in pipeline mode or sending a <link linkend="protocol-flow-ext-query">sync message</link> - is failed. + failed. </para> </listitem> </varlistentry> @@ -5349,7 +5350,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is most useful when the server is distant, i.e., network latency (<quote>ping time</quote>) is high, and also when many small operations - are being performed in rapid sequence. There is usually less benefit + are being performed in rapid succession. There is usually less benefit in using pipelined commands when each query takes many multiples of the client/server round-trip time to execute. A 100-statement operation run on a server 300ms round-trip-time away would take 30 seconds in network latency alone @@ -5367,7 +5368,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is not useful when information from one operation is required by the client to produce the next operation. In such cases, the client - must introduce a synchronization point and wait for a full client/server + would have to introduce a synchronization point and wait for a full client/server round-trip to get the results it needs. However, it's often possible to adjust the client design to exchange the required information server-side. Read-modify-write cycles are especially good candidates; for example: @@ -5392,10 +5393,10 @@ UPDATE mytable SET x = x + 1 WHERE id = 42; <note> <para> - The pipeline API was introduced in PostgreSQL 14, but clients using - the PostgreSQL 14 version of <application>libpq</application> can use - pipelines on server versions 7.4 and newer. Pipeline mode works on any server - that supports the v3 extended query protocol. + The pipeline API was introduced in <productname>PostgreSQL</productname> 14. + Pipeline mode is a client-side feature which doesn't require server + support, and works on any server that supports the v3 extended query + protocol. </para> </note> </sect2> diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c index 7ae7c14948..d7b036a35c 100644 --- a/src/interfaces/libpq/fe-exec.c +++ b/src/interfaces/libpq/fe-exec.c @@ -3803,7 +3803,7 @@ pqPipelineFlush(PGconn *conn) { if ((conn->pipelineStatus == PQ_PIPELINE_OFF) || (conn->outCount >= OUTBUFFER_THRESHOLD)) - return (pqFlush(conn)); + return pqFlush(conn); return 0; } diff --git a/src/test/modules/test_libpq/pipeline.c b/src/test/modules/test_libpq/pipeline.c index f4a2bdec57..01d5a9a8ff 100644 --- a/src/test/modules/test_libpq/pipeline.c +++ b/src/test/modules/test_libpq/pipeline.c @@ -126,7 +126,7 @@ test_simple_pipeline(PGconn *conn) * process the results of as they come in. * * For a simple case we should be able to do this without interim - * processing of results since our out buffer will give us enough slush to + * processing of results since our output buffer will give us enough slush to * work with and we won't block on sending. So blocking mode is fine. */ if (PQisnonblocking(conn)) @@ -603,7 +603,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) /* * Now we start inserting. We'll be sending enough data that we could fill - * our out buffer, so to avoid deadlocking we need to enter nonblocking + * our output buffer, so to avoid deadlocking we need to enter nonblocking * mode and consume input while we send more output. As results of each * query are processed we should pop them to allow processing of the next * query. There's no need to finish the pipeline before processing @@ -635,7 +635,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) } /* - * Process any results, so we keep the server's out buffer free + * Process any results, so we keep the server's output buffer free * flowing and it can continue to process input */ if (FD_ISSET(sock, &input_mask)) -- 2.17.0 --oC1+HKm2/end4ao3-- ^ permalink raw reply [nested|flat] 27+ messages in thread
* [PATCH 2/2] doc review @ 2021-03-03 22:36 Justin Pryzby <[email protected]> 0 siblings, 0 replies; 27+ messages in thread From: Justin Pryzby @ 2021-03-03 22:36 UTC (permalink / raw) --- doc/src/sgml/libpq.sgml | 67 +++++++++++++------------- src/interfaces/libpq/fe-exec.c | 2 +- src/test/modules/test_libpq/pipeline.c | 6 +-- 3 files changed, 38 insertions(+), 37 deletions(-) diff --git a/doc/src/sgml/libpq.sgml b/doc/src/sgml/libpq.sgml index a34ecbb144..e2865a218b 100644 --- a/doc/src/sgml/libpq.sgml +++ b/doc/src/sgml/libpq.sgml @@ -3130,8 +3130,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_SYNC</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a point in a - pipeline where a synchronization point has been established. + The <structname>PGresult</structname> represents a + synchronization point in pipeline mode, requested by + <xref linkend="libpq-PQsendPipeline"/>. This status occurs only when pipeline mode has been selected. </para> </listitem> @@ -3141,9 +3142,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_ABORTED</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a pipeline that's + The <structname>PGresult</structname> represents a pipeline that has received an error from the server. <function>PQgetResult</function> - must be called repeatedly, and it will return this status code, + must be called repeatedly, and each time it will return this status code until the end of the current pipeline, at which point it will return <literal>PGRES_PIPELINE_SYNC</literal> and normal processing can resume. @@ -4953,7 +4954,7 @@ int PQflush(PGconn *conn); <title>Using Pipeline Mode</title> <para> - To issue pipelines the application must switch a connection into pipeline mode. + To issue pipelines, the application must switch a connection into pipeline mode. Enter pipeline mode with <xref linkend="libpq-PQenterPipelineMode"/> or test whether pipeline mode is active with <xref linkend="libpq-PQpipelineStatus"/>. @@ -4975,7 +4976,7 @@ int PQflush(PGconn *conn); server will block trying to send results to the client from queries it has already processed. This only occurs when the client sends enough queries to fill both its output buffer and the server's receive - buffer before switching to processing input from the server, + buffer before it switches to processing input from the server, but it's hard to predict exactly when that will happen. </para> </footnote> @@ -5025,7 +5026,7 @@ int PQflush(PGconn *conn); <title>Processing Results</title> <para> - To process the result of one pipeline query, the application calls + To process the result of one query in a pipeline, the application calls <function>PQgetResult</function> repeatedly and handles each result until <function>PQgetResult</function> returns null. The result from the next query in the pipeline may then be retrieved using @@ -5057,10 +5058,10 @@ int PQflush(PGconn *conn); <type>PGresult</type> types <literal>PGRES_PIPELINE_SYNC</literal> and <literal>PGRES_PIPELINE_ABORTED</literal>. <literal>PGRES_PIPELINE_SYNC</literal> is reported exactly once for each - <function>PQsendPipeline</function> call at the corresponding point in - the result stream. + <function>PQsendPipeline</function> after retrieving results for all + queries in the pipeline. <literal>PGRES_PIPELINE_ABORTED</literal> is emitted in place of a normal - result stream result for the first error and all subsequent results + stream result for the first error and all subsequent results except <literal>PGRES_PIPELINE_SYNC</literal> and null; see <xref linkend="libpq-pipeline-errors"/>. </para> @@ -5075,7 +5076,8 @@ int PQflush(PGconn *conn); application about the query currently being processed (except that <function>PQgetResult</function> returns null to indicate that we start returning the results of next query). The application must keep track - of the order in which it sent queries and the expected results. + of the order in which it sent queries, to associate them with their + corresponding results. Applications will typically use a state machine or a FIFO queue for this. </para> @@ -5091,10 +5093,10 @@ int PQflush(PGconn *conn); </para> <para> - From the client perspective, after the client gets a - <literal>PGRES_FATAL_ERROR</literal> return from - <function>PQresultStatus</function> the pipeline is flagged as aborted. - <application>libpq</application> will report + From the client perspective, after <function>PQresultStatus</function> + returns <literal>PGRES_FATAL_ERROR</literal>, + the pipeline is flagged as aborted. + <function>PQresultStatus</function>, will report a <literal>PGRES_PIPELINE_ABORTED</literal> result for each remaining queued operation in an aborted pipeline. The result for <function>PQsendPipeline</function> is reported as @@ -5108,8 +5110,8 @@ int PQflush(PGconn *conn); </para> <para> - If the pipeline used an implicit transaction then operations that have - already executed are rolled back and operations that were queued for after + If the pipeline used an implicit transaction, then operations that have + already executed are rolled back and operations that were queued to follow the failed operation are skipped entirely. The same behaviour holds if the pipeline starts and commits a single explicit transaction (i.e. the first statement is <literal>BEGIN</literal> and the last is @@ -5145,11 +5147,11 @@ int PQflush(PGconn *conn); <para> The client application should generally maintain a queue of work - still to be dispatched and a queue of work that has been dispatched + remaining to be dispatched and a queue of work that has been dispatched but not yet had its results processed. When the socket is writable it should dispatch more work. When the socket is readable it should read results and process them, matching them up to the next entry in - its expected results queue. Based on available memory, results from + its expected results queue. Based on available memory, results from the socket should be read frequently: there's no need to wait until the pipeline end to read the results. Pipelines should be scoped to logical units of work, usually (but not necessarily) one transaction per pipeline. @@ -5191,8 +5193,8 @@ int PQflush(PGconn *conn); <listitem> <para> - Returns current pipeline mode status of the <application>libpq</application> - connection. + Returns the current pipeline mode status of the + <application>libpq</application> connection. <synopsis> PGpipelineStatus PQpipelineStatus(const PGconn *conn); </synopsis> @@ -5233,11 +5235,10 @@ PGpipelineStatus PQpipelineStatus(const PGconn *conn); <listitem> <para> The <application>libpq</application> connection is in pipeline - mode and an error has occurred while processing the current - pipeline. - The aborted flag is cleared as soon as the result - of the <function>PQsendPipeline</function> at the end of the aborted - pipeline is processed. Clients don't usually need this function to + mode and an error occurred while processing the current pipeline. + The aborted flag is cleared when <function>PQresultStatus</function> + returns PGRES_PIPELINE_SYNC at the end of the pipeline. + Clients don't usually need this function to verify aborted status, as they can tell that the pipeline is aborted from the <literal>PGRES_PIPELINE_ABORTED</literal> result code. </para> @@ -5328,7 +5329,7 @@ int PQsendPipeline(PGconn *conn); Returns 1 for success. Returns 0 if the connection is not in pipeline mode or sending a <link linkend="protocol-flow-ext-query">sync message</link> - is failed. + failed. </para> </listitem> </varlistentry> @@ -5349,7 +5350,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is most useful when the server is distant, i.e., network latency (<quote>ping time</quote>) is high, and also when many small operations - are being performed in rapid sequence. There is usually less benefit + are being performed in rapid succession. There is usually less benefit in using pipelined commands when each query takes many multiples of the client/server round-trip time to execute. A 100-statement operation run on a server 300ms round-trip-time away would take 30 seconds in network latency alone @@ -5367,7 +5368,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is not useful when information from one operation is required by the client to produce the next operation. In such cases, the client - must introduce a synchronization point and wait for a full client/server + would have to introduce a synchronization point and wait for a full client/server round-trip to get the results it needs. However, it's often possible to adjust the client design to exchange the required information server-side. Read-modify-write cycles are especially good candidates; for example: @@ -5392,10 +5393,10 @@ UPDATE mytable SET x = x + 1 WHERE id = 42; <note> <para> - The pipeline API was introduced in PostgreSQL 14, but clients using - the PostgreSQL 14 version of <application>libpq</application> can use - pipelines on server versions 7.4 and newer. Pipeline mode works on any server - that supports the v3 extended query protocol. + The pipeline API was introduced in <productname>PostgreSQL</productname> 14. + Pipeline mode is a client-side feature which doesn't require server + support, and works on any server that supports the v3 extended query + protocol. </para> </note> </sect2> diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c index 7ae7c14948..d7b036a35c 100644 --- a/src/interfaces/libpq/fe-exec.c +++ b/src/interfaces/libpq/fe-exec.c @@ -3803,7 +3803,7 @@ pqPipelineFlush(PGconn *conn) { if ((conn->pipelineStatus == PQ_PIPELINE_OFF) || (conn->outCount >= OUTBUFFER_THRESHOLD)) - return (pqFlush(conn)); + return pqFlush(conn); return 0; } diff --git a/src/test/modules/test_libpq/pipeline.c b/src/test/modules/test_libpq/pipeline.c index f4a2bdec57..01d5a9a8ff 100644 --- a/src/test/modules/test_libpq/pipeline.c +++ b/src/test/modules/test_libpq/pipeline.c @@ -126,7 +126,7 @@ test_simple_pipeline(PGconn *conn) * process the results of as they come in. * * For a simple case we should be able to do this without interim - * processing of results since our out buffer will give us enough slush to + * processing of results since our output buffer will give us enough slush to * work with and we won't block on sending. So blocking mode is fine. */ if (PQisnonblocking(conn)) @@ -603,7 +603,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) /* * Now we start inserting. We'll be sending enough data that we could fill - * our out buffer, so to avoid deadlocking we need to enter nonblocking + * our output buffer, so to avoid deadlocking we need to enter nonblocking * mode and consume input while we send more output. As results of each * query are processed we should pop them to allow processing of the next * query. There's no need to finish the pipeline before processing @@ -635,7 +635,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) } /* - * Process any results, so we keep the server's out buffer free + * Process any results, so we keep the server's output buffer free * flowing and it can continue to process input */ if (FD_ISSET(sock, &input_mask)) -- 2.17.0 --oC1+HKm2/end4ao3-- ^ permalink raw reply [nested|flat] 27+ messages in thread
* [PATCH 2/2] doc review @ 2021-03-03 22:36 Justin Pryzby <[email protected]> 0 siblings, 0 replies; 27+ messages in thread From: Justin Pryzby @ 2021-03-03 22:36 UTC (permalink / raw) --- doc/src/sgml/libpq.sgml | 67 +++++++++++++------------- src/interfaces/libpq/fe-exec.c | 2 +- src/test/modules/test_libpq/pipeline.c | 6 +-- 3 files changed, 38 insertions(+), 37 deletions(-) diff --git a/doc/src/sgml/libpq.sgml b/doc/src/sgml/libpq.sgml index a34ecbb144..e2865a218b 100644 --- a/doc/src/sgml/libpq.sgml +++ b/doc/src/sgml/libpq.sgml @@ -3130,8 +3130,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_SYNC</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a point in a - pipeline where a synchronization point has been established. + The <structname>PGresult</structname> represents a + synchronization point in pipeline mode, requested by + <xref linkend="libpq-PQsendPipeline"/>. This status occurs only when pipeline mode has been selected. </para> </listitem> @@ -3141,9 +3142,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_ABORTED</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a pipeline that's + The <structname>PGresult</structname> represents a pipeline that has received an error from the server. <function>PQgetResult</function> - must be called repeatedly, and it will return this status code, + must be called repeatedly, and each time it will return this status code until the end of the current pipeline, at which point it will return <literal>PGRES_PIPELINE_SYNC</literal> and normal processing can resume. @@ -4953,7 +4954,7 @@ int PQflush(PGconn *conn); <title>Using Pipeline Mode</title> <para> - To issue pipelines the application must switch a connection into pipeline mode. + To issue pipelines, the application must switch a connection into pipeline mode. Enter pipeline mode with <xref linkend="libpq-PQenterPipelineMode"/> or test whether pipeline mode is active with <xref linkend="libpq-PQpipelineStatus"/>. @@ -4975,7 +4976,7 @@ int PQflush(PGconn *conn); server will block trying to send results to the client from queries it has already processed. This only occurs when the client sends enough queries to fill both its output buffer and the server's receive - buffer before switching to processing input from the server, + buffer before it switches to processing input from the server, but it's hard to predict exactly when that will happen. </para> </footnote> @@ -5025,7 +5026,7 @@ int PQflush(PGconn *conn); <title>Processing Results</title> <para> - To process the result of one pipeline query, the application calls + To process the result of one query in a pipeline, the application calls <function>PQgetResult</function> repeatedly and handles each result until <function>PQgetResult</function> returns null. The result from the next query in the pipeline may then be retrieved using @@ -5057,10 +5058,10 @@ int PQflush(PGconn *conn); <type>PGresult</type> types <literal>PGRES_PIPELINE_SYNC</literal> and <literal>PGRES_PIPELINE_ABORTED</literal>. <literal>PGRES_PIPELINE_SYNC</literal> is reported exactly once for each - <function>PQsendPipeline</function> call at the corresponding point in - the result stream. + <function>PQsendPipeline</function> after retrieving results for all + queries in the pipeline. <literal>PGRES_PIPELINE_ABORTED</literal> is emitted in place of a normal - result stream result for the first error and all subsequent results + stream result for the first error and all subsequent results except <literal>PGRES_PIPELINE_SYNC</literal> and null; see <xref linkend="libpq-pipeline-errors"/>. </para> @@ -5075,7 +5076,8 @@ int PQflush(PGconn *conn); application about the query currently being processed (except that <function>PQgetResult</function> returns null to indicate that we start returning the results of next query). The application must keep track - of the order in which it sent queries and the expected results. + of the order in which it sent queries, to associate them with their + corresponding results. Applications will typically use a state machine or a FIFO queue for this. </para> @@ -5091,10 +5093,10 @@ int PQflush(PGconn *conn); </para> <para> - From the client perspective, after the client gets a - <literal>PGRES_FATAL_ERROR</literal> return from - <function>PQresultStatus</function> the pipeline is flagged as aborted. - <application>libpq</application> will report + From the client perspective, after <function>PQresultStatus</function> + returns <literal>PGRES_FATAL_ERROR</literal>, + the pipeline is flagged as aborted. + <function>PQresultStatus</function>, will report a <literal>PGRES_PIPELINE_ABORTED</literal> result for each remaining queued operation in an aborted pipeline. The result for <function>PQsendPipeline</function> is reported as @@ -5108,8 +5110,8 @@ int PQflush(PGconn *conn); </para> <para> - If the pipeline used an implicit transaction then operations that have - already executed are rolled back and operations that were queued for after + If the pipeline used an implicit transaction, then operations that have + already executed are rolled back and operations that were queued to follow the failed operation are skipped entirely. The same behaviour holds if the pipeline starts and commits a single explicit transaction (i.e. the first statement is <literal>BEGIN</literal> and the last is @@ -5145,11 +5147,11 @@ int PQflush(PGconn *conn); <para> The client application should generally maintain a queue of work - still to be dispatched and a queue of work that has been dispatched + remaining to be dispatched and a queue of work that has been dispatched but not yet had its results processed. When the socket is writable it should dispatch more work. When the socket is readable it should read results and process them, matching them up to the next entry in - its expected results queue. Based on available memory, results from + its expected results queue. Based on available memory, results from the socket should be read frequently: there's no need to wait until the pipeline end to read the results. Pipelines should be scoped to logical units of work, usually (but not necessarily) one transaction per pipeline. @@ -5191,8 +5193,8 @@ int PQflush(PGconn *conn); <listitem> <para> - Returns current pipeline mode status of the <application>libpq</application> - connection. + Returns the current pipeline mode status of the + <application>libpq</application> connection. <synopsis> PGpipelineStatus PQpipelineStatus(const PGconn *conn); </synopsis> @@ -5233,11 +5235,10 @@ PGpipelineStatus PQpipelineStatus(const PGconn *conn); <listitem> <para> The <application>libpq</application> connection is in pipeline - mode and an error has occurred while processing the current - pipeline. - The aborted flag is cleared as soon as the result - of the <function>PQsendPipeline</function> at the end of the aborted - pipeline is processed. Clients don't usually need this function to + mode and an error occurred while processing the current pipeline. + The aborted flag is cleared when <function>PQresultStatus</function> + returns PGRES_PIPELINE_SYNC at the end of the pipeline. + Clients don't usually need this function to verify aborted status, as they can tell that the pipeline is aborted from the <literal>PGRES_PIPELINE_ABORTED</literal> result code. </para> @@ -5328,7 +5329,7 @@ int PQsendPipeline(PGconn *conn); Returns 1 for success. Returns 0 if the connection is not in pipeline mode or sending a <link linkend="protocol-flow-ext-query">sync message</link> - is failed. + failed. </para> </listitem> </varlistentry> @@ -5349,7 +5350,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is most useful when the server is distant, i.e., network latency (<quote>ping time</quote>) is high, and also when many small operations - are being performed in rapid sequence. There is usually less benefit + are being performed in rapid succession. There is usually less benefit in using pipelined commands when each query takes many multiples of the client/server round-trip time to execute. A 100-statement operation run on a server 300ms round-trip-time away would take 30 seconds in network latency alone @@ -5367,7 +5368,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is not useful when information from one operation is required by the client to produce the next operation. In such cases, the client - must introduce a synchronization point and wait for a full client/server + would have to introduce a synchronization point and wait for a full client/server round-trip to get the results it needs. However, it's often possible to adjust the client design to exchange the required information server-side. Read-modify-write cycles are especially good candidates; for example: @@ -5392,10 +5393,10 @@ UPDATE mytable SET x = x + 1 WHERE id = 42; <note> <para> - The pipeline API was introduced in PostgreSQL 14, but clients using - the PostgreSQL 14 version of <application>libpq</application> can use - pipelines on server versions 7.4 and newer. Pipeline mode works on any server - that supports the v3 extended query protocol. + The pipeline API was introduced in <productname>PostgreSQL</productname> 14. + Pipeline mode is a client-side feature which doesn't require server + support, and works on any server that supports the v3 extended query + protocol. </para> </note> </sect2> diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c index 7ae7c14948..d7b036a35c 100644 --- a/src/interfaces/libpq/fe-exec.c +++ b/src/interfaces/libpq/fe-exec.c @@ -3803,7 +3803,7 @@ pqPipelineFlush(PGconn *conn) { if ((conn->pipelineStatus == PQ_PIPELINE_OFF) || (conn->outCount >= OUTBUFFER_THRESHOLD)) - return (pqFlush(conn)); + return pqFlush(conn); return 0; } diff --git a/src/test/modules/test_libpq/pipeline.c b/src/test/modules/test_libpq/pipeline.c index f4a2bdec57..01d5a9a8ff 100644 --- a/src/test/modules/test_libpq/pipeline.c +++ b/src/test/modules/test_libpq/pipeline.c @@ -126,7 +126,7 @@ test_simple_pipeline(PGconn *conn) * process the results of as they come in. * * For a simple case we should be able to do this without interim - * processing of results since our out buffer will give us enough slush to + * processing of results since our output buffer will give us enough slush to * work with and we won't block on sending. So blocking mode is fine. */ if (PQisnonblocking(conn)) @@ -603,7 +603,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) /* * Now we start inserting. We'll be sending enough data that we could fill - * our out buffer, so to avoid deadlocking we need to enter nonblocking + * our output buffer, so to avoid deadlocking we need to enter nonblocking * mode and consume input while we send more output. As results of each * query are processed we should pop them to allow processing of the next * query. There's no need to finish the pipeline before processing @@ -635,7 +635,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) } /* - * Process any results, so we keep the server's out buffer free + * Process any results, so we keep the server's output buffer free * flowing and it can continue to process input */ if (FD_ISSET(sock, &input_mask)) -- 2.17.0 --oC1+HKm2/end4ao3-- ^ permalink raw reply [nested|flat] 27+ messages in thread
* [PATCH 2/2] doc review @ 2021-03-03 22:36 Justin Pryzby <[email protected]> 0 siblings, 0 replies; 27+ messages in thread From: Justin Pryzby @ 2021-03-03 22:36 UTC (permalink / raw) --- doc/src/sgml/libpq.sgml | 67 +++++++++++++------------- src/interfaces/libpq/fe-exec.c | 2 +- src/test/modules/test_libpq/pipeline.c | 6 +-- 3 files changed, 38 insertions(+), 37 deletions(-) diff --git a/doc/src/sgml/libpq.sgml b/doc/src/sgml/libpq.sgml index a34ecbb144..e2865a218b 100644 --- a/doc/src/sgml/libpq.sgml +++ b/doc/src/sgml/libpq.sgml @@ -3130,8 +3130,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_SYNC</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a point in a - pipeline where a synchronization point has been established. + The <structname>PGresult</structname> represents a + synchronization point in pipeline mode, requested by + <xref linkend="libpq-PQsendPipeline"/>. This status occurs only when pipeline mode has been selected. </para> </listitem> @@ -3141,9 +3142,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_ABORTED</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a pipeline that's + The <structname>PGresult</structname> represents a pipeline that has received an error from the server. <function>PQgetResult</function> - must be called repeatedly, and it will return this status code, + must be called repeatedly, and each time it will return this status code until the end of the current pipeline, at which point it will return <literal>PGRES_PIPELINE_SYNC</literal> and normal processing can resume. @@ -4953,7 +4954,7 @@ int PQflush(PGconn *conn); <title>Using Pipeline Mode</title> <para> - To issue pipelines the application must switch a connection into pipeline mode. + To issue pipelines, the application must switch a connection into pipeline mode. Enter pipeline mode with <xref linkend="libpq-PQenterPipelineMode"/> or test whether pipeline mode is active with <xref linkend="libpq-PQpipelineStatus"/>. @@ -4975,7 +4976,7 @@ int PQflush(PGconn *conn); server will block trying to send results to the client from queries it has already processed. This only occurs when the client sends enough queries to fill both its output buffer and the server's receive - buffer before switching to processing input from the server, + buffer before it switches to processing input from the server, but it's hard to predict exactly when that will happen. </para> </footnote> @@ -5025,7 +5026,7 @@ int PQflush(PGconn *conn); <title>Processing Results</title> <para> - To process the result of one pipeline query, the application calls + To process the result of one query in a pipeline, the application calls <function>PQgetResult</function> repeatedly and handles each result until <function>PQgetResult</function> returns null. The result from the next query in the pipeline may then be retrieved using @@ -5057,10 +5058,10 @@ int PQflush(PGconn *conn); <type>PGresult</type> types <literal>PGRES_PIPELINE_SYNC</literal> and <literal>PGRES_PIPELINE_ABORTED</literal>. <literal>PGRES_PIPELINE_SYNC</literal> is reported exactly once for each - <function>PQsendPipeline</function> call at the corresponding point in - the result stream. + <function>PQsendPipeline</function> after retrieving results for all + queries in the pipeline. <literal>PGRES_PIPELINE_ABORTED</literal> is emitted in place of a normal - result stream result for the first error and all subsequent results + stream result for the first error and all subsequent results except <literal>PGRES_PIPELINE_SYNC</literal> and null; see <xref linkend="libpq-pipeline-errors"/>. </para> @@ -5075,7 +5076,8 @@ int PQflush(PGconn *conn); application about the query currently being processed (except that <function>PQgetResult</function> returns null to indicate that we start returning the results of next query). The application must keep track - of the order in which it sent queries and the expected results. + of the order in which it sent queries, to associate them with their + corresponding results. Applications will typically use a state machine or a FIFO queue for this. </para> @@ -5091,10 +5093,10 @@ int PQflush(PGconn *conn); </para> <para> - From the client perspective, after the client gets a - <literal>PGRES_FATAL_ERROR</literal> return from - <function>PQresultStatus</function> the pipeline is flagged as aborted. - <application>libpq</application> will report + From the client perspective, after <function>PQresultStatus</function> + returns <literal>PGRES_FATAL_ERROR</literal>, + the pipeline is flagged as aborted. + <function>PQresultStatus</function>, will report a <literal>PGRES_PIPELINE_ABORTED</literal> result for each remaining queued operation in an aborted pipeline. The result for <function>PQsendPipeline</function> is reported as @@ -5108,8 +5110,8 @@ int PQflush(PGconn *conn); </para> <para> - If the pipeline used an implicit transaction then operations that have - already executed are rolled back and operations that were queued for after + If the pipeline used an implicit transaction, then operations that have + already executed are rolled back and operations that were queued to follow the failed operation are skipped entirely. The same behaviour holds if the pipeline starts and commits a single explicit transaction (i.e. the first statement is <literal>BEGIN</literal> and the last is @@ -5145,11 +5147,11 @@ int PQflush(PGconn *conn); <para> The client application should generally maintain a queue of work - still to be dispatched and a queue of work that has been dispatched + remaining to be dispatched and a queue of work that has been dispatched but not yet had its results processed. When the socket is writable it should dispatch more work. When the socket is readable it should read results and process them, matching them up to the next entry in - its expected results queue. Based on available memory, results from + its expected results queue. Based on available memory, results from the socket should be read frequently: there's no need to wait until the pipeline end to read the results. Pipelines should be scoped to logical units of work, usually (but not necessarily) one transaction per pipeline. @@ -5191,8 +5193,8 @@ int PQflush(PGconn *conn); <listitem> <para> - Returns current pipeline mode status of the <application>libpq</application> - connection. + Returns the current pipeline mode status of the + <application>libpq</application> connection. <synopsis> PGpipelineStatus PQpipelineStatus(const PGconn *conn); </synopsis> @@ -5233,11 +5235,10 @@ PGpipelineStatus PQpipelineStatus(const PGconn *conn); <listitem> <para> The <application>libpq</application> connection is in pipeline - mode and an error has occurred while processing the current - pipeline. - The aborted flag is cleared as soon as the result - of the <function>PQsendPipeline</function> at the end of the aborted - pipeline is processed. Clients don't usually need this function to + mode and an error occurred while processing the current pipeline. + The aborted flag is cleared when <function>PQresultStatus</function> + returns PGRES_PIPELINE_SYNC at the end of the pipeline. + Clients don't usually need this function to verify aborted status, as they can tell that the pipeline is aborted from the <literal>PGRES_PIPELINE_ABORTED</literal> result code. </para> @@ -5328,7 +5329,7 @@ int PQsendPipeline(PGconn *conn); Returns 1 for success. Returns 0 if the connection is not in pipeline mode or sending a <link linkend="protocol-flow-ext-query">sync message</link> - is failed. + failed. </para> </listitem> </varlistentry> @@ -5349,7 +5350,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is most useful when the server is distant, i.e., network latency (<quote>ping time</quote>) is high, and also when many small operations - are being performed in rapid sequence. There is usually less benefit + are being performed in rapid succession. There is usually less benefit in using pipelined commands when each query takes many multiples of the client/server round-trip time to execute. A 100-statement operation run on a server 300ms round-trip-time away would take 30 seconds in network latency alone @@ -5367,7 +5368,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is not useful when information from one operation is required by the client to produce the next operation. In such cases, the client - must introduce a synchronization point and wait for a full client/server + would have to introduce a synchronization point and wait for a full client/server round-trip to get the results it needs. However, it's often possible to adjust the client design to exchange the required information server-side. Read-modify-write cycles are especially good candidates; for example: @@ -5392,10 +5393,10 @@ UPDATE mytable SET x = x + 1 WHERE id = 42; <note> <para> - The pipeline API was introduced in PostgreSQL 14, but clients using - the PostgreSQL 14 version of <application>libpq</application> can use - pipelines on server versions 7.4 and newer. Pipeline mode works on any server - that supports the v3 extended query protocol. + The pipeline API was introduced in <productname>PostgreSQL</productname> 14. + Pipeline mode is a client-side feature which doesn't require server + support, and works on any server that supports the v3 extended query + protocol. </para> </note> </sect2> diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c index 7ae7c14948..d7b036a35c 100644 --- a/src/interfaces/libpq/fe-exec.c +++ b/src/interfaces/libpq/fe-exec.c @@ -3803,7 +3803,7 @@ pqPipelineFlush(PGconn *conn) { if ((conn->pipelineStatus == PQ_PIPELINE_OFF) || (conn->outCount >= OUTBUFFER_THRESHOLD)) - return (pqFlush(conn)); + return pqFlush(conn); return 0; } diff --git a/src/test/modules/test_libpq/pipeline.c b/src/test/modules/test_libpq/pipeline.c index f4a2bdec57..01d5a9a8ff 100644 --- a/src/test/modules/test_libpq/pipeline.c +++ b/src/test/modules/test_libpq/pipeline.c @@ -126,7 +126,7 @@ test_simple_pipeline(PGconn *conn) * process the results of as they come in. * * For a simple case we should be able to do this without interim - * processing of results since our out buffer will give us enough slush to + * processing of results since our output buffer will give us enough slush to * work with and we won't block on sending. So blocking mode is fine. */ if (PQisnonblocking(conn)) @@ -603,7 +603,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) /* * Now we start inserting. We'll be sending enough data that we could fill - * our out buffer, so to avoid deadlocking we need to enter nonblocking + * our output buffer, so to avoid deadlocking we need to enter nonblocking * mode and consume input while we send more output. As results of each * query are processed we should pop them to allow processing of the next * query. There's no need to finish the pipeline before processing @@ -635,7 +635,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) } /* - * Process any results, so we keep the server's out buffer free + * Process any results, so we keep the server's output buffer free * flowing and it can continue to process input */ if (FD_ISSET(sock, &input_mask)) -- 2.17.0 --oC1+HKm2/end4ao3-- ^ permalink raw reply [nested|flat] 27+ messages in thread
* [PATCH 2/2] doc review @ 2021-03-03 22:36 Justin Pryzby <[email protected]> 0 siblings, 0 replies; 27+ messages in thread From: Justin Pryzby @ 2021-03-03 22:36 UTC (permalink / raw) --- doc/src/sgml/libpq.sgml | 67 +++++++++++++------------- src/interfaces/libpq/fe-exec.c | 2 +- src/test/modules/test_libpq/pipeline.c | 6 +-- 3 files changed, 38 insertions(+), 37 deletions(-) diff --git a/doc/src/sgml/libpq.sgml b/doc/src/sgml/libpq.sgml index a34ecbb144..e2865a218b 100644 --- a/doc/src/sgml/libpq.sgml +++ b/doc/src/sgml/libpq.sgml @@ -3130,8 +3130,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_SYNC</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a point in a - pipeline where a synchronization point has been established. + The <structname>PGresult</structname> represents a + synchronization point in pipeline mode, requested by + <xref linkend="libpq-PQsendPipeline"/>. This status occurs only when pipeline mode has been selected. </para> </listitem> @@ -3141,9 +3142,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_ABORTED</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a pipeline that's + The <structname>PGresult</structname> represents a pipeline that has received an error from the server. <function>PQgetResult</function> - must be called repeatedly, and it will return this status code, + must be called repeatedly, and each time it will return this status code until the end of the current pipeline, at which point it will return <literal>PGRES_PIPELINE_SYNC</literal> and normal processing can resume. @@ -4953,7 +4954,7 @@ int PQflush(PGconn *conn); <title>Using Pipeline Mode</title> <para> - To issue pipelines the application must switch a connection into pipeline mode. + To issue pipelines, the application must switch a connection into pipeline mode. Enter pipeline mode with <xref linkend="libpq-PQenterPipelineMode"/> or test whether pipeline mode is active with <xref linkend="libpq-PQpipelineStatus"/>. @@ -4975,7 +4976,7 @@ int PQflush(PGconn *conn); server will block trying to send results to the client from queries it has already processed. This only occurs when the client sends enough queries to fill both its output buffer and the server's receive - buffer before switching to processing input from the server, + buffer before it switches to processing input from the server, but it's hard to predict exactly when that will happen. </para> </footnote> @@ -5025,7 +5026,7 @@ int PQflush(PGconn *conn); <title>Processing Results</title> <para> - To process the result of one pipeline query, the application calls + To process the result of one query in a pipeline, the application calls <function>PQgetResult</function> repeatedly and handles each result until <function>PQgetResult</function> returns null. The result from the next query in the pipeline may then be retrieved using @@ -5057,10 +5058,10 @@ int PQflush(PGconn *conn); <type>PGresult</type> types <literal>PGRES_PIPELINE_SYNC</literal> and <literal>PGRES_PIPELINE_ABORTED</literal>. <literal>PGRES_PIPELINE_SYNC</literal> is reported exactly once for each - <function>PQsendPipeline</function> call at the corresponding point in - the result stream. + <function>PQsendPipeline</function> after retrieving results for all + queries in the pipeline. <literal>PGRES_PIPELINE_ABORTED</literal> is emitted in place of a normal - result stream result for the first error and all subsequent results + stream result for the first error and all subsequent results except <literal>PGRES_PIPELINE_SYNC</literal> and null; see <xref linkend="libpq-pipeline-errors"/>. </para> @@ -5075,7 +5076,8 @@ int PQflush(PGconn *conn); application about the query currently being processed (except that <function>PQgetResult</function> returns null to indicate that we start returning the results of next query). The application must keep track - of the order in which it sent queries and the expected results. + of the order in which it sent queries, to associate them with their + corresponding results. Applications will typically use a state machine or a FIFO queue for this. </para> @@ -5091,10 +5093,10 @@ int PQflush(PGconn *conn); </para> <para> - From the client perspective, after the client gets a - <literal>PGRES_FATAL_ERROR</literal> return from - <function>PQresultStatus</function> the pipeline is flagged as aborted. - <application>libpq</application> will report + From the client perspective, after <function>PQresultStatus</function> + returns <literal>PGRES_FATAL_ERROR</literal>, + the pipeline is flagged as aborted. + <function>PQresultStatus</function>, will report a <literal>PGRES_PIPELINE_ABORTED</literal> result for each remaining queued operation in an aborted pipeline. The result for <function>PQsendPipeline</function> is reported as @@ -5108,8 +5110,8 @@ int PQflush(PGconn *conn); </para> <para> - If the pipeline used an implicit transaction then operations that have - already executed are rolled back and operations that were queued for after + If the pipeline used an implicit transaction, then operations that have + already executed are rolled back and operations that were queued to follow the failed operation are skipped entirely. The same behaviour holds if the pipeline starts and commits a single explicit transaction (i.e. the first statement is <literal>BEGIN</literal> and the last is @@ -5145,11 +5147,11 @@ int PQflush(PGconn *conn); <para> The client application should generally maintain a queue of work - still to be dispatched and a queue of work that has been dispatched + remaining to be dispatched and a queue of work that has been dispatched but not yet had its results processed. When the socket is writable it should dispatch more work. When the socket is readable it should read results and process them, matching them up to the next entry in - its expected results queue. Based on available memory, results from + its expected results queue. Based on available memory, results from the socket should be read frequently: there's no need to wait until the pipeline end to read the results. Pipelines should be scoped to logical units of work, usually (but not necessarily) one transaction per pipeline. @@ -5191,8 +5193,8 @@ int PQflush(PGconn *conn); <listitem> <para> - Returns current pipeline mode status of the <application>libpq</application> - connection. + Returns the current pipeline mode status of the + <application>libpq</application> connection. <synopsis> PGpipelineStatus PQpipelineStatus(const PGconn *conn); </synopsis> @@ -5233,11 +5235,10 @@ PGpipelineStatus PQpipelineStatus(const PGconn *conn); <listitem> <para> The <application>libpq</application> connection is in pipeline - mode and an error has occurred while processing the current - pipeline. - The aborted flag is cleared as soon as the result - of the <function>PQsendPipeline</function> at the end of the aborted - pipeline is processed. Clients don't usually need this function to + mode and an error occurred while processing the current pipeline. + The aborted flag is cleared when <function>PQresultStatus</function> + returns PGRES_PIPELINE_SYNC at the end of the pipeline. + Clients don't usually need this function to verify aborted status, as they can tell that the pipeline is aborted from the <literal>PGRES_PIPELINE_ABORTED</literal> result code. </para> @@ -5328,7 +5329,7 @@ int PQsendPipeline(PGconn *conn); Returns 1 for success. Returns 0 if the connection is not in pipeline mode or sending a <link linkend="protocol-flow-ext-query">sync message</link> - is failed. + failed. </para> </listitem> </varlistentry> @@ -5349,7 +5350,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is most useful when the server is distant, i.e., network latency (<quote>ping time</quote>) is high, and also when many small operations - are being performed in rapid sequence. There is usually less benefit + are being performed in rapid succession. There is usually less benefit in using pipelined commands when each query takes many multiples of the client/server round-trip time to execute. A 100-statement operation run on a server 300ms round-trip-time away would take 30 seconds in network latency alone @@ -5367,7 +5368,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is not useful when information from one operation is required by the client to produce the next operation. In such cases, the client - must introduce a synchronization point and wait for a full client/server + would have to introduce a synchronization point and wait for a full client/server round-trip to get the results it needs. However, it's often possible to adjust the client design to exchange the required information server-side. Read-modify-write cycles are especially good candidates; for example: @@ -5392,10 +5393,10 @@ UPDATE mytable SET x = x + 1 WHERE id = 42; <note> <para> - The pipeline API was introduced in PostgreSQL 14, but clients using - the PostgreSQL 14 version of <application>libpq</application> can use - pipelines on server versions 7.4 and newer. Pipeline mode works on any server - that supports the v3 extended query protocol. + The pipeline API was introduced in <productname>PostgreSQL</productname> 14. + Pipeline mode is a client-side feature which doesn't require server + support, and works on any server that supports the v3 extended query + protocol. </para> </note> </sect2> diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c index 7ae7c14948..d7b036a35c 100644 --- a/src/interfaces/libpq/fe-exec.c +++ b/src/interfaces/libpq/fe-exec.c @@ -3803,7 +3803,7 @@ pqPipelineFlush(PGconn *conn) { if ((conn->pipelineStatus == PQ_PIPELINE_OFF) || (conn->outCount >= OUTBUFFER_THRESHOLD)) - return (pqFlush(conn)); + return pqFlush(conn); return 0; } diff --git a/src/test/modules/test_libpq/pipeline.c b/src/test/modules/test_libpq/pipeline.c index f4a2bdec57..01d5a9a8ff 100644 --- a/src/test/modules/test_libpq/pipeline.c +++ b/src/test/modules/test_libpq/pipeline.c @@ -126,7 +126,7 @@ test_simple_pipeline(PGconn *conn) * process the results of as they come in. * * For a simple case we should be able to do this without interim - * processing of results since our out buffer will give us enough slush to + * processing of results since our output buffer will give us enough slush to * work with and we won't block on sending. So blocking mode is fine. */ if (PQisnonblocking(conn)) @@ -603,7 +603,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) /* * Now we start inserting. We'll be sending enough data that we could fill - * our out buffer, so to avoid deadlocking we need to enter nonblocking + * our output buffer, so to avoid deadlocking we need to enter nonblocking * mode and consume input while we send more output. As results of each * query are processed we should pop them to allow processing of the next * query. There's no need to finish the pipeline before processing @@ -635,7 +635,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) } /* - * Process any results, so we keep the server's out buffer free + * Process any results, so we keep the server's output buffer free * flowing and it can continue to process input */ if (FD_ISSET(sock, &input_mask)) -- 2.17.0 --oC1+HKm2/end4ao3-- ^ permalink raw reply [nested|flat] 27+ messages in thread
* [PATCH 2/2] doc review @ 2021-03-03 22:36 Justin Pryzby <[email protected]> 0 siblings, 0 replies; 27+ messages in thread From: Justin Pryzby @ 2021-03-03 22:36 UTC (permalink / raw) --- doc/src/sgml/libpq.sgml | 67 +++++++++++++------------- src/interfaces/libpq/fe-exec.c | 2 +- src/test/modules/test_libpq/pipeline.c | 6 +-- 3 files changed, 38 insertions(+), 37 deletions(-) diff --git a/doc/src/sgml/libpq.sgml b/doc/src/sgml/libpq.sgml index a34ecbb144..e2865a218b 100644 --- a/doc/src/sgml/libpq.sgml +++ b/doc/src/sgml/libpq.sgml @@ -3130,8 +3130,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_SYNC</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a point in a - pipeline where a synchronization point has been established. + The <structname>PGresult</structname> represents a + synchronization point in pipeline mode, requested by + <xref linkend="libpq-PQsendPipeline"/>. This status occurs only when pipeline mode has been selected. </para> </listitem> @@ -3141,9 +3142,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_ABORTED</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a pipeline that's + The <structname>PGresult</structname> represents a pipeline that has received an error from the server. <function>PQgetResult</function> - must be called repeatedly, and it will return this status code, + must be called repeatedly, and each time it will return this status code until the end of the current pipeline, at which point it will return <literal>PGRES_PIPELINE_SYNC</literal> and normal processing can resume. @@ -4953,7 +4954,7 @@ int PQflush(PGconn *conn); <title>Using Pipeline Mode</title> <para> - To issue pipelines the application must switch a connection into pipeline mode. + To issue pipelines, the application must switch a connection into pipeline mode. Enter pipeline mode with <xref linkend="libpq-PQenterPipelineMode"/> or test whether pipeline mode is active with <xref linkend="libpq-PQpipelineStatus"/>. @@ -4975,7 +4976,7 @@ int PQflush(PGconn *conn); server will block trying to send results to the client from queries it has already processed. This only occurs when the client sends enough queries to fill both its output buffer and the server's receive - buffer before switching to processing input from the server, + buffer before it switches to processing input from the server, but it's hard to predict exactly when that will happen. </para> </footnote> @@ -5025,7 +5026,7 @@ int PQflush(PGconn *conn); <title>Processing Results</title> <para> - To process the result of one pipeline query, the application calls + To process the result of one query in a pipeline, the application calls <function>PQgetResult</function> repeatedly and handles each result until <function>PQgetResult</function> returns null. The result from the next query in the pipeline may then be retrieved using @@ -5057,10 +5058,10 @@ int PQflush(PGconn *conn); <type>PGresult</type> types <literal>PGRES_PIPELINE_SYNC</literal> and <literal>PGRES_PIPELINE_ABORTED</literal>. <literal>PGRES_PIPELINE_SYNC</literal> is reported exactly once for each - <function>PQsendPipeline</function> call at the corresponding point in - the result stream. + <function>PQsendPipeline</function> after retrieving results for all + queries in the pipeline. <literal>PGRES_PIPELINE_ABORTED</literal> is emitted in place of a normal - result stream result for the first error and all subsequent results + stream result for the first error and all subsequent results except <literal>PGRES_PIPELINE_SYNC</literal> and null; see <xref linkend="libpq-pipeline-errors"/>. </para> @@ -5075,7 +5076,8 @@ int PQflush(PGconn *conn); application about the query currently being processed (except that <function>PQgetResult</function> returns null to indicate that we start returning the results of next query). The application must keep track - of the order in which it sent queries and the expected results. + of the order in which it sent queries, to associate them with their + corresponding results. Applications will typically use a state machine or a FIFO queue for this. </para> @@ -5091,10 +5093,10 @@ int PQflush(PGconn *conn); </para> <para> - From the client perspective, after the client gets a - <literal>PGRES_FATAL_ERROR</literal> return from - <function>PQresultStatus</function> the pipeline is flagged as aborted. - <application>libpq</application> will report + From the client perspective, after <function>PQresultStatus</function> + returns <literal>PGRES_FATAL_ERROR</literal>, + the pipeline is flagged as aborted. + <function>PQresultStatus</function>, will report a <literal>PGRES_PIPELINE_ABORTED</literal> result for each remaining queued operation in an aborted pipeline. The result for <function>PQsendPipeline</function> is reported as @@ -5108,8 +5110,8 @@ int PQflush(PGconn *conn); </para> <para> - If the pipeline used an implicit transaction then operations that have - already executed are rolled back and operations that were queued for after + If the pipeline used an implicit transaction, then operations that have + already executed are rolled back and operations that were queued to follow the failed operation are skipped entirely. The same behaviour holds if the pipeline starts and commits a single explicit transaction (i.e. the first statement is <literal>BEGIN</literal> and the last is @@ -5145,11 +5147,11 @@ int PQflush(PGconn *conn); <para> The client application should generally maintain a queue of work - still to be dispatched and a queue of work that has been dispatched + remaining to be dispatched and a queue of work that has been dispatched but not yet had its results processed. When the socket is writable it should dispatch more work. When the socket is readable it should read results and process them, matching them up to the next entry in - its expected results queue. Based on available memory, results from + its expected results queue. Based on available memory, results from the socket should be read frequently: there's no need to wait until the pipeline end to read the results. Pipelines should be scoped to logical units of work, usually (but not necessarily) one transaction per pipeline. @@ -5191,8 +5193,8 @@ int PQflush(PGconn *conn); <listitem> <para> - Returns current pipeline mode status of the <application>libpq</application> - connection. + Returns the current pipeline mode status of the + <application>libpq</application> connection. <synopsis> PGpipelineStatus PQpipelineStatus(const PGconn *conn); </synopsis> @@ -5233,11 +5235,10 @@ PGpipelineStatus PQpipelineStatus(const PGconn *conn); <listitem> <para> The <application>libpq</application> connection is in pipeline - mode and an error has occurred while processing the current - pipeline. - The aborted flag is cleared as soon as the result - of the <function>PQsendPipeline</function> at the end of the aborted - pipeline is processed. Clients don't usually need this function to + mode and an error occurred while processing the current pipeline. + The aborted flag is cleared when <function>PQresultStatus</function> + returns PGRES_PIPELINE_SYNC at the end of the pipeline. + Clients don't usually need this function to verify aborted status, as they can tell that the pipeline is aborted from the <literal>PGRES_PIPELINE_ABORTED</literal> result code. </para> @@ -5328,7 +5329,7 @@ int PQsendPipeline(PGconn *conn); Returns 1 for success. Returns 0 if the connection is not in pipeline mode or sending a <link linkend="protocol-flow-ext-query">sync message</link> - is failed. + failed. </para> </listitem> </varlistentry> @@ -5349,7 +5350,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is most useful when the server is distant, i.e., network latency (<quote>ping time</quote>) is high, and also when many small operations - are being performed in rapid sequence. There is usually less benefit + are being performed in rapid succession. There is usually less benefit in using pipelined commands when each query takes many multiples of the client/server round-trip time to execute. A 100-statement operation run on a server 300ms round-trip-time away would take 30 seconds in network latency alone @@ -5367,7 +5368,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is not useful when information from one operation is required by the client to produce the next operation. In such cases, the client - must introduce a synchronization point and wait for a full client/server + would have to introduce a synchronization point and wait for a full client/server round-trip to get the results it needs. However, it's often possible to adjust the client design to exchange the required information server-side. Read-modify-write cycles are especially good candidates; for example: @@ -5392,10 +5393,10 @@ UPDATE mytable SET x = x + 1 WHERE id = 42; <note> <para> - The pipeline API was introduced in PostgreSQL 14, but clients using - the PostgreSQL 14 version of <application>libpq</application> can use - pipelines on server versions 7.4 and newer. Pipeline mode works on any server - that supports the v3 extended query protocol. + The pipeline API was introduced in <productname>PostgreSQL</productname> 14. + Pipeline mode is a client-side feature which doesn't require server + support, and works on any server that supports the v3 extended query + protocol. </para> </note> </sect2> diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c index 7ae7c14948..d7b036a35c 100644 --- a/src/interfaces/libpq/fe-exec.c +++ b/src/interfaces/libpq/fe-exec.c @@ -3803,7 +3803,7 @@ pqPipelineFlush(PGconn *conn) { if ((conn->pipelineStatus == PQ_PIPELINE_OFF) || (conn->outCount >= OUTBUFFER_THRESHOLD)) - return (pqFlush(conn)); + return pqFlush(conn); return 0; } diff --git a/src/test/modules/test_libpq/pipeline.c b/src/test/modules/test_libpq/pipeline.c index f4a2bdec57..01d5a9a8ff 100644 --- a/src/test/modules/test_libpq/pipeline.c +++ b/src/test/modules/test_libpq/pipeline.c @@ -126,7 +126,7 @@ test_simple_pipeline(PGconn *conn) * process the results of as they come in. * * For a simple case we should be able to do this without interim - * processing of results since our out buffer will give us enough slush to + * processing of results since our output buffer will give us enough slush to * work with and we won't block on sending. So blocking mode is fine. */ if (PQisnonblocking(conn)) @@ -603,7 +603,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) /* * Now we start inserting. We'll be sending enough data that we could fill - * our out buffer, so to avoid deadlocking we need to enter nonblocking + * our output buffer, so to avoid deadlocking we need to enter nonblocking * mode and consume input while we send more output. As results of each * query are processed we should pop them to allow processing of the next * query. There's no need to finish the pipeline before processing @@ -635,7 +635,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) } /* - * Process any results, so we keep the server's out buffer free + * Process any results, so we keep the server's output buffer free * flowing and it can continue to process input */ if (FD_ISSET(sock, &input_mask)) -- 2.17.0 --oC1+HKm2/end4ao3-- ^ permalink raw reply [nested|flat] 27+ messages in thread
* [PATCH 2/2] doc review @ 2021-03-03 22:36 Justin Pryzby <[email protected]> 0 siblings, 0 replies; 27+ messages in thread From: Justin Pryzby @ 2021-03-03 22:36 UTC (permalink / raw) --- doc/src/sgml/libpq.sgml | 67 +++++++++++++------------- src/interfaces/libpq/fe-exec.c | 2 +- src/test/modules/test_libpq/pipeline.c | 6 +-- 3 files changed, 38 insertions(+), 37 deletions(-) diff --git a/doc/src/sgml/libpq.sgml b/doc/src/sgml/libpq.sgml index a34ecbb144..e2865a218b 100644 --- a/doc/src/sgml/libpq.sgml +++ b/doc/src/sgml/libpq.sgml @@ -3130,8 +3130,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_SYNC</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a point in a - pipeline where a synchronization point has been established. + The <structname>PGresult</structname> represents a + synchronization point in pipeline mode, requested by + <xref linkend="libpq-PQsendPipeline"/>. This status occurs only when pipeline mode has been selected. </para> </listitem> @@ -3141,9 +3142,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_ABORTED</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a pipeline that's + The <structname>PGresult</structname> represents a pipeline that has received an error from the server. <function>PQgetResult</function> - must be called repeatedly, and it will return this status code, + must be called repeatedly, and each time it will return this status code until the end of the current pipeline, at which point it will return <literal>PGRES_PIPELINE_SYNC</literal> and normal processing can resume. @@ -4953,7 +4954,7 @@ int PQflush(PGconn *conn); <title>Using Pipeline Mode</title> <para> - To issue pipelines the application must switch a connection into pipeline mode. + To issue pipelines, the application must switch a connection into pipeline mode. Enter pipeline mode with <xref linkend="libpq-PQenterPipelineMode"/> or test whether pipeline mode is active with <xref linkend="libpq-PQpipelineStatus"/>. @@ -4975,7 +4976,7 @@ int PQflush(PGconn *conn); server will block trying to send results to the client from queries it has already processed. This only occurs when the client sends enough queries to fill both its output buffer and the server's receive - buffer before switching to processing input from the server, + buffer before it switches to processing input from the server, but it's hard to predict exactly when that will happen. </para> </footnote> @@ -5025,7 +5026,7 @@ int PQflush(PGconn *conn); <title>Processing Results</title> <para> - To process the result of one pipeline query, the application calls + To process the result of one query in a pipeline, the application calls <function>PQgetResult</function> repeatedly and handles each result until <function>PQgetResult</function> returns null. The result from the next query in the pipeline may then be retrieved using @@ -5057,10 +5058,10 @@ int PQflush(PGconn *conn); <type>PGresult</type> types <literal>PGRES_PIPELINE_SYNC</literal> and <literal>PGRES_PIPELINE_ABORTED</literal>. <literal>PGRES_PIPELINE_SYNC</literal> is reported exactly once for each - <function>PQsendPipeline</function> call at the corresponding point in - the result stream. + <function>PQsendPipeline</function> after retrieving results for all + queries in the pipeline. <literal>PGRES_PIPELINE_ABORTED</literal> is emitted in place of a normal - result stream result for the first error and all subsequent results + stream result for the first error and all subsequent results except <literal>PGRES_PIPELINE_SYNC</literal> and null; see <xref linkend="libpq-pipeline-errors"/>. </para> @@ -5075,7 +5076,8 @@ int PQflush(PGconn *conn); application about the query currently being processed (except that <function>PQgetResult</function> returns null to indicate that we start returning the results of next query). The application must keep track - of the order in which it sent queries and the expected results. + of the order in which it sent queries, to associate them with their + corresponding results. Applications will typically use a state machine or a FIFO queue for this. </para> @@ -5091,10 +5093,10 @@ int PQflush(PGconn *conn); </para> <para> - From the client perspective, after the client gets a - <literal>PGRES_FATAL_ERROR</literal> return from - <function>PQresultStatus</function> the pipeline is flagged as aborted. - <application>libpq</application> will report + From the client perspective, after <function>PQresultStatus</function> + returns <literal>PGRES_FATAL_ERROR</literal>, + the pipeline is flagged as aborted. + <function>PQresultStatus</function>, will report a <literal>PGRES_PIPELINE_ABORTED</literal> result for each remaining queued operation in an aborted pipeline. The result for <function>PQsendPipeline</function> is reported as @@ -5108,8 +5110,8 @@ int PQflush(PGconn *conn); </para> <para> - If the pipeline used an implicit transaction then operations that have - already executed are rolled back and operations that were queued for after + If the pipeline used an implicit transaction, then operations that have + already executed are rolled back and operations that were queued to follow the failed operation are skipped entirely. The same behaviour holds if the pipeline starts and commits a single explicit transaction (i.e. the first statement is <literal>BEGIN</literal> and the last is @@ -5145,11 +5147,11 @@ int PQflush(PGconn *conn); <para> The client application should generally maintain a queue of work - still to be dispatched and a queue of work that has been dispatched + remaining to be dispatched and a queue of work that has been dispatched but not yet had its results processed. When the socket is writable it should dispatch more work. When the socket is readable it should read results and process them, matching them up to the next entry in - its expected results queue. Based on available memory, results from + its expected results queue. Based on available memory, results from the socket should be read frequently: there's no need to wait until the pipeline end to read the results. Pipelines should be scoped to logical units of work, usually (but not necessarily) one transaction per pipeline. @@ -5191,8 +5193,8 @@ int PQflush(PGconn *conn); <listitem> <para> - Returns current pipeline mode status of the <application>libpq</application> - connection. + Returns the current pipeline mode status of the + <application>libpq</application> connection. <synopsis> PGpipelineStatus PQpipelineStatus(const PGconn *conn); </synopsis> @@ -5233,11 +5235,10 @@ PGpipelineStatus PQpipelineStatus(const PGconn *conn); <listitem> <para> The <application>libpq</application> connection is in pipeline - mode and an error has occurred while processing the current - pipeline. - The aborted flag is cleared as soon as the result - of the <function>PQsendPipeline</function> at the end of the aborted - pipeline is processed. Clients don't usually need this function to + mode and an error occurred while processing the current pipeline. + The aborted flag is cleared when <function>PQresultStatus</function> + returns PGRES_PIPELINE_SYNC at the end of the pipeline. + Clients don't usually need this function to verify aborted status, as they can tell that the pipeline is aborted from the <literal>PGRES_PIPELINE_ABORTED</literal> result code. </para> @@ -5328,7 +5329,7 @@ int PQsendPipeline(PGconn *conn); Returns 1 for success. Returns 0 if the connection is not in pipeline mode or sending a <link linkend="protocol-flow-ext-query">sync message</link> - is failed. + failed. </para> </listitem> </varlistentry> @@ -5349,7 +5350,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is most useful when the server is distant, i.e., network latency (<quote>ping time</quote>) is high, and also when many small operations - are being performed in rapid sequence. There is usually less benefit + are being performed in rapid succession. There is usually less benefit in using pipelined commands when each query takes many multiples of the client/server round-trip time to execute. A 100-statement operation run on a server 300ms round-trip-time away would take 30 seconds in network latency alone @@ -5367,7 +5368,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is not useful when information from one operation is required by the client to produce the next operation. In such cases, the client - must introduce a synchronization point and wait for a full client/server + would have to introduce a synchronization point and wait for a full client/server round-trip to get the results it needs. However, it's often possible to adjust the client design to exchange the required information server-side. Read-modify-write cycles are especially good candidates; for example: @@ -5392,10 +5393,10 @@ UPDATE mytable SET x = x + 1 WHERE id = 42; <note> <para> - The pipeline API was introduced in PostgreSQL 14, but clients using - the PostgreSQL 14 version of <application>libpq</application> can use - pipelines on server versions 7.4 and newer. Pipeline mode works on any server - that supports the v3 extended query protocol. + The pipeline API was introduced in <productname>PostgreSQL</productname> 14. + Pipeline mode is a client-side feature which doesn't require server + support, and works on any server that supports the v3 extended query + protocol. </para> </note> </sect2> diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c index 7ae7c14948..d7b036a35c 100644 --- a/src/interfaces/libpq/fe-exec.c +++ b/src/interfaces/libpq/fe-exec.c @@ -3803,7 +3803,7 @@ pqPipelineFlush(PGconn *conn) { if ((conn->pipelineStatus == PQ_PIPELINE_OFF) || (conn->outCount >= OUTBUFFER_THRESHOLD)) - return (pqFlush(conn)); + return pqFlush(conn); return 0; } diff --git a/src/test/modules/test_libpq/pipeline.c b/src/test/modules/test_libpq/pipeline.c index f4a2bdec57..01d5a9a8ff 100644 --- a/src/test/modules/test_libpq/pipeline.c +++ b/src/test/modules/test_libpq/pipeline.c @@ -126,7 +126,7 @@ test_simple_pipeline(PGconn *conn) * process the results of as they come in. * * For a simple case we should be able to do this without interim - * processing of results since our out buffer will give us enough slush to + * processing of results since our output buffer will give us enough slush to * work with and we won't block on sending. So blocking mode is fine. */ if (PQisnonblocking(conn)) @@ -603,7 +603,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) /* * Now we start inserting. We'll be sending enough data that we could fill - * our out buffer, so to avoid deadlocking we need to enter nonblocking + * our output buffer, so to avoid deadlocking we need to enter nonblocking * mode and consume input while we send more output. As results of each * query are processed we should pop them to allow processing of the next * query. There's no need to finish the pipeline before processing @@ -635,7 +635,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) } /* - * Process any results, so we keep the server's out buffer free + * Process any results, so we keep the server's output buffer free * flowing and it can continue to process input */ if (FD_ISSET(sock, &input_mask)) -- 2.17.0 --oC1+HKm2/end4ao3-- ^ permalink raw reply [nested|flat] 27+ messages in thread
* [PATCH 2/2] doc review @ 2021-03-03 22:36 Justin Pryzby <[email protected]> 0 siblings, 0 replies; 27+ messages in thread From: Justin Pryzby @ 2021-03-03 22:36 UTC (permalink / raw) --- doc/src/sgml/libpq.sgml | 67 +++++++++++++------------- src/interfaces/libpq/fe-exec.c | 2 +- src/test/modules/test_libpq/pipeline.c | 6 +-- 3 files changed, 38 insertions(+), 37 deletions(-) diff --git a/doc/src/sgml/libpq.sgml b/doc/src/sgml/libpq.sgml index a34ecbb144..e2865a218b 100644 --- a/doc/src/sgml/libpq.sgml +++ b/doc/src/sgml/libpq.sgml @@ -3130,8 +3130,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_SYNC</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a point in a - pipeline where a synchronization point has been established. + The <structname>PGresult</structname> represents a + synchronization point in pipeline mode, requested by + <xref linkend="libpq-PQsendPipeline"/>. This status occurs only when pipeline mode has been selected. </para> </listitem> @@ -3141,9 +3142,9 @@ ExecStatusType PQresultStatus(const PGresult *res); <term><literal>PGRES_PIPELINE_ABORTED</literal></term> <listitem> <para> - The <structname>PGresult</structname> represents a pipeline that's + The <structname>PGresult</structname> represents a pipeline that has received an error from the server. <function>PQgetResult</function> - must be called repeatedly, and it will return this status code, + must be called repeatedly, and each time it will return this status code until the end of the current pipeline, at which point it will return <literal>PGRES_PIPELINE_SYNC</literal> and normal processing can resume. @@ -4953,7 +4954,7 @@ int PQflush(PGconn *conn); <title>Using Pipeline Mode</title> <para> - To issue pipelines the application must switch a connection into pipeline mode. + To issue pipelines, the application must switch a connection into pipeline mode. Enter pipeline mode with <xref linkend="libpq-PQenterPipelineMode"/> or test whether pipeline mode is active with <xref linkend="libpq-PQpipelineStatus"/>. @@ -4975,7 +4976,7 @@ int PQflush(PGconn *conn); server will block trying to send results to the client from queries it has already processed. This only occurs when the client sends enough queries to fill both its output buffer and the server's receive - buffer before switching to processing input from the server, + buffer before it switches to processing input from the server, but it's hard to predict exactly when that will happen. </para> </footnote> @@ -5025,7 +5026,7 @@ int PQflush(PGconn *conn); <title>Processing Results</title> <para> - To process the result of one pipeline query, the application calls + To process the result of one query in a pipeline, the application calls <function>PQgetResult</function> repeatedly and handles each result until <function>PQgetResult</function> returns null. The result from the next query in the pipeline may then be retrieved using @@ -5057,10 +5058,10 @@ int PQflush(PGconn *conn); <type>PGresult</type> types <literal>PGRES_PIPELINE_SYNC</literal> and <literal>PGRES_PIPELINE_ABORTED</literal>. <literal>PGRES_PIPELINE_SYNC</literal> is reported exactly once for each - <function>PQsendPipeline</function> call at the corresponding point in - the result stream. + <function>PQsendPipeline</function> after retrieving results for all + queries in the pipeline. <literal>PGRES_PIPELINE_ABORTED</literal> is emitted in place of a normal - result stream result for the first error and all subsequent results + stream result for the first error and all subsequent results except <literal>PGRES_PIPELINE_SYNC</literal> and null; see <xref linkend="libpq-pipeline-errors"/>. </para> @@ -5075,7 +5076,8 @@ int PQflush(PGconn *conn); application about the query currently being processed (except that <function>PQgetResult</function> returns null to indicate that we start returning the results of next query). The application must keep track - of the order in which it sent queries and the expected results. + of the order in which it sent queries, to associate them with their + corresponding results. Applications will typically use a state machine or a FIFO queue for this. </para> @@ -5091,10 +5093,10 @@ int PQflush(PGconn *conn); </para> <para> - From the client perspective, after the client gets a - <literal>PGRES_FATAL_ERROR</literal> return from - <function>PQresultStatus</function> the pipeline is flagged as aborted. - <application>libpq</application> will report + From the client perspective, after <function>PQresultStatus</function> + returns <literal>PGRES_FATAL_ERROR</literal>, + the pipeline is flagged as aborted. + <function>PQresultStatus</function>, will report a <literal>PGRES_PIPELINE_ABORTED</literal> result for each remaining queued operation in an aborted pipeline. The result for <function>PQsendPipeline</function> is reported as @@ -5108,8 +5110,8 @@ int PQflush(PGconn *conn); </para> <para> - If the pipeline used an implicit transaction then operations that have - already executed are rolled back and operations that were queued for after + If the pipeline used an implicit transaction, then operations that have + already executed are rolled back and operations that were queued to follow the failed operation are skipped entirely. The same behaviour holds if the pipeline starts and commits a single explicit transaction (i.e. the first statement is <literal>BEGIN</literal> and the last is @@ -5145,11 +5147,11 @@ int PQflush(PGconn *conn); <para> The client application should generally maintain a queue of work - still to be dispatched and a queue of work that has been dispatched + remaining to be dispatched and a queue of work that has been dispatched but not yet had its results processed. When the socket is writable it should dispatch more work. When the socket is readable it should read results and process them, matching them up to the next entry in - its expected results queue. Based on available memory, results from + its expected results queue. Based on available memory, results from the socket should be read frequently: there's no need to wait until the pipeline end to read the results. Pipelines should be scoped to logical units of work, usually (but not necessarily) one transaction per pipeline. @@ -5191,8 +5193,8 @@ int PQflush(PGconn *conn); <listitem> <para> - Returns current pipeline mode status of the <application>libpq</application> - connection. + Returns the current pipeline mode status of the + <application>libpq</application> connection. <synopsis> PGpipelineStatus PQpipelineStatus(const PGconn *conn); </synopsis> @@ -5233,11 +5235,10 @@ PGpipelineStatus PQpipelineStatus(const PGconn *conn); <listitem> <para> The <application>libpq</application> connection is in pipeline - mode and an error has occurred while processing the current - pipeline. - The aborted flag is cleared as soon as the result - of the <function>PQsendPipeline</function> at the end of the aborted - pipeline is processed. Clients don't usually need this function to + mode and an error occurred while processing the current pipeline. + The aborted flag is cleared when <function>PQresultStatus</function> + returns PGRES_PIPELINE_SYNC at the end of the pipeline. + Clients don't usually need this function to verify aborted status, as they can tell that the pipeline is aborted from the <literal>PGRES_PIPELINE_ABORTED</literal> result code. </para> @@ -5328,7 +5329,7 @@ int PQsendPipeline(PGconn *conn); Returns 1 for success. Returns 0 if the connection is not in pipeline mode or sending a <link linkend="protocol-flow-ext-query">sync message</link> - is failed. + failed. </para> </listitem> </varlistentry> @@ -5349,7 +5350,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is most useful when the server is distant, i.e., network latency (<quote>ping time</quote>) is high, and also when many small operations - are being performed in rapid sequence. There is usually less benefit + are being performed in rapid succession. There is usually less benefit in using pipelined commands when each query takes many multiples of the client/server round-trip time to execute. A 100-statement operation run on a server 300ms round-trip-time away would take 30 seconds in network latency alone @@ -5367,7 +5368,7 @@ int PQsendPipeline(PGconn *conn); <para> Pipeline mode is not useful when information from one operation is required by the client to produce the next operation. In such cases, the client - must introduce a synchronization point and wait for a full client/server + would have to introduce a synchronization point and wait for a full client/server round-trip to get the results it needs. However, it's often possible to adjust the client design to exchange the required information server-side. Read-modify-write cycles are especially good candidates; for example: @@ -5392,10 +5393,10 @@ UPDATE mytable SET x = x + 1 WHERE id = 42; <note> <para> - The pipeline API was introduced in PostgreSQL 14, but clients using - the PostgreSQL 14 version of <application>libpq</application> can use - pipelines on server versions 7.4 and newer. Pipeline mode works on any server - that supports the v3 extended query protocol. + The pipeline API was introduced in <productname>PostgreSQL</productname> 14. + Pipeline mode is a client-side feature which doesn't require server + support, and works on any server that supports the v3 extended query + protocol. </para> </note> </sect2> diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c index 7ae7c14948..d7b036a35c 100644 --- a/src/interfaces/libpq/fe-exec.c +++ b/src/interfaces/libpq/fe-exec.c @@ -3803,7 +3803,7 @@ pqPipelineFlush(PGconn *conn) { if ((conn->pipelineStatus == PQ_PIPELINE_OFF) || (conn->outCount >= OUTBUFFER_THRESHOLD)) - return (pqFlush(conn)); + return pqFlush(conn); return 0; } diff --git a/src/test/modules/test_libpq/pipeline.c b/src/test/modules/test_libpq/pipeline.c index f4a2bdec57..01d5a9a8ff 100644 --- a/src/test/modules/test_libpq/pipeline.c +++ b/src/test/modules/test_libpq/pipeline.c @@ -126,7 +126,7 @@ test_simple_pipeline(PGconn *conn) * process the results of as they come in. * * For a simple case we should be able to do this without interim - * processing of results since our out buffer will give us enough slush to + * processing of results since our output buffer will give us enough slush to * work with and we won't block on sending. So blocking mode is fine. */ if (PQisnonblocking(conn)) @@ -603,7 +603,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) /* * Now we start inserting. We'll be sending enough data that we could fill - * our out buffer, so to avoid deadlocking we need to enter nonblocking + * our output buffer, so to avoid deadlocking we need to enter nonblocking * mode and consume input while we send more output. As results of each * query are processed we should pop them to allow processing of the next * query. There's no need to finish the pipeline before processing @@ -635,7 +635,7 @@ test_pipelined_insert(PGconn *conn, int n_rows) } /* - * Process any results, so we keep the server's out buffer free + * Process any results, so we keep the server's output buffer free * flowing and it can continue to process input */ if (FD_ISSET(sock, &input_mask)) -- 2.17.0 --oC1+HKm2/end4ao3-- ^ permalink raw reply [nested|flat] 27+ messages in thread
* Re: generating function default settings from pg_proc.dat @ 2026-02-17 17:04 Corey Huinker <[email protected]> 0 siblings, 1 reply; 27+ messages in thread From: Corey Huinker @ 2026-02-17 17:04 UTC (permalink / raw) To: Tom Lane <[email protected]>; +Cc: Álvaro Herrera <[email protected]>; Andres Freund <[email protected]>; Andrew Dunstan <[email protected]>; pgsql-hackers On Tue, Feb 17, 2026 at 10:00 AM Tom Lane <[email protected]> wrote: > Corey Huinker <[email protected]> writes: > > I like this a lot too, but I'm noticing that with each iteration we're > > getting closer to re-inventing SQL. > > Really? Neither pg_proc.dat nor the constructed postgres.bki file > look anything like SQL to my eye. > It doesn't _look_ like SQL, but we're trying to cover all things that a CREATE FUNCTION can do, and if we don't model them all, then we're back to needing CREATE OR REPLACE overlays. > > Would it make sense in the long run to > > have a mode on the CREATE FUNCTION command that cues initdb to create the > > minimal function skeleton with prescribed oid on the first pass, but then > > stores the defer-able parts (if any) for a later pass, perhaps in > parallel? > > I seriously, seriously doubt it. That would involve allowing large > amounts of the parser to run in bootstrap mode, and would probably > end in plastering warts all over backend/parser/ to say "do this > in one way normally but some other way in bootstrap". Also, it's > really just syntactic sugar and does nothing for the harder problems > that bootstrap mode has to solve, such as supporting references to > objects that've not been created yet. > Fair enough. But in the interest of keeping a single source of truth, what if we reversed the process, having a build-time perl script parse system_functions.sql to build a minimal pg_proc.dat-like file just for bootstrap? We would probably want to break up system_functions.sql into several files, (admin functions go here, adt-related functions go there, etc), but we'd have total clarity on all function definitions, and we wouldn't have to modify the generation of pg_proc.dat unless a new function syntax feature affected bootstrapping. ^ permalink raw reply [nested|flat] 27+ messages in thread
* Re: generating function default settings from pg_proc.dat @ 2026-02-18 19:33 Tom Lane <[email protected]> parent: Corey Huinker <[email protected]> 0 siblings, 0 replies; 27+ messages in thread From: Tom Lane @ 2026-02-18 19:33 UTC (permalink / raw) To: Corey Huinker <[email protected]>; +Cc: Álvaro Herrera <[email protected]>; Andres Freund <[email protected]>; Andrew Dunstan <[email protected]>; pgsql-hackers Corey Huinker <[email protected]> writes: > Fair enough. But in the interest of keeping a single source of truth, what > if we reversed the process, having a build-time perl script parse > system_functions.sql to build a minimal pg_proc.dat-like file just for > bootstrap? We would probably want to break up system_functions.sql into > several files, (admin functions go here, adt-related functions go there, > etc), but we'd have total clarity on all function definitions, and we > wouldn't have to modify the generation of pg_proc.dat unless a new function > syntax feature affected bootstrapping. Seems like an awful lot of work compared to the benefit. We are down to two kinds of entries in system_functions.sql: 1. New-style SQL-language functions. To my mind, for those functions system_functions.sql *is* the source of truth; the pg_proc entry is vestigial and present mostly to reserve the OID. If we decided it's okay to not have nailed-down OIDs for those functions, we could drop their pg_proc entries altogether. But even as things stand, those entries all say prosrc => 'see system_functions.sql' }, which I think makes it pretty clear where the primary definition is. 2. GRANT/REVOKE operations. Andres muttered something about trying to integrate those into pg_proc.dat too, and I'm going to go look at the idea. regards, tom lane ^ permalink raw reply [nested|flat] 27+ messages in thread
end of thread, other threads:[~2026-02-18 19:33 UTC | newest] Thread overview: 27+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2017-05-05 16:14 [PATCH] Remove the NODROP SLOT option from DROP SUBSCRIPTION Petr Jelinek <[email protected]> 2021-03-03 22:36 [PATCH 2/2] doc review Justin Pryzby <[email protected]> 2021-03-03 22:36 [PATCH 2/2] doc review Justin Pryzby <[email protected]> 2021-03-03 22:36 [PATCH 2/2] doc review Justin Pryzby <[email protected]> 2021-03-03 22:36 [PATCH 2/2] doc review Justin Pryzby <[email protected]> 2021-03-03 22:36 [PATCH 2/2] doc review Justin Pryzby <[email protected]> 2021-03-03 22:36 [PATCH 2/2] doc review Justin Pryzby <[email protected]> 2021-03-03 22:36 [PATCH 2/2] doc review Justin Pryzby <[email protected]> 2021-03-03 22:36 [PATCH 2/2] doc review Justin Pryzby <[email protected]> 2021-03-03 22:36 [PATCH 2/2] doc review Justin Pryzby <[email protected]> 2021-03-03 22:36 [PATCH 2/2] doc review Justin Pryzby <[email protected]> 2021-03-03 22:36 [PATCH 2/2] doc review Justin Pryzby <[email protected]> 2021-03-03 22:36 [PATCH 2/2] doc review Justin Pryzby <[email protected]> 2021-03-03 22:36 [PATCH 2/2] doc review Justin Pryzby <[email protected]> 2021-03-03 22:36 [PATCH 2/2] doc review Justin Pryzby <[email protected]> 2021-03-03 22:36 [PATCH 2/2] doc review Justin Pryzby <[email protected]> 2021-03-03 22:36 [PATCH 2/2] doc review Justin Pryzby <[email protected]> 2021-03-03 22:36 [PATCH 2/2] doc review Justin Pryzby <[email protected]> 2021-03-03 22:36 [PATCH 2/2] doc review Justin Pryzby <[email protected]> 2021-03-03 22:36 [PATCH 2/2] doc review Justin Pryzby <[email protected]> 2021-03-03 22:36 [PATCH 2/2] doc review Justin Pryzby <[email protected]> 2021-03-03 22:36 [PATCH 2/2] doc review Justin Pryzby <[email protected]> 2021-03-03 22:36 [PATCH 2/2] doc review Justin Pryzby <[email protected]> 2021-03-03 22:36 [PATCH 2/2] doc review Justin Pryzby <[email protected]> 2021-03-03 22:36 [PATCH 2/2] doc review Justin Pryzby <[email protected]> 2026-02-17 17:04 Re: generating function default settings from pg_proc.dat Corey Huinker <[email protected]> 2026-02-18 19:33 ` Re: generating function default settings from pg_proc.dat Tom Lane <[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