Received: from malur.postgresql.org ([217.196.149.56]) by arkaria.postgresql.org with esmtps (TLS1.3) tls TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384 (Exim 4.94.2) (envelope-from ) id 1rPHjp-00AVZt-3U for pgsql-hackers@arkaria.postgresql.org; Mon, 15 Jan 2024 07:50:30 +0000 Received: from localhost ([127.0.0.1] helo=malur.postgresql.org) by malur.postgresql.org with esmtp (Exim 4.94.2) (envelope-from ) id 1rPHjo-00E24l-3B for pgsql-hackers@arkaria.postgresql.org; Mon, 15 Jan 2024 07:50:28 +0000 Received: from makus.postgresql.org ([2001:4800:3e1:1::229]) by malur.postgresql.org with esmtps (TLS1.3) tls TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384 (Exim 4.94.2) (envelope-from ) id 1rPHjm-00E24b-5h for pgsql-hackers@lists.postgresql.org; Mon, 15 Jan 2024 07:50:27 +0000 Received: from out2-smtp.messagingengine.com ([66.111.4.26]) by makus.postgresql.org with esmtps (TLS1.3) tls TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384 (Exim 4.94.2) (envelope-from ) id 1rPHji-001cFW-FB for pgsql-hackers@postgresql.org; Mon, 15 Jan 2024 07:50:24 +0000 Received: from compute3.internal (compute3.nyi.internal [10.202.2.43]) by mailout.nyi.internal (Postfix) with ESMTP id DE8FC5C00F7; Mon, 15 Jan 2024 02:50:20 -0500 (EST) Received: from mailfrontend2 ([10.202.2.163]) by compute3.internal (MEProxy); Mon, 15 Jan 2024 02:50:20 -0500 DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=paquier.xyz; h= cc:cc:content-type:content-type:date:date:from:from:in-reply-to :in-reply-to:message-id:mime-version:references:reply-to:subject :subject:to:to; s=fm3; t=1705305020; x=1705391420; bh=dWy/C1NiyR naLFFclwYxD8CfSHkvIjQMcmo1jvrPLCs=; b=VHnXMpwFmAuVRXIs32QkIPvZWv 8k5WZkG8oB0nZubxTWm6uHRWjKqlDjrx9yRmFxrFTGxrwaSRnK1IWdV7nNarRHlk BeNu1jJ33bysPLnRlq7yWuogL90/DlPzSmHmKd5Ek2p2VpMwoHr368ECeyEjkbsy A8tpMh8tlBDaLSUEggcLCdFTW5svJiieBrqIBMOoGt48Xd8T+rnTvoObh3EkK6ru VQ9aNI+aNjswqOsm0h2kz/iXiwUpAZ4rOOVmyi5WW2iNhe7DiLMRR/gpi3iYvkf+ 5uw46HgEzrpFGCyC16ERKicoX2keih9nDVSN+F2cAkr1632SiGfACUADCs/A== DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d= messagingengine.com; h=cc:cc:content-type:content-type:date:date :feedback-id:feedback-id:from:from:in-reply-to:in-reply-to :message-id:mime-version:references:reply-to:subject:subject:to :to:x-me-proxy:x-me-proxy:x-me-sender:x-me-sender:x-sasl-enc; s= fm3; t=1705305020; x=1705391420; bh=dWy/C1NiyRnaLFFclwYxD8CfSHkv IjQMcmo1jvrPLCs=; b=JVywCAsj/pEhuAm2saFwQQYv7Q6RggfYFMt6VYbgEBOj ahKm7ByqHsu+ELq/UK2Kbthr0Pi89Mr3EtZxUfQnJKwjN3sgHGi/OZzpTkMWQjZ1 4pL0NYj6Nz1Ch3GpAbOETW+2tsym35IFGskfteRK/jirWNOpqtRHnMU7V3I+sPdh sUv6HpSLvUn2q5SC/VQABL+rd7YPDT4mHW621ug8Vl+q93ScU9rQnVbri2LK+8+b ImWwuilFQ84NOgFOHIYcLew+iXoK/lA6FrBF0FCYpz2oU+h7AoNqQiverhn31w7L vrTd1ye2kILZWmX3Qcirb5l50kCGrVsvSntwdRvNOA== X-ME-Sender: X-ME-Received: X-ME-Proxy-Cause: gggruggvucftvghtrhhoucdtuddrgedvkedrvdejtddgudduvdcutefuodetggdotefrod ftvfcurfhrohhfihhlvgemucfhrghsthforghilhdpqfgfvfdpuffrtefokffrpgfnqfgh necuuegrihhlohhuthemuceftddtnecusecvtfgvtghiphhivghnthhsucdlqddutddtmd enfghrlhcuvffnffculdejtddmnecujfgurhepfffhvfevuffkfhggtggujgesghdtreer tddtvdenucfhrhhomhepofhitghhrggvlhcurfgrqhhuihgvrhcuoehmihgthhgrvghlse hprghquhhivghrrdighiiiqeenucggtffrrghtthgvrhhnpeetleeifedufffhhfdtteel geeggeffhfekueevteeigfduudevudetgfegiedvjeenucevlhhushhtvghrufhiiigvpe dtnecurfgrrhgrmhepmhgrihhlfhhrohhmpehmihgthhgrvghlsehprghquhhivghrrdig hiii X-ME-Proxy: Feedback-ID: i0fe9450f:Fastmail Received: by mail.messagingengine.com (Postfix) with ESMTPA; Mon, 15 Jan 2024 02:50:18 -0500 (EST) Date: Mon, 15 Jan 2024 16:50:07 +0900 From: Michael Paquier To: Jelte Fennema-Nio Cc: Anthonin Bonnefoy , Anton Kirilov , Alvaro Herrera , Robert Haas , pgsql-hackers@postgresql.org Subject: Re: Add PQsendSyncMessage() to libpq Message-ID: References: <202311081620.s2a7ao5jtlpe@alvherre.pgsql> MIME-Version: 1.0 Content-Type: multipart/signed; micalg=pgp-sha512; protocol="application/pgp-signature"; boundary="epws8oteElA/2YUs" Content-Disposition: inline In-Reply-To: List-Id: List-Help: List-Subscribe: List-Post: List-Owner: List-Archive: Archived-At: Precedence: bulk --epws8oteElA/2YUs Content-Type: multipart/mixed; boundary="mv3uqqS97jfWEvjE" Content-Disposition: inline --mv3uqqS97jfWEvjE Content-Type: text/plain; charset=us-ascii Content-Disposition: inline On Wed, Jan 10, 2024 at 03:40:36PM +0900, Michael Paquier wrote: > Hence, as a whole, wouldn't it be more consistent if the new > PQsendPipelineSync() and the existing PQpipelineSync() call an > internal static routine (PQPipelineSyncInternal?) that can switch > between both modes? Let's just make the extra argument a boolean. Yeah, I'll go with that after a second look. Attached is what I am finishing with, and I have reproduced some numbers with the pgbench metacommand mentioned upthread, which is reeeaaally nice. I have also made a few edits to the tests. -- Michael --mv3uqqS97jfWEvjE Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v6-0001-Add-PQsendPipelineSync-to-libpq.patch" Content-Transfer-Encoding: quoted-printable =46rom b653759f8dfdfe83d0d8bc1d4a0ac9d4e272a061 Mon Sep 17 00:00:00 2001 =46rom: Michael Paquier Date: Mon, 15 Jan 2024 16:15:00 +0900 Subject: [PATCH v6] Add PQsendPipelineSync() to libpq This new function is equivalent to PQpipelineSync(), except that it does not flush anything to the server; the user must subsequently call PQflush() instead. Its purpose is to reduce the system call overhead of pipeline mode. --- src/interfaces/libpq/exports.txt | 1 + src/interfaces/libpq/fe-exec.c | 41 +++++++++++++++-- src/interfaces/libpq/libpq-fe.h | 1 + .../modules/libpq_pipeline/libpq_pipeline.c | 43 ++++++++++++++++++ .../traces/multi_pipelines.trace | 11 +++++ doc/src/sgml/libpq.sgml | 45 ++++++++++++++++--- 6 files changed, 131 insertions(+), 11 deletions(-) diff --git a/src/interfaces/libpq/exports.txt b/src/interfaces/libpq/export= s.txt index 28b861fd93..088592deb1 100644 --- a/src/interfaces/libpq/exports.txt +++ b/src/interfaces/libpq/exports.txt @@ -192,3 +192,4 @@ PQclosePortal 189 PQsendClosePrepared 190 PQsendClosePortal 191 PQchangePassword 192 +PQsendPipelineSync 193 diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c index 106d14e6ee..9e7e670921 100644 --- a/src/interfaces/libpq/fe-exec.c +++ b/src/interfaces/libpq/fe-exec.c @@ -81,6 +81,7 @@ static int PQsendTypedCommand(PGconn *conn, char command,= char type, const char *target); static int check_field_number(const PGresult *res, int field_num); static void pqPipelineProcessQueue(PGconn *conn); +static int pqPipelineSyncInternal(PGconn *conn, bool immediate_flush); static int pqPipelineFlush(PGconn *conn); =20 =20 @@ -3224,6 +3225,25 @@ pqPipelineProcessQueue(PGconn *conn) /* * PQpipelineSync * Send a Sync message as part of a pipeline, and flush to server + */ +int +PQpipelineSync(PGconn *conn) +{ + return pqPipelineSyncInternal(conn, true); +} + +/* + * PQsendPipelineSync + * Send a Sync message as part of a pipeline, without flushing to server + */ +int +PQsendPipelineSync(PGconn *conn) +{ + return pqPipelineSyncInternal(conn, false); +} + +/* + * Wrapper for PQpipelineSync and PQsendPipelineSync. * * It's legal to start submitting more commands in the pipeline immediatel= y, * without waiting for the results of the current pipeline. There's no nee= d to @@ -3240,9 +3260,12 @@ pqPipelineProcessQueue(PGconn *conn) * The connection will remain in pipeline mode and unavailable for new * synchronous command execution functions until all results from the pipe= line * are processed by the client. + * + * immediate_flush controls if the flush happens immediately after sending= the + * Sync message or not. */ -int -PQpipelineSync(PGconn *conn) +static int +pqPipelineSyncInternal(PGconn *conn, bool immediate_flush) { PGcmdQueueEntry *entry; =20 @@ -3288,9 +3311,19 @@ PQpipelineSync(PGconn *conn) /* * Give the data a push. In nonblock mode, don't complain if we're unable * to send it all; PQgetResult() will do any additional flushing needed. + * If immediate_flush is disabled, the data is pushed if we are past the + * size threshold. */ - if (PQflush(conn) < 0) - goto sendFailed; + if (immediate_flush) + { + if (pqFlush(conn) < 0) + goto sendFailed; + } + else + { + if (pqPipelineFlush(conn) < 0) + goto sendFailed; + } =20 /* OK, it's launched! */ pqAppendCmdQueueEntry(conn, entry); diff --git a/src/interfaces/libpq/libpq-fe.h b/src/interfaces/libpq/libpq-f= e.h index f0ec660cb6..defc415fa3 100644 --- a/src/interfaces/libpq/libpq-fe.h +++ b/src/interfaces/libpq/libpq-fe.h @@ -474,6 +474,7 @@ extern int PQenterPipelineMode(PGconn *conn); extern int PQexitPipelineMode(PGconn *conn); extern int PQpipelineSync(PGconn *conn); extern int PQsendFlushRequest(PGconn *conn); +extern int PQsendPipelineSync(PGconn *conn); =20 /* LISTEN/NOTIFY support */ extern PGnotify *PQnotifies(PGconn *conn); diff --git a/src/test/modules/libpq_pipeline/libpq_pipeline.c b/src/test/mo= dules/libpq_pipeline/libpq_pipeline.c index 71cd04c5f2..c68e20d0b5 100644 --- a/src/test/modules/libpq_pipeline/libpq_pipeline.c +++ b/src/test/modules/libpq_pipeline/libpq_pipeline.c @@ -162,6 +162,7 @@ test_multi_pipelines(PGconn *conn) if (PQenterPipelineMode(conn) !=3D 1) pg_fatal("failed to enter pipeline mode: %s", PQerrorMessage(conn)); =20 + /* first pipeline */ if (PQsendQueryParams(conn, "SELECT $1", 1, dummy_param_oids, dummy_params, NULL, NULL, 0) !=3D 1) pg_fatal("dispatching first SELECT failed: %s", PQerrorMessage(conn)); @@ -169,6 +170,16 @@ test_multi_pipelines(PGconn *conn) if (PQpipelineSync(conn) !=3D 1) pg_fatal("Pipeline sync failed: %s", PQerrorMessage(conn)); =20 + /* second pipeline */ + if (PQsendQueryParams(conn, "SELECT $1", 1, dummy_param_oids, + dummy_params, NULL, NULL, 0) !=3D 1) + pg_fatal("dispatching first SELECT failed: %s", PQerrorMessage(conn)); + + /* Skip flushing once. */ + if (PQsendPipelineSync(conn) !=3D 1) + pg_fatal("Pipeline sync failed: %s", PQerrorMessage(conn)); + + /* third pipeline */ if (PQsendQueryParams(conn, "SELECT $1", 1, dummy_param_oids, dummy_params, NULL, NULL, 0) !=3D 1) pg_fatal("dispatching second SELECT failed: %s", PQerrorMessage(conn)); @@ -177,6 +188,9 @@ test_multi_pipelines(PGconn *conn) pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn)); =20 /* OK, start processing the results */ + + /* first pipeline */ + res =3D PQgetResult(conn); if (res =3D=3D NULL) pg_fatal("PQgetResult returned null when there's a pipeline item: %s", @@ -206,6 +220,35 @@ test_multi_pipelines(PGconn *conn) =20 /* second pipeline */ =20 + res =3D PQgetResult(conn); + if (res =3D=3D NULL) + pg_fatal("PQgetResult returned null when there's a pipeline item: %s", + PQerrorMessage(conn)); + + if (PQresultStatus(res) !=3D PGRES_TUPLES_OK) + pg_fatal("Unexpected result code %s from first pipeline item", + PQresStatus(PQresultStatus(res))); + PQclear(res); + res =3D NULL; + + if (PQgetResult(conn) !=3D NULL) + pg_fatal("PQgetResult returned something extra after first result"); + + if (PQexitPipelineMode(conn) !=3D 0) + pg_fatal("exiting pipeline mode after query but before sync succeeded in= correctly"); + + res =3D PQgetResult(conn); + if (res =3D=3D NULL) + pg_fatal("PQgetResult returned null when sync result expected: %s", + PQerrorMessage(conn)); + + if (PQresultStatus(res) !=3D PGRES_PIPELINE_SYNC) + pg_fatal("Unexpected result code %s instead of sync result, error: %s", + PQresStatus(PQresultStatus(res)), PQerrorMessage(conn)); + PQclear(res); + + /* third pipeline */ + res =3D PQgetResult(conn); if (res =3D=3D NULL) pg_fatal("PQgetResult returned null when there's a pipeline item: %s", diff --git a/src/test/modules/libpq_pipeline/traces/multi_pipelines.trace b= /src/test/modules/libpq_pipeline/traces/multi_pipelines.trace index 4b9ab07ca4..1ee21f61dc 100644 --- a/src/test/modules/libpq_pipeline/traces/multi_pipelines.trace +++ b/src/test/modules/libpq_pipeline/traces/multi_pipelines.trace @@ -8,6 +8,17 @@ F 19 Bind "" "" 0 1 1 '1' 1 0 F 6 Describe P "" F 9 Execute "" 0 F 4 Sync +F 21 Parse "" "SELECT $1" 1 NNNN +F 19 Bind "" "" 0 1 1 '1' 1 0 +F 6 Describe P "" +F 9 Execute "" 0 +F 4 Sync +B 4 ParseComplete +B 4 BindComplete +B 33 RowDescription 1 "?column?" NNNN 0 NNNN 4 -1 0 +B 11 DataRow 1 1 '1' +B 13 CommandComplete "SELECT 1" +B 5 ReadyForQuery I B 4 ParseComplete B 4 BindComplete B 33 RowDescription 1 "?column?" NNNN 0 NNNN 4 -1 0 diff --git a/doc/src/sgml/libpq.sgml b/doc/src/sgml/libpq.sgml index 21195e0e72..6d1d477276 100644 --- a/doc/src/sgml/libpq.sgml +++ b/doc/src/sgml/libpq.sgml @@ -3547,8 +3547,9 @@ ExecStatusType PQresultStatus(const PGresult *res); The PGresult represents a - synchronization point in pipeline mode, requested by - . + synchronization point in pipeline mode, requested by either + or + . This status occurs only when pipeline mode has been selected. @@ -5122,7 +5123,8 @@ int PQsendClosePortal(PGconn *conn, const char *porta= lName); , , , - , or + , + , or call, and returns it. A null pointer is returned when the command is complete and there @@ -5507,8 +5509,9 @@ int PQflush(PGconn *conn); client sends them. The server will begin executing the commands in t= he pipeline immediately, not waiting for the end of the pipeline. Note that results are buffered on the server side; the server flushes - that buffer when a synchronization point is established with - PQpipelineSync, or when + that buffer when a synchronization point is established with either + PQpipelineSync or + PQsendPipelineSync, or when PQsendFlushRequest is called. If any statement encounters an error, the server aborts the current transaction and does not execute any subsequent command in the queue @@ -5565,7 +5568,8 @@ int PQflush(PGconn *conn); PGresult types PGRES_PIPELINE_SYNC and PGRES_PIPELINE_ABORTED. PGRES_PIPELINE_SYNC is reported exactly once for e= ach - PQpipelineSync at the corresponding point + PQpipelineSync or + PQsendPipelineSync at the corresponding point in the pipeline. PGRES_PIPELINE_ABORTED is emitted in place of a no= rmal query result for the first error and all subsequent results @@ -5603,7 +5607,8 @@ int PQflush(PGconn *conn); PQresultStatus will report a PGRES_PIPELINE_ABORTED result for each remaining q= ueued operation in an aborted pipeline. The result for - PQpipelineSync is reported as + PQpipelineSync or + PQsendPipelineSync is reported as PGRES_PIPELINE_SYNC to signal the end of the abort= ed pipeline and resumption of normal result processing. @@ -5835,6 +5840,32 @@ int PQsendFlushRequest(PGconn *conn); + + + PQsendPipelineSyncPQse= ndPipelineSync + + + + Marks a synchronization point in a pipeline by sending a + sync message + without flushing the send buffer. This serves as + the delimiter of an implicit transaction and an error recovery + point; see . + + +int PQsendPipelineSync(PGconn *conn); + + + + Returns 1 for success. Returns 0 if the connection is not in + pipeline mode or sending a + sync message + failed. + Note that the message is not itself flushed to the server automatic= ally; + use PQflush if necessary. + + + =20 --=20 2.43.0 --mv3uqqS97jfWEvjE-- --epws8oteElA/2YUs Content-Type: application/pgp-signature; name="signature.asc" -----BEGIN PGP SIGNATURE----- iQIzBAABCgAdFiEEG72nH6vTowiyblFKnvQgOdbyQH0FAmWk468ACgkQnvQgOdby QH3cHxAAkJAu3cMUlhwgsUIiBhmVyyNqDzRLuN48t8jZOlPNs7PYqwyGRE95HTl8 MPFrelaxSvY6IojUxB8pT/AaE5RjfmES3HxRgG+WcdlXIEZFVCuU24eSnebUW2g0 LhqGd5M/BFWh0OXfQaV8oLSWuBxdpLQ3I5bb3CM9eMElH8POQ0wVEjvENUwap0VF jSf+b4L3wKMFB0Q0+ZBex/DghrgNEISXKs4kyIFWVsI+v+BOfV3CY3VfscrG9tax TpvSjZRVs1m1zY8kbpBmMxrBx2OxUMm0A9Jv/FEdv03Vm5V5vjRTiWQ3caFRRIGL CN8zr1uedChRdowSCR73IjrDXaG/3evX55Y5mIMQCzyO8DiwSxIwVtAYP8OvztvV sIndi0qurEg7UTgkDwYNkknEvJTOUub+ANyOM72DrjyEDnWqfixBgBl2v/yR0Nrf vjfJN7xf/WDBvH+F24gpe13u6zRlHP59EEePVjBEg3rWs0E+Fk3QPjGSJkxAtho6 5/inmAIdynKzvYaX6z1VeuhgN9GMkQXAnRPfLA2ZJQX85lSpt5zQ+GtieRgLcjQN 5fiHndK2j3L+fuevAOKkSbNprPYow+FFZAz5BKtVBau6jO3eWyrvmBShC4r4yEMC WsoGrViLAbGL8S4PKZZwoQZhwqMga+yDKHNR0vGvkhm86x4oDPY= =tj4J -----END PGP SIGNATURE----- --epws8oteElA/2YUs--