Movatterモバイル変換

[0]ホーム
URL:

Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

 Sign up
Appearance settings

Commit4794c2d

Browse files
committed
libpq: Add PQsendPipelineSync()
This new function is equivalent to PQpipelineSync(), except that it doesnot flush anything to the server except if the size threshold of theoutput buffer is reached; the user must subsequently call PQflush()instead.Its purpose is to reduce the system call overhead of pipeline mode, bygiving to applications more control over the timing of the flushes whenmanipulating commands in pipeline mode.Author: Anton KirilovReviewed-by: Jelte Fennema-Nio, Robert Haas, Álvaro Herrera, DenisLaxalde, Michael PaquierDiscussion:https://postgr.es/m/CACV6eE5arHFZEA717=iKEa_OewpVFfWJOmsOdGrqqsr8CJVfWQ@mail.gmail.com
1 parent83eb244 commit4794c2d

File tree

6 files changed

+138
-17
lines changed

6 files changed

+138
-17
lines changed

‎doc/src/sgml/libpq.sgml

Lines changed: 38 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -3547,8 +3547,9 @@ ExecStatusType PQresultStatus(const PGresult *res);
35473547
<listitem>
35483548
<para>
35493549
The <structname>PGresult</structname> represents a
3550-
synchronization point in pipeline mode, requested by
3551-
<xref linkend="libpq-PQpipelineSync"/>.
3550+
synchronization point in pipeline mode, requested by either
3551+
<xref linkend="libpq-PQpipelineSync"/> or
3552+
<xref linkend="libpq-PQsendPipelineSync"/>.
35523553
This status occurs only when pipeline mode has been selected.
35533554
</para>
35543555
</listitem>
@@ -5122,7 +5123,8 @@ int PQsendClosePortal(PGconn *conn, const char *portalName);
51225123
<xref linkend="libpq-PQsendDescribePrepared"/>,
51235124
<xref linkend="libpq-PQsendDescribePortal"/>,
51245125
<xref linkend="libpq-PQsendClosePrepared"/>,
5125-
<xref linkend="libpq-PQsendClosePortal"/>, or
5126+
<xref linkend="libpq-PQsendClosePortal"/>,
5127+
<xref linkend="libpq-PQsendPipelineSync"/>, or
51265128
<xref linkend="libpq-PQpipelineSync"/>
51275129
call, and returns it.
51285130
A null pointer is returned when the command is complete and there
@@ -5507,8 +5509,9 @@ int PQflush(PGconn *conn);
55075509
client sends them. The server will begin executing the commands in the
55085510
pipeline immediately, not waiting for the end of the pipeline.
55095511
Note that results are buffered on the server side; the server flushes
5510-
that buffer when a synchronization point is established with
5511-
<function>PQpipelineSync</function>, or when
5512+
that buffer when a synchronization point is established with either
5513+
<function>PQpipelineSync</function> or
5514+
<function>PQsendPipelineSync</function>, or when
55125515
<function>PQsendFlushRequest</function> is called.
55135516
If any statement encounters an error, the server aborts the current
55145517
transaction and does not execute any subsequent command in the queue
@@ -5565,7 +5568,8 @@ int PQflush(PGconn *conn);
55655568
<type>PGresult</type> types <literal>PGRES_PIPELINE_SYNC</literal>
55665569
and <literal>PGRES_PIPELINE_ABORTED</literal>.
55675570
<literal>PGRES_PIPELINE_SYNC</literal> is reported exactly once for each
5568-
<function>PQpipelineSync</function> at the corresponding point
5571+
<function>PQpipelineSync</function> or
5572+
<function>PQsendPipelineSync</function> at the corresponding point
55695573
in the pipeline.
55705574
<literal>PGRES_PIPELINE_ABORTED</literal> is emitted in place of a normal
55715575
query result for the first error and all subsequent results
@@ -5603,7 +5607,8 @@ int PQflush(PGconn *conn);
56035607
<function>PQresultStatus</function> will report a
56045608
<literal>PGRES_PIPELINE_ABORTED</literal> result for each remaining queued
56055609
operation in an aborted pipeline. The result for
5606-
<function>PQpipelineSync</function> is reported as
5610+
<function>PQpipelineSync</function> or
5611+
<function>PQsendPipelineSync</function> is reported as
56075612
<literal>PGRES_PIPELINE_SYNC</literal> to signal the end of the aborted pipeline
56085613
and resumption of normal result processing.
56095614
</para>
@@ -5810,6 +5815,32 @@ int PQpipelineSync(PGconn *conn);
58105815
</listitem>
58115816
</varlistentry>
58125817

5818+
<varlistentry id="libpq-PQsendPipelineSync">
5819+
<term><function>PQsendPipelineSync</function><indexterm><primary>PQsendPipelineSync</primary></indexterm></term>
5820+
5821+
<listitem>
5822+
<para>
5823+
Marks a synchronization point in a pipeline by sending a
5824+
<link linkend="protocol-flow-ext-query">sync message</link>
5825+
without flushing the send buffer. This serves as
5826+
the delimiter of an implicit transaction and an error recovery
5827+
point; see <xref linkend="libpq-pipeline-errors"/>.
5828+
5829+
<synopsis>
5830+
int PQsendPipelineSync(PGconn *conn);
5831+
</synopsis>
5832+
</para>
5833+
<para>
5834+
Returns 1 for success. Returns 0 if the connection is not in
5835+
pipeline mode or sending a
5836+
<link linkend="protocol-flow-ext-query">sync message</link>
5837+
failed.
5838+
Note that the message is not itself flushed to the server automatically;
5839+
use <function>PQflush</function> if necessary.
5840+
</para>
5841+
</listitem>
5842+
</varlistentry>
5843+
58135844
<varlistentry id="libpq-PQsendFlushRequest">
58145845
<term><function>PQsendFlushRequest</function><indexterm><primary>PQsendFlushRequest</primary></indexterm></term>
58155846

‎src/interfaces/libpq/exports.txt

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -192,3 +192,4 @@ PQclosePortal 189
192192
PQsendClosePrepared 190
193193
PQsendClosePortal 191
194194
PQchangePassword 192
195+
PQsendPipelineSync 193

‎src/interfaces/libpq/fe-exec.c

Lines changed: 44 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -81,6 +81,7 @@ static intPQsendTypedCommand(PGconn *conn, char command, char type,
8181
constchar*target);
8282
staticintcheck_field_number(constPGresult*res,intfield_num);
8383
staticvoidpqPipelineProcessQueue(PGconn*conn);
84+
staticintpqPipelineSyncInternal(PGconn*conn,boolimmediate_flush);
8485
staticintpqPipelineFlush(PGconn*conn);
8586

8687

@@ -3224,25 +3225,48 @@ pqPipelineProcessQueue(PGconn *conn)
32243225
/*
32253226
* PQpipelineSync
32263227
*Send a Sync message as part of a pipeline, and flush to server
3228+
*/
3229+
int
3230+
PQpipelineSync(PGconn*conn)
3231+
{
3232+
returnpqPipelineSyncInternal(conn, true);
3233+
}
3234+
3235+
/*
3236+
* PQsendPipelineSync
3237+
*Send a Sync message as part of a pipeline, without flushing to server
3238+
*/
3239+
int
3240+
PQsendPipelineSync(PGconn*conn)
3241+
{
3242+
returnpqPipelineSyncInternal(conn, false);
3243+
}
3244+
3245+
/*
3246+
* Workhorse function for PQpipelineSync and PQsendPipelineSync.
32273247
*
32283248
* It's legal to start submitting more commands in the pipeline immediately,
32293249
* without waiting for the results of the current pipeline. There's no need to
32303250
* end pipeline mode and start it again.
32313251
*
3232-
* If a command in a pipeline fails, every subsequent command up to and including
3233-
* the result to the Sync message sent byPQpipelineSync gets set to
3234-
* PGRES_PIPELINE_ABORTED state. If the whole pipeline is processed without
3235-
* error, a PGresult with PGRES_PIPELINE_SYNC is produced.
3252+
* If a command in a pipeline fails, every subsequent command up to and
3253+
*includingthe result to the Sync message sent bypqPipelineSyncInternal
3254+
*gets set toPGRES_PIPELINE_ABORTED state. If the whole pipeline is
3255+
*processed withouterror, a PGresult with PGRES_PIPELINE_SYNC is produced.
32363256
*
3237-
* Queries can already have been sent before PQpipelineSync is called, but
3238-
* PQpipelineSync needs to be called before retrieving command results.
3257+
* Queries can already have been sent before pqPipelineSyncInternal is called,
3258+
* but pqPipelineSyncInternal needs to be called before retrieving command
3259+
* results.
32393260
*
32403261
* The connection will remain in pipeline mode and unavailable for new
32413262
* synchronous command execution functions until all results from the pipeline
32423263
* are processed by the client.
3264+
*
3265+
* immediate_flush controls if the flush happens immediately after sending the
3266+
* Sync message or not.
32433267
*/
3244-
int
3245-
PQpipelineSync(PGconn*conn)
3268+
staticint
3269+
pqPipelineSyncInternal(PGconn*conn,boolimmediate_flush)
32463270
{
32473271
PGcmdQueueEntry*entry;
32483272

@@ -3288,9 +3312,19 @@ PQpipelineSync(PGconn *conn)
32883312
/*
32893313
* Give the data a push. In nonblock mode, don't complain if we're unable
32903314
* to send it all; PQgetResult() will do any additional flushing needed.
3315+
* If immediate_flush is disabled, the data is pushed if we are past the
3316+
* size threshold.
32913317
*/
3292-
if (PQflush(conn)<0)
3293-
gotosendFailed;
3318+
if (immediate_flush)
3319+
{
3320+
if (pqFlush(conn)<0)
3321+
gotosendFailed;
3322+
}
3323+
else
3324+
{
3325+
if (pqPipelineFlush(conn)<0)
3326+
gotosendFailed;
3327+
}
32943328

32953329
/* OK, it's launched! */
32963330
pqAppendCmdQueueEntry(conn,entry);

‎src/interfaces/libpq/libpq-fe.h

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -474,6 +474,7 @@ extern intPQenterPipelineMode(PGconn *conn);
474474
externintPQexitPipelineMode(PGconn*conn);
475475
externintPQpipelineSync(PGconn*conn);
476476
externintPQsendFlushRequest(PGconn*conn);
477+
externintPQsendPipelineSync(PGconn*conn);
477478

478479
/* LISTEN/NOTIFY support */
479480
externPGnotify*PQnotifies(PGconn*conn);

‎src/test/modules/libpq_pipeline/libpq_pipeline.c

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -162,21 +162,35 @@ test_multi_pipelines(PGconn *conn)
162162
if (PQenterPipelineMode(conn)!=1)
163163
pg_fatal("failed to enter pipeline mode: %s",PQerrorMessage(conn));
164164

165+
/* first pipeline */
165166
if (PQsendQueryParams(conn,"SELECT $1",1,dummy_param_oids,
166167
dummy_params,NULL,NULL,0)!=1)
167168
pg_fatal("dispatching first SELECT failed: %s",PQerrorMessage(conn));
168169

169170
if (PQpipelineSync(conn)!=1)
170171
pg_fatal("Pipeline sync failed: %s",PQerrorMessage(conn));
171172

173+
/* second pipeline */
172174
if (PQsendQueryParams(conn,"SELECT $1",1,dummy_param_oids,
173175
dummy_params,NULL,NULL,0)!=1)
174176
pg_fatal("dispatching second SELECT failed: %s",PQerrorMessage(conn));
175177

178+
/* Skip flushing once. */
179+
if (PQsendPipelineSync(conn)!=1)
180+
pg_fatal("Pipeline sync failed: %s",PQerrorMessage(conn));
181+
182+
/* third pipeline */
183+
if (PQsendQueryParams(conn,"SELECT $1",1,dummy_param_oids,
184+
dummy_params,NULL,NULL,0)!=1)
185+
pg_fatal("dispatching third SELECT failed: %s",PQerrorMessage(conn));
186+
176187
if (PQpipelineSync(conn)!=1)
177188
pg_fatal("pipeline sync failed: %s",PQerrorMessage(conn));
178189

179190
/* OK, start processing the results */
191+
192+
/* first pipeline */
193+
180194
res=PQgetResult(conn);
181195
if (res==NULL)
182196
pg_fatal("PQgetResult returned null when there's a pipeline item: %s",
@@ -214,6 +228,35 @@ test_multi_pipelines(PGconn *conn)
214228
if (PQresultStatus(res)!=PGRES_TUPLES_OK)
215229
pg_fatal("Unexpected result code %s from second pipeline item",
216230
PQresStatus(PQresultStatus(res)));
231+
PQclear(res);
232+
res=NULL;
233+
234+
if (PQgetResult(conn)!=NULL)
235+
pg_fatal("PQgetResult returned something extra after first result");
236+
237+
if (PQexitPipelineMode(conn)!=0)
238+
pg_fatal("exiting pipeline mode after query but before sync succeeded incorrectly");
239+
240+
res=PQgetResult(conn);
241+
if (res==NULL)
242+
pg_fatal("PQgetResult returned null when sync result expected: %s",
243+
PQerrorMessage(conn));
244+
245+
if (PQresultStatus(res)!=PGRES_PIPELINE_SYNC)
246+
pg_fatal("Unexpected result code %s instead of sync result, error: %s",
247+
PQresStatus(PQresultStatus(res)),PQerrorMessage(conn));
248+
PQclear(res);
249+
250+
/* third pipeline */
251+
252+
res=PQgetResult(conn);
253+
if (res==NULL)
254+
pg_fatal("PQgetResult returned null when there's a pipeline item: %s",
255+
PQerrorMessage(conn));
256+
257+
if (PQresultStatus(res)!=PGRES_TUPLES_OK)
258+
pg_fatal("Unexpected result code %s from third pipeline item",
259+
PQresStatus(PQresultStatus(res)));
217260

218261
res=PQgetResult(conn);
219262
if (res!=NULL)

‎src/test/modules/libpq_pipeline/traces/multi_pipelines.trace

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,17 @@ F19Bind "" "" 0 1 1 '1' 1 0
88
F6Describe P ""
99
F9Execute "" 0
1010
F4Sync
11+
F21Parse "" "SELECT $1" 1 NNNN
12+
F19Bind "" "" 0 1 1 '1' 1 0
13+
F6Describe P ""
14+
F9Execute "" 0
15+
F4Sync
16+
B4ParseComplete
17+
B4BindComplete
18+
B33RowDescription 1 "?column?" NNNN 0 NNNN 4 -1 0
19+
B11DataRow 1 1 '1'
20+
B13CommandComplete "SELECT 1"
21+
B5ReadyForQuery I
1122
B4ParseComplete
1223
B4BindComplete
1324
B33RowDescription 1 "?column?" NNNN 0 NNNN 4 -1 0

0 commit comments

Comments
 (0)
[8]ページ先頭
©2009-2025 Movatter.jp