public inbox for [email protected]help / color / mirror / Atom feed
[PATCH v6] Avoid going IDLE in pipeline mode 7+ messages / 4 participants [nested] [flat]
* [PATCH v6] Avoid going IDLE in pipeline mode @ 2022-06-15 17:56 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 7+ messages in thread From: Alvaro Herrera @ 2022-06-15 17:56 UTC (permalink / raw) Introduce a new PGASYNC_PIPELINE_IDLE state, which helps PQgetResult distinguish the case of "really idle". This fixes the problem that ReadyForQuery is sent too soon, which caused a CloseComplete to be received when in idle state. XXX -- this is still WIP. Co-authored-by: Kyotaro Horiguchi <[email protected]> Reported-by: Daniele Varrazzo <[email protected]> Discussion: https://postgr.es/m/CA+mi_8bvD0_CW3sumgwPvWdNzXY32itoG_16tDYRu_1S2gV2iw@mail.gmail.com --- src/interfaces/libpq/fe-connect.c | 1 + src/interfaces/libpq/fe-exec.c | 56 +++++++---- src/interfaces/libpq/fe-protocol3.c | 12 --- src/interfaces/libpq/libpq-int.h | 3 +- .../modules/libpq_pipeline/libpq_pipeline.c | 99 +++++++++++++++++++ .../traces/simple_pipeline.trace | 37 +++++++ 6 files changed, 175 insertions(+), 33 deletions(-) diff --git a/src/interfaces/libpq/fe-connect.c b/src/interfaces/libpq/fe-connect.c index 709ba15220..afd0bc809a 100644 --- a/src/interfaces/libpq/fe-connect.c +++ b/src/interfaces/libpq/fe-connect.c @@ -6751,6 +6751,7 @@ PQtransactionStatus(const PGconn *conn) { if (!conn || conn->status != CONNECTION_OK) return PQTRANS_UNKNOWN; + /* XXX what should we do here if status is PGASYNC_PIPELINE_IDLE? */ if (conn->asyncStatus != PGASYNC_IDLE) return PQTRANS_ACTIVE; return conn->xactStatus; diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c index 4180683194..3cf59e45e1 100644 --- a/src/interfaces/libpq/fe-exec.c +++ b/src/interfaces/libpq/fe-exec.c @@ -1279,7 +1279,8 @@ pqAppendCmdQueueEntry(PGconn *conn, PGcmdQueueEntry *entry) * itself consume commands from the queue; if we're in any other * state, we don't have to do anything. */ - if (conn->asyncStatus == PGASYNC_IDLE) + if (conn->asyncStatus == PGASYNC_IDLE || + conn->asyncStatus == PGASYNC_PIPELINE_IDLE) { resetPQExpBuffer(&conn->errorMessage); pqPipelineProcessQueue(conn); @@ -1642,9 +1643,9 @@ PQsendQueryStart(PGconn *conn, bool newQuery) return false; } - /* Can't send while already busy, either, unless enqueuing for later */ - if (conn->asyncStatus != PGASYNC_IDLE && - conn->pipelineStatus == PQ_PIPELINE_OFF) + /* In non-pipeline mode, we can't send anything while already busy */ + if (conn->pipelineStatus == PQ_PIPELINE_OFF && + conn->asyncStatus != PGASYNC_IDLE) { appendPQExpBufferStr(&conn->errorMessage, libpq_gettext("another command is already in progress\n")); @@ -1667,11 +1668,13 @@ PQsendQueryStart(PGconn *conn, bool newQuery) switch (conn->asyncStatus) { case PGASYNC_IDLE: + case PGASYNC_PIPELINE_IDLE: case PGASYNC_READY: case PGASYNC_READY_MORE: case PGASYNC_BUSY: /* ok to queue */ break; + case PGASYNC_COPY_IN: case PGASYNC_COPY_OUT: case PGASYNC_COPY_BOTH: @@ -1881,6 +1884,7 @@ PQsetSingleRowMode(PGconn *conn) */ if (!conn) return 0; + /* XXX modify this? */ if (conn->asyncStatus != PGASYNC_BUSY) return 0; if (!conn->cmd_queue_head || @@ -2047,19 +2051,19 @@ PQgetResult(PGconn *conn) { case PGASYNC_IDLE: res = NULL; /* query is complete */ - if (conn->pipelineStatus != PQ_PIPELINE_OFF) - { - /* - * We're about to return the NULL that terminates the round of - * results from the current query; prepare to send the results - * of the next query when we're called next. Also, since this - * is the start of the results of the next query, clear any - * prior error message. - */ - resetPQExpBuffer(&conn->errorMessage); - pqPipelineProcessQueue(conn); - } break; + case PGASYNC_PIPELINE_IDLE: + Assert(conn->pipelineStatus != PQ_PIPELINE_OFF); + + res = NULL; /* query is complete */ + /* + * We're about to return the NULL that terminates the round of + * results from the current query; prepare to send the results + * of the next query when we're called next. + */ + pqPipelineProcessQueue(conn); + break; + case PGASYNC_READY: /* @@ -2080,7 +2084,7 @@ PQgetResult(PGconn *conn) * We're about to send the results of the current query. Set * us idle now, and ... */ - conn->asyncStatus = PGASYNC_IDLE; + conn->asyncStatus = PGASYNC_PIPELINE_IDLE; /* * ... in cases when we're sending a pipeline-sync result, @@ -3008,17 +3012,28 @@ pqPipelineProcessQueue(PGconn *conn) case PGASYNC_READY: case PGASYNC_READY_MORE: case PGASYNC_BUSY: + case PGASYNC_IDLE: /* client still has to process current query or results */ return; - case PGASYNC_IDLE: + case PGASYNC_PIPELINE_IDLE: /* next query please */ break; } /* Nothing to do if not in pipeline mode, or queue is empty */ - if (conn->pipelineStatus == PQ_PIPELINE_OFF || - conn->cmd_queue_head == NULL) + if (conn->pipelineStatus == PQ_PIPELINE_OFF) + { + conn->asyncStatus = PGASYNC_IDLE; return; + } + + if (conn->cmd_queue_head == NULL) + { + if (conn->pipelineStatus != PQ_PIPELINE_ABORTED) + conn->asyncStatus = PGASYNC_IDLE; + + return; + } /* Initialize async result-accumulation state */ pqClearAsyncResult(conn); @@ -3105,6 +3120,7 @@ PQpipelineSync(PGconn *conn) case PGASYNC_READY_MORE: case PGASYNC_BUSY: case PGASYNC_IDLE: + case PGASYNC_PIPELINE_IDLE: /* OK to send sync */ break; } diff --git a/src/interfaces/libpq/fe-protocol3.c b/src/interfaces/libpq/fe-protocol3.c index 9ab3bf1fcb..bab8926a63 100644 --- a/src/interfaces/libpq/fe-protocol3.c +++ b/src/interfaces/libpq/fe-protocol3.c @@ -158,18 +158,6 @@ pqParseInput3(PGconn *conn) if (conn->asyncStatus != PGASYNC_IDLE) return; - /* - * We're also notionally not-IDLE when in pipeline mode the state - * says "idle" (so we have completed receiving the results of one - * query from the server and dispatched them to the application) - * but another query is queued; yield back control to caller so - * that they can initiate processing of the next query in the - * queue. - */ - if (conn->pipelineStatus != PQ_PIPELINE_OFF && - conn->cmd_queue_head != NULL) - return; - /* * Unexpected message in IDLE state; need to recover somehow. * ERROR messages are handled using the notice processor; diff --git a/src/interfaces/libpq/libpq-int.h b/src/interfaces/libpq/libpq-int.h index 334aea4b6e..44a65e41b7 100644 --- a/src/interfaces/libpq/libpq-int.h +++ b/src/interfaces/libpq/libpq-int.h @@ -224,7 +224,8 @@ typedef enum * query */ PGASYNC_COPY_IN, /* Copy In data transfer in progress */ PGASYNC_COPY_OUT, /* Copy Out data transfer in progress */ - PGASYNC_COPY_BOTH /* Copy In/Out data transfer in progress */ + PGASYNC_COPY_BOTH, /* Copy In/Out data transfer in progress */ + PGASYNC_PIPELINE_IDLE, /* Pipeline mode */ } PGAsyncStatusType; /* Target server type (decoded value of target_session_attrs) */ diff --git a/src/test/modules/libpq_pipeline/libpq_pipeline.c b/src/test/modules/libpq_pipeline/libpq_pipeline.c index c27c4e0ada..2bdd4e308f 100644 --- a/src/test/modules/libpq_pipeline/libpq_pipeline.c +++ b/src/test/modules/libpq_pipeline/libpq_pipeline.c @@ -968,15 +968,29 @@ test_prepared(PGconn *conn) fprintf(stderr, "ok\n"); } +/* Notice processor: print them, and count how many we got */ +static void +notice_processor(void *arg, const char *message) +{ + int *n_notices = (int *) arg; + + (*n_notices)++; + fprintf(stderr, "NOTICE %d: %s", *n_notices, message); +} + static void test_simple_pipeline(PGconn *conn) { PGresult *res = NULL; const char *dummy_params[1] = {"1"}; Oid dummy_param_oids[1] = {INT4OID}; + PQnoticeProcessor oldproc; + int n_notices = 0; fprintf(stderr, "simple pipeline... "); + oldproc = PQsetNoticeProcessor(conn, notice_processor, &n_notices); + /* * Enter pipeline mode and dispatch a set of operations, which we'll then * process the results of as they come in. @@ -1052,6 +1066,91 @@ test_simple_pipeline(PGconn *conn) if (PQpipelineStatus(conn) != PQ_PIPELINE_OFF) pg_fatal("Exiting pipeline mode didn't seem to work"); + if (n_notices > 0) + pg_fatal("unexpected notice"); + + + /* Try the same thing with PQsendQuery */ + if (PQenterPipelineMode(conn) != 1) + pg_fatal("failed to enter pipeline mode: %s", PQerrorMessage(conn)); + + if (PQsendQuery(conn, "SELECT 1") != 1) + pg_fatal("failed to send query: %s", PQerrorMessage(conn)); + PQsendFlushRequest(conn); + res = PQgetResult(conn); + if (res == NULL) + pg_fatal("PQgetResult returned null when there's a pipeline item: %s", + PQerrorMessage(conn)); + if (PQresultStatus(res) != PGRES_TUPLES_OK) + pg_fatal("Unexpected result code %s from first pipeline item", + PQresStatus(PQresultStatus(res))); + PQclear(res); + + res = PQgetResult(conn); + if (res != NULL) + pg_fatal("expected NULL result"); + + if (PQpipelineSync(conn) != 1) + pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn)); + res = PQgetResult(conn); + if (res == NULL) + pg_fatal("PQgetResult returned null when there's a pipeline item: %s", + PQerrorMessage(conn)); + if (PQresultStatus(res) != PGRES_PIPELINE_SYNC) + pg_fatal("Unexpected result code %s instead of PGRES_PIPELINE_SYNC, error: %s", + PQresStatus(PQresultStatus(res)), PQerrorMessage(conn)); + PQclear(res); + res = NULL; + + if (PQexitPipelineMode(conn) != 1) + pg_fatal("attempt to exit pipeline mode failed when it should've succeeded: %s", + PQerrorMessage(conn)); + + /* + * Must not have got any notices here; note bug as described in + * https://postgr.es/m/CA+mi_8bvD0_CW3sumgwPvWdNzXY32itoG_16tDYRu_1S2gV2iw@mail.gmail.com + */ + if (n_notices > 0) + pg_fatal("got %d notice(s)", n_notices); + + PQsetNoticeProcessor(conn, oldproc, NULL); + + /* + * Send a second command when libpq is in "pipeline-idle" state. + */ + if (PQenterPipelineMode(conn) != 1) + pg_fatal("failed to enter pipeline mode: %s", PQerrorMessage(conn)); + if (PQsendQuery(conn, "SELECT 1") != 1) + pg_fatal("failed to send query: %s", PQerrorMessage(conn)); + PQsendFlushRequest(conn); + res = PQgetResult(conn); + if (res == NULL) + pg_fatal("PQgetResult returned null when there's a pipeline item: %s", + PQerrorMessage(conn)); + if (PQresultStatus(res) != PGRES_TUPLES_OK) + pg_fatal("unexpected result code %s from first pipeline item", + PQresStatus(PQresultStatus(res))); + if (PQsendQuery(conn, "SELECT 2") != 1) + pg_fatal("failed to send query: %s", PQerrorMessage(conn)); + PQsendFlushRequest(conn); + /* read terminating null from first query */ + res = PQgetResult(conn); + if (res != NULL) + pg_fatal("did not receive terminating NULL"); + res = PQgetResult(conn); + if (res == NULL) + pg_fatal("PQgetResult returned null when there's a pipeline item: %s", + PQerrorMessage(conn)); + if (PQresultStatus(res) != PGRES_TUPLES_OK) + pg_fatal("unexpected result code %s from first pipeline item", + PQresStatus(PQresultStatus(res))); + res = PQgetResult(conn); + if (res != NULL) + pg_fatal("did not receive terminating NULL"); + if (PQexitPipelineMode(conn) != 1) + pg_fatal("attempt to exit pipeline mode failed when it should've succeeded: %s", + PQerrorMessage(conn)); + fprintf(stderr, "ok\n"); } diff --git a/src/test/modules/libpq_pipeline/traces/simple_pipeline.trace b/src/test/modules/libpq_pipeline/traces/simple_pipeline.trace index 5c94749bc1..e1c0cba07e 100644 --- a/src/test/modules/libpq_pipeline/traces/simple_pipeline.trace +++ b/src/test/modules/libpq_pipeline/traces/simple_pipeline.trace @@ -9,4 +9,41 @@ 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 +F 16 Parse "" "SELECT 1" 0 +F 12 Bind "" "" 0 0 0 +F 6 Describe P "" +F 9 Execute "" 0 +F 6 Close P "" +F 4 Flush +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" +F 4 Sync +B 4 CloseComplete +B 5 ReadyForQuery I +F 16 Parse "" "SELECT 1" 0 +F 12 Bind "" "" 0 0 0 +F 6 Describe P "" +F 9 Execute "" 0 +F 6 Close P "" +F 4 Flush +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" +F 16 Parse "" "SELECT 2" 0 +F 12 Bind "" "" 0 0 0 +F 6 Describe P "" +F 9 Execute "" 0 +F 6 Close P "" +F 4 Flush +B 4 CloseComplete +B 4 ParseComplete +B 4 BindComplete +B 33 RowDescription 1 "?column?" NNNN 0 NNNN 4 -1 0 +B 11 DataRow 1 1 '2' +B 13 CommandComplete "SELECT 1" F 4 Terminate -- 2.31.1 ----Next_Part(Tue_Jun_21_17_46_54_2022_518)---- ^ permalink raw reply [nested|flat] 7+ messages in thread
* [PATCH v5 1/2] Avoid going IDLE in pipeline mode @ 2022-06-15 17:56 Alvaro Herrera <[email protected]> 0 siblings, 0 replies; 7+ messages in thread From: Alvaro Herrera @ 2022-06-15 17:56 UTC (permalink / raw) Introduce a new PGASYNC_PIPELINE_IDLE state, which helps PQgetResult distinguish the case of "really idle". This fixes the problem that ReadyForQuery is sent too soon, which caused a CloseComplete to be received when in idle state. XXX -- this is still WIP. Co-authored-by: Kyotaro Horiguchi <[email protected]> Reported-by: Daniele Varrazzo <[email protected]> Discussion: https://postgr.es/m/CA+mi_8bvD0_CW3sumgwPvWdNzXY32itoG_16tDYRu_1S2gV2iw@mail.gmail.com --- src/interfaces/libpq/fe-connect.c | 1 + src/interfaces/libpq/fe-exec.c | 45 +++++---- src/interfaces/libpq/fe-protocol3.c | 12 --- src/interfaces/libpq/libpq-int.h | 3 +- .../modules/libpq_pipeline/libpq_pipeline.c | 99 +++++++++++++++++++ .../traces/simple_pipeline.trace | 37 +++++++ 6 files changed, 166 insertions(+), 31 deletions(-) diff --git a/src/interfaces/libpq/fe-connect.c b/src/interfaces/libpq/fe-connect.c index 709ba15220..afd0bc809a 100644 --- a/src/interfaces/libpq/fe-connect.c +++ b/src/interfaces/libpq/fe-connect.c @@ -6751,6 +6751,7 @@ PQtransactionStatus(const PGconn *conn) { if (!conn || conn->status != CONNECTION_OK) return PQTRANS_UNKNOWN; + /* XXX what should we do here if status is PGASYNC_PIPELINE_IDLE? */ if (conn->asyncStatus != PGASYNC_IDLE) return PQTRANS_ACTIVE; return conn->xactStatus; diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c index 4180683194..59f2e7f724 100644 --- a/src/interfaces/libpq/fe-exec.c +++ b/src/interfaces/libpq/fe-exec.c @@ -1279,7 +1279,8 @@ pqAppendCmdQueueEntry(PGconn *conn, PGcmdQueueEntry *entry) * itself consume commands from the queue; if we're in any other * state, we don't have to do anything. */ - if (conn->asyncStatus == PGASYNC_IDLE) + if (conn->asyncStatus == PGASYNC_IDLE || + conn->asyncStatus == PGASYNC_PIPELINE_IDLE) { resetPQExpBuffer(&conn->errorMessage); pqPipelineProcessQueue(conn); @@ -1642,9 +1643,9 @@ PQsendQueryStart(PGconn *conn, bool newQuery) return false; } - /* Can't send while already busy, either, unless enqueuing for later */ - if (conn->asyncStatus != PGASYNC_IDLE && - conn->pipelineStatus == PQ_PIPELINE_OFF) + /* In non-pipeline mode, we can't send anything while already busy */ + if (conn->pipelineStatus == PQ_PIPELINE_OFF && + conn->asyncStatus != PGASYNC_IDLE) { appendPQExpBufferStr(&conn->errorMessage, libpq_gettext("another command is already in progress\n")); @@ -1667,11 +1668,13 @@ PQsendQueryStart(PGconn *conn, bool newQuery) switch (conn->asyncStatus) { case PGASYNC_IDLE: + case PGASYNC_PIPELINE_IDLE: case PGASYNC_READY: case PGASYNC_READY_MORE: case PGASYNC_BUSY: /* ok to queue */ break; + case PGASYNC_COPY_IN: case PGASYNC_COPY_OUT: case PGASYNC_COPY_BOTH: @@ -1881,6 +1884,7 @@ PQsetSingleRowMode(PGconn *conn) */ if (!conn) return 0; + /* XXX modify this? */ if (conn->asyncStatus != PGASYNC_BUSY) return 0; if (!conn->cmd_queue_head || @@ -2047,19 +2051,19 @@ PQgetResult(PGconn *conn) { case PGASYNC_IDLE: res = NULL; /* query is complete */ - if (conn->pipelineStatus != PQ_PIPELINE_OFF) - { - /* - * We're about to return the NULL that terminates the round of - * results from the current query; prepare to send the results - * of the next query when we're called next. Also, since this - * is the start of the results of the next query, clear any - * prior error message. - */ - resetPQExpBuffer(&conn->errorMessage); - pqPipelineProcessQueue(conn); - } break; + case PGASYNC_PIPELINE_IDLE: + Assert(conn->pipelineStatus != PQ_PIPELINE_OFF); + + res = NULL; /* query is complete */ + /* + * We're about to return the NULL that terminates the round of + * results from the current query; prepare to send the results + * of the next query when we're called next. + */ + pqPipelineProcessQueue(conn); + break; + case PGASYNC_READY: /* @@ -2080,7 +2084,7 @@ PQgetResult(PGconn *conn) * We're about to send the results of the current query. Set * us idle now, and ... */ - conn->asyncStatus = PGASYNC_IDLE; + conn->asyncStatus = PGASYNC_PIPELINE_IDLE; /* * ... in cases when we're sending a pipeline-sync result, @@ -3008,9 +3012,10 @@ pqPipelineProcessQueue(PGconn *conn) case PGASYNC_READY: case PGASYNC_READY_MORE: case PGASYNC_BUSY: + case PGASYNC_IDLE: /* client still has to process current query or results */ return; - case PGASYNC_IDLE: + case PGASYNC_PIPELINE_IDLE: /* next query please */ break; } @@ -3018,7 +3023,10 @@ pqPipelineProcessQueue(PGconn *conn) /* Nothing to do if not in pipeline mode, or queue is empty */ if (conn->pipelineStatus == PQ_PIPELINE_OFF || conn->cmd_queue_head == NULL) + { + conn->asyncStatus = PGASYNC_IDLE; return; + } /* Initialize async result-accumulation state */ pqClearAsyncResult(conn); @@ -3105,6 +3113,7 @@ PQpipelineSync(PGconn *conn) case PGASYNC_READY_MORE: case PGASYNC_BUSY: case PGASYNC_IDLE: + case PGASYNC_PIPELINE_IDLE: /* OK to send sync */ break; } diff --git a/src/interfaces/libpq/fe-protocol3.c b/src/interfaces/libpq/fe-protocol3.c index 9ab3bf1fcb..bab8926a63 100644 --- a/src/interfaces/libpq/fe-protocol3.c +++ b/src/interfaces/libpq/fe-protocol3.c @@ -158,18 +158,6 @@ pqParseInput3(PGconn *conn) if (conn->asyncStatus != PGASYNC_IDLE) return; - /* - * We're also notionally not-IDLE when in pipeline mode the state - * says "idle" (so we have completed receiving the results of one - * query from the server and dispatched them to the application) - * but another query is queued; yield back control to caller so - * that they can initiate processing of the next query in the - * queue. - */ - if (conn->pipelineStatus != PQ_PIPELINE_OFF && - conn->cmd_queue_head != NULL) - return; - /* * Unexpected message in IDLE state; need to recover somehow. * ERROR messages are handled using the notice processor; diff --git a/src/interfaces/libpq/libpq-int.h b/src/interfaces/libpq/libpq-int.h index 334aea4b6e..44a65e41b7 100644 --- a/src/interfaces/libpq/libpq-int.h +++ b/src/interfaces/libpq/libpq-int.h @@ -224,7 +224,8 @@ typedef enum * query */ PGASYNC_COPY_IN, /* Copy In data transfer in progress */ PGASYNC_COPY_OUT, /* Copy Out data transfer in progress */ - PGASYNC_COPY_BOTH /* Copy In/Out data transfer in progress */ + PGASYNC_COPY_BOTH, /* Copy In/Out data transfer in progress */ + PGASYNC_PIPELINE_IDLE, /* Pipeline mode */ } PGAsyncStatusType; /* Target server type (decoded value of target_session_attrs) */ diff --git a/src/test/modules/libpq_pipeline/libpq_pipeline.c b/src/test/modules/libpq_pipeline/libpq_pipeline.c index c27c4e0ada..2bdd4e308f 100644 --- a/src/test/modules/libpq_pipeline/libpq_pipeline.c +++ b/src/test/modules/libpq_pipeline/libpq_pipeline.c @@ -968,15 +968,29 @@ test_prepared(PGconn *conn) fprintf(stderr, "ok\n"); } +/* Notice processor: print them, and count how many we got */ +static void +notice_processor(void *arg, const char *message) +{ + int *n_notices = (int *) arg; + + (*n_notices)++; + fprintf(stderr, "NOTICE %d: %s", *n_notices, message); +} + static void test_simple_pipeline(PGconn *conn) { PGresult *res = NULL; const char *dummy_params[1] = {"1"}; Oid dummy_param_oids[1] = {INT4OID}; + PQnoticeProcessor oldproc; + int n_notices = 0; fprintf(stderr, "simple pipeline... "); + oldproc = PQsetNoticeProcessor(conn, notice_processor, &n_notices); + /* * Enter pipeline mode and dispatch a set of operations, which we'll then * process the results of as they come in. @@ -1052,6 +1066,91 @@ test_simple_pipeline(PGconn *conn) if (PQpipelineStatus(conn) != PQ_PIPELINE_OFF) pg_fatal("Exiting pipeline mode didn't seem to work"); + if (n_notices > 0) + pg_fatal("unexpected notice"); + + + /* Try the same thing with PQsendQuery */ + if (PQenterPipelineMode(conn) != 1) + pg_fatal("failed to enter pipeline mode: %s", PQerrorMessage(conn)); + + if (PQsendQuery(conn, "SELECT 1") != 1) + pg_fatal("failed to send query: %s", PQerrorMessage(conn)); + PQsendFlushRequest(conn); + res = PQgetResult(conn); + if (res == NULL) + pg_fatal("PQgetResult returned null when there's a pipeline item: %s", + PQerrorMessage(conn)); + if (PQresultStatus(res) != PGRES_TUPLES_OK) + pg_fatal("Unexpected result code %s from first pipeline item", + PQresStatus(PQresultStatus(res))); + PQclear(res); + + res = PQgetResult(conn); + if (res != NULL) + pg_fatal("expected NULL result"); + + if (PQpipelineSync(conn) != 1) + pg_fatal("pipeline sync failed: %s", PQerrorMessage(conn)); + res = PQgetResult(conn); + if (res == NULL) + pg_fatal("PQgetResult returned null when there's a pipeline item: %s", + PQerrorMessage(conn)); + if (PQresultStatus(res) != PGRES_PIPELINE_SYNC) + pg_fatal("Unexpected result code %s instead of PGRES_PIPELINE_SYNC, error: %s", + PQresStatus(PQresultStatus(res)), PQerrorMessage(conn)); + PQclear(res); + res = NULL; + + if (PQexitPipelineMode(conn) != 1) + pg_fatal("attempt to exit pipeline mode failed when it should've succeeded: %s", + PQerrorMessage(conn)); + + /* + * Must not have got any notices here; note bug as described in + * https://postgr.es/m/CA+mi_8bvD0_CW3sumgwPvWdNzXY32itoG_16tDYRu_1S2gV2iw@mail.gmail.com + */ + if (n_notices > 0) + pg_fatal("got %d notice(s)", n_notices); + + PQsetNoticeProcessor(conn, oldproc, NULL); + + /* + * Send a second command when libpq is in "pipeline-idle" state. + */ + if (PQenterPipelineMode(conn) != 1) + pg_fatal("failed to enter pipeline mode: %s", PQerrorMessage(conn)); + if (PQsendQuery(conn, "SELECT 1") != 1) + pg_fatal("failed to send query: %s", PQerrorMessage(conn)); + PQsendFlushRequest(conn); + res = PQgetResult(conn); + if (res == NULL) + pg_fatal("PQgetResult returned null when there's a pipeline item: %s", + PQerrorMessage(conn)); + if (PQresultStatus(res) != PGRES_TUPLES_OK) + pg_fatal("unexpected result code %s from first pipeline item", + PQresStatus(PQresultStatus(res))); + if (PQsendQuery(conn, "SELECT 2") != 1) + pg_fatal("failed to send query: %s", PQerrorMessage(conn)); + PQsendFlushRequest(conn); + /* read terminating null from first query */ + res = PQgetResult(conn); + if (res != NULL) + pg_fatal("did not receive terminating NULL"); + res = PQgetResult(conn); + if (res == NULL) + pg_fatal("PQgetResult returned null when there's a pipeline item: %s", + PQerrorMessage(conn)); + if (PQresultStatus(res) != PGRES_TUPLES_OK) + pg_fatal("unexpected result code %s from first pipeline item", + PQresStatus(PQresultStatus(res))); + res = PQgetResult(conn); + if (res != NULL) + pg_fatal("did not receive terminating NULL"); + if (PQexitPipelineMode(conn) != 1) + pg_fatal("attempt to exit pipeline mode failed when it should've succeeded: %s", + PQerrorMessage(conn)); + fprintf(stderr, "ok\n"); } diff --git a/src/test/modules/libpq_pipeline/traces/simple_pipeline.trace b/src/test/modules/libpq_pipeline/traces/simple_pipeline.trace index 5c94749bc1..e1c0cba07e 100644 --- a/src/test/modules/libpq_pipeline/traces/simple_pipeline.trace +++ b/src/test/modules/libpq_pipeline/traces/simple_pipeline.trace @@ -9,4 +9,41 @@ 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 +F 16 Parse "" "SELECT 1" 0 +F 12 Bind "" "" 0 0 0 +F 6 Describe P "" +F 9 Execute "" 0 +F 6 Close P "" +F 4 Flush +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" +F 4 Sync +B 4 CloseComplete +B 5 ReadyForQuery I +F 16 Parse "" "SELECT 1" 0 +F 12 Bind "" "" 0 0 0 +F 6 Describe P "" +F 9 Execute "" 0 +F 6 Close P "" +F 4 Flush +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" +F 16 Parse "" "SELECT 2" 0 +F 12 Bind "" "" 0 0 0 +F 6 Describe P "" +F 9 Execute "" 0 +F 6 Close P "" +F 4 Flush +B 4 CloseComplete +B 4 ParseComplete +B 4 BindComplete +B 33 RowDescription 1 "?column?" NNNN 0 NNNN 4 -1 0 +B 11 DataRow 1 1 '2' +B 13 CommandComplete "SELECT 1" F 4 Terminate -- 2.30.2 --igg5o36yau2li7um Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v5-0002-Use-Test-Differences-if-available.patch" ^ permalink raw reply [nested|flat] 7+ messages in thread
* Re: Use streaming read API in ANALYZE @ 2024-09-04 15:36 Robert Haas <[email protected]> 0 siblings, 1 reply; 7+ messages in thread From: Robert Haas @ 2024-09-04 15:36 UTC (permalink / raw) To: Thomas Munro <[email protected]>; +Cc: Mats Kindahl <[email protected]>; Melanie Plageman <[email protected]>; Nazir Bilal Yavuz <[email protected]>; Andres Freund <[email protected]>; Heikki Linnakangas <[email protected]>; PostgreSQL Hackers <[email protected]>; [email protected] On Wed, Sep 4, 2024 at 6:38 AM Thomas Munro <[email protected]> wrote: > Thanks for the explanation. I think we should revert it. IMHO it was > a nice clean example of a streaming transformation, but unfortunately > it transformed an API that nobody liked in the first place, and broke > some weird and wonderful workarounds. Let's try again in 18. The problem I have with this is that we just released RC1. I suppose if we have to make this change it's better to do it sooner than later, but are we sure we want to whack this around this close to final release? -- Robert Haas EDB: http://www.enterprisedb.com ^ permalink raw reply [nested|flat] 7+ messages in thread
* Re: Use streaming read API in ANALYZE @ 2024-09-04 23:34 Thomas Munro <[email protected]> parent: Robert Haas <[email protected]> 0 siblings, 1 reply; 7+ messages in thread From: Thomas Munro @ 2024-09-04 23:34 UTC (permalink / raw) To: Robert Haas <[email protected]>; +Cc: Mats Kindahl <[email protected]>; Melanie Plageman <[email protected]>; Nazir Bilal Yavuz <[email protected]>; Andres Freund <[email protected]>; Heikki Linnakangas <[email protected]>; PostgreSQL Hackers <[email protected]>; [email protected] On Thu, Sep 5, 2024 at 3:36 AM Robert Haas <[email protected]> wrote: > On Wed, Sep 4, 2024 at 6:38 AM Thomas Munro <[email protected]> wrote: > > Thanks for the explanation. I think we should revert it. IMHO it was > > a nice clean example of a streaming transformation, but unfortunately > > it transformed an API that nobody liked in the first place, and broke > > some weird and wonderful workarounds. Let's try again in 18. > > The problem I have with this is that we just released RC1. I suppose > if we have to make this change it's better to do it sooner than later, > but are we sure we want to whack this around this close to final > release? I hear you. But I definitely don't want to (and likely can't at this point) make any of the other proposed changes, and I also don't want to break Timescale. That seems to leave only one option: go back to the v16 API for RC2, and hope that the ongoing table AM discussions for v18 (CF #4866) will fix all the problems for the people whose TAMs don't quack like a "heap", and the people whose TAMs do and who would not like to duplicate the code, and the people who want streaming I/O. ^ permalink raw reply [nested|flat] 7+ messages in thread
* Re: Use streaming read API in ANALYZE @ 2024-09-05 06:45 Mats Kindahl <[email protected]> parent: Thomas Munro <[email protected]> 0 siblings, 1 reply; 7+ messages in thread From: Mats Kindahl @ 2024-09-05 06:45 UTC (permalink / raw) To: Thomas Munro <[email protected]>; +Cc: Robert Haas <[email protected]>; Melanie Plageman <[email protected]>; Nazir Bilal Yavuz <[email protected]>; Andres Freund <[email protected]>; Heikki Linnakangas <[email protected]>; PostgreSQL Hackers <[email protected]>; [email protected] On Thu, Sep 5, 2024 at 1:34 AM Thomas Munro <[email protected]> wrote: > On Thu, Sep 5, 2024 at 3:36 AM Robert Haas <[email protected]> wrote: > > On Wed, Sep 4, 2024 at 6:38 AM Thomas Munro <[email protected]> > wrote: > > > Thanks for the explanation. I think we should revert it. IMHO it was > > > a nice clean example of a streaming transformation, but unfortunately > > > it transformed an API that nobody liked in the first place, and broke > > > some weird and wonderful workarounds. Let's try again in 18. > > > > The problem I have with this is that we just released RC1. I suppose > > if we have to make this change it's better to do it sooner than later, > > but are we sure we want to whack this around this close to final > > release? > > I hear you. But I definitely don't want to (and likely can't at this > point) make any of the other proposed changes, and I also don't want > to break Timescale. That seems to leave only one option: go back to > the v16 API for RC2, and hope that the ongoing table AM discussions > for v18 (CF #4866) will fix all the problems for the people whose TAMs > don't quack like a "heap", and the people whose TAMs do and who would > not like to duplicate the code, and the people who want streaming I/O. > Forgive me for asking, but I am not entirely sure why the ReadStream struct is opaque. The usual reasons are: - You want to provide an ABI to allow extensions to work with new major versions without re-compiling. Right now it is necessary to recompile extensions anyway, this does not seem to apply. (Because there are a lot of other changes that you need when switching versions because of the lack of a stable ABI for other parts of the code. However, it might be that the goal is to support it eventually, and then it would make sense to start making structs opaque.) - You want to ensure that you can make modifications *inside* a major version without breaking ABIs and requiring a re-compile. In this case, you could still follow safe practice of adding new fields last, not relying on the size of the struct for anything (e.g., no arrays of these structures, just pointers to them), etc. However, if you want to be *very* safe and support very drastic changes inside a major version, it needs to be opaque, so this could be the reason. Is it either of these reasons, or is there another reason? Making the ReadStream API non-opaque (that is, moving the definition to the header file) would at least solve our problem (unless I am mistaken). However, I am ignorant about long-term plans which might affect this, so there might be a good reason to revert it for reasons I am not aware of. -- Best wishes, Mats Kindahl, Timescale ^ permalink raw reply [nested|flat] 7+ messages in thread
* Re: Use streaming read API in ANALYZE @ 2024-09-05 09:12 Thomas Munro <[email protected]> parent: Mats Kindahl <[email protected]> 0 siblings, 1 reply; 7+ messages in thread From: Thomas Munro @ 2024-09-05 09:12 UTC (permalink / raw) To: Mats Kindahl <[email protected]>; +Cc: Robert Haas <[email protected]>; Melanie Plageman <[email protected]>; Nazir Bilal Yavuz <[email protected]>; Andres Freund <[email protected]>; Heikki Linnakangas <[email protected]>; PostgreSQL Hackers <[email protected]>; [email protected] On Thu, Sep 5, 2024 at 6:45 PM Mats Kindahl <[email protected]> wrote: > Forgive me for asking, but I am not entirely sure why the ReadStream struct is opaque. The usual reasons are: > > You want to provide an ABI to allow extensions to work with new major versions without re-compiling. Right now it is necessary to recompile extensions anyway, this does not seem to apply. (Because there are a lot of other changes that you need when switching versions because of the lack of a stable ABI for other parts of the code. However, it might be that the goal is to support it eventually, and then it would make sense to start making structs opaque.) > You want to ensure that you can make modifications inside a major version without breaking ABIs and requiring a re-compile. In this case, you could still follow safe practice of adding new fields last, not relying on the size of the struct for anything (e.g., no arrays of these structures, just pointers to them), etc. However, if you want to be very safe and support very drastic changes inside a major version, it needs to be opaque, so this could be the reason. > > Is it either of these reasons, or is there another reason? > > Making the ReadStream API non-opaque (that is, moving the definition to the header file) would at least solve our problem (unless I am mistaken). However, I am ignorant about long-term plans which might affect this, so there might be a good reason to revert it for reasons I am not aware of. The second thing. Also there are very active plans[1] to change the internal design of ReadStream in 18, since the goal is to drive true asynchronous I/O, and the idea of ReadStream was to create a simple API to let many consumers start using it, so that we can drive efficient modern system interfaces below that API, so having people depending on how it works would not be great. But let's talk about how that would actually look, for example if we exposed the struct or you took a photocopy of it... I think your idea must be something like: if you could access struct ReadStream's internals, you could replace stream->callback with an interceptor callback, and if the BlockSampler had been given the fake N + M relation size, the interceptor could overwrite stream->ios[next_io_index].op.smgr and return x - N if the intercepted callback returned x >= N. (Small detail: need to check stream->fast_path and use 0 instead or something like that, but maybe we could change that.) One minor problem that jumps out is that read_stream.c could inappropriately merge blocks from the two relations into one I/O. Hmm, I guess you'd have to teach the interceptor not to allow that: if switching between the two relation, and if the block number would coincide with stream->pending_read_blocknum + stream->pending_read_nblocks, it would need to pick a new block instead (interfering with the block sampling algorithm, but only very rarely). Is this what you had in mind, or something else? (BTW I have a patch to teach read_stream.c about multi-smgr-relation streams, by adding a different constructor with a different callback that returns smgr, fork, block instead of just the block, but it didn't make it into 17.) [1] https://www.postgresql.org/message-id/flat/uvrtrknj4kdytuboidbhwclo4gxhswwcpgadptsjvjqcluzmah@brqs62... ^ permalink raw reply [nested|flat] 7+ messages in thread
* Re: Use streaming read API in ANALYZE @ 2024-09-10 17:07 Mats Kindahl <[email protected]> parent: Thomas Munro <[email protected]> 0 siblings, 0 replies; 7+ messages in thread From: Mats Kindahl @ 2024-09-10 17:07 UTC (permalink / raw) To: Thomas Munro <[email protected]>; +Cc: Robert Haas <[email protected]>; Melanie Plageman <[email protected]>; Nazir Bilal Yavuz <[email protected]>; Andres Freund <[email protected]>; Heikki Linnakangas <[email protected]>; PostgreSQL Hackers <[email protected]>; [email protected] On Thu, Sep 5, 2024 at 11:12 AM Thomas Munro <[email protected]> wrote: > On Thu, Sep 5, 2024 at 6:45 PM Mats Kindahl <[email protected]> wrote: > > Forgive me for asking, but I am not entirely sure why the ReadStream > struct is opaque. The usual reasons are: > > > > You want to provide an ABI to allow extensions to work with new major > versions without re-compiling. Right now it is necessary to recompile > extensions anyway, this does not seem to apply. (Because there are a lot of > other changes that you need when switching versions because of the lack of > a stable ABI for other parts of the code. However, it might be that the > goal is to support it eventually, and then it would make sense to start > making structs opaque.) > > You want to ensure that you can make modifications inside a major > version without breaking ABIs and requiring a re-compile. In this case, you > could still follow safe practice of adding new fields last, not relying on > the size of the struct for anything (e.g., no arrays of these structures, > just pointers to them), etc. However, if you want to be very safe and > support very drastic changes inside a major version, it needs to be opaque, > so this could be the reason. > > > > Is it either of these reasons, or is there another reason? > > > > Making the ReadStream API non-opaque (that is, moving the definition to > the header file) would at least solve our problem (unless I am mistaken). > However, I am ignorant about long-term plans which might affect this, so > there might be a good reason to revert it for reasons I am not aware of. > > The second thing. Also there are very active plans[1] to change the > internal design of ReadStream in 18, since the goal is to drive true > asynchronous I/O, and the idea of ReadStream was to create a simple > API to let many consumers start using it, so that we can drive > efficient modern system interfaces below that API, so having people > depending on how it works would not be great. > That is understandable, since you usually do not want to have to re-compile the extension for different minor versions. However, it would be a rare case with extensions that are meddling with this, so might not turn out to be a big problem in reality, as long as it is very clear to all involved that this might change and that you make an effort to avoid binary incompatibility by removing or changing types for fields. > But let's talk about how that would actually look, for example if we > exposed the struct or you took a photocopy of it... I think your idea > must be something like: if you could access struct ReadStream's > internals, you could replace stream->callback with an interceptor > callback, and if the BlockSampler had been given the fake N + M > relation size, the interceptor could overwrite > stream->ios[next_io_index].op.smgr and return x - N if the intercepted > callback returned x >= N. (Small detail: need to check > stream->fast_path and use 0 instead or something like that, but maybe > we could change that.) Yes, this is what I had in mind, but I did not dig too deeply into the code. > One minor problem that jumps out is that > read_stream.c could inappropriately merge blocks from the two > relations into one I/O. Hmm, I guess you'd have to teach the > interceptor not to allow that: if switching between the two relation, > and if the block number would coincide with > stream->pending_read_blocknum + stream->pending_read_nblocks, it would > need to pick a new block instead (interfering with the block sampling > algorithm, but only very rarely). Is this what you had in mind, or > something else? > Hmmm... I didn't look too closely at this. Since the block number comes from the callback, I guess we could make sure to have a "padding" block between the regions so that we "break" any suite of blocks, which I think is what you mean with "teach the interceptor not to allow that", but I would have to write a patch to make sure. > > (BTW I have a patch to teach read_stream.c about multi-smgr-relation > streams, by adding a different constructor with a different callback > that returns smgr, fork, block instead of just the block, but it > didn't make it into 17.) > Without having looked at the patch, this sounds like the correct way to do it. > > [1] > https://www.postgresql.org/message-id/flat/uvrtrknj4kdytuboidbhwclo4gxhswwcpgadptsjvjqcluzmah@brqs62... > -- Best wishes, Mats Kindahl, Timescale ^ permalink raw reply [nested|flat] 7+ messages in thread
end of thread, other threads:[~2024-09-10 17:07 UTC | newest] Thread overview: 7+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2022-06-15 17:56 [PATCH v6] Avoid going IDLE in pipeline mode Alvaro Herrera <[email protected]> 2022-06-15 17:56 [PATCH v5 1/2] Avoid going IDLE in pipeline mode Alvaro Herrera <[email protected]> 2024-09-04 15:36 Re: Use streaming read API in ANALYZE Robert Haas <[email protected]> 2024-09-04 23:34 ` Re: Use streaming read API in ANALYZE Thomas Munro <[email protected]> 2024-09-05 06:45 ` Re: Use streaming read API in ANALYZE Mats Kindahl <[email protected]> 2024-09-05 09:12 ` Re: Use streaming read API in ANALYZE Thomas Munro <[email protected]> 2024-09-10 17:07 ` Re: Use streaming read API in ANALYZE Mats Kindahl <[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