From f7787e2e4fc3428353acc213c12f6604b3784a9a Mon Sep 17 00:00:00 2001 From: Jelte Fennema-Nio Date: Fri, 26 Jan 2024 17:01:00 +0100 Subject: [PATCH v31 4/5] libpq: Add encrypted and non-blocking versions of PQcancel The existing PQcancel API is using blocking IO. This makes PQcancel impossible to use in an event loop based codebase, without blocking the event loop until the call returns. PQcancelConn can now be used instead, to have a non-blocking way of sending cancel requests. It also doesn't encrypt the connection over which the cancel request is sent, even when the original connection required encryption. This patch adds a bunch of new functions which, together, allow users to send cancel requests in an encrypted and performant way. The primary new functions are PQcancelBlocking and PQcancelStart (for blocking and non-blocking requests respectively). These functions reuse the normal connection establishement code, so that they can apply the same connection options such sslmode and gssencmode that the original connection used. --- doc/src/sgml/libpq.sgml | 354 ++++++++++++++++-- src/interfaces/libpq/exports.txt | 9 + src/interfaces/libpq/fe-cancel.c | 282 ++++++++++++++ src/interfaces/libpq/fe-connect.c | 130 ++++++- src/interfaces/libpq/libpq-fe.h | 31 +- src/interfaces/libpq/libpq-int.h | 10 + .../modules/libpq_pipeline/libpq_pipeline.c | 263 ++++++++++++- src/tools/pgindent/typedefs.list | 1 + 8 files changed, 1035 insertions(+), 45 deletions(-) diff --git a/doc/src/sgml/libpq.sgml b/doc/src/sgml/libpq.sgml index 1d8998efb2a..1613fcc7bb4 100644 --- a/doc/src/sgml/libpq.sgml +++ b/doc/src/sgml/libpq.sgml @@ -265,7 +265,7 @@ PGconn *PQsetdb(char *pghost, PQconnectStartParamsPQconnectStartParams PQconnectStartPQconnectStart - PQconnectPollPQconnectPoll + PQconnectPollPQconnectPoll nonblocking connection @@ -5278,7 +5278,7 @@ int PQisBusy(PGconn *conn); / can also attempt to cancel a command that is still being processed by the server; see . But regardless of - the return value of , the application + the return value of , the application must continue with the normal result-reading sequence using . A successful cancellation will simply cause the command to terminate sooner than it would have @@ -6025,10 +6025,295 @@ int PQsetSingleRowMode(PGconn *conn); SQL command - - A client application can request cancellation of a command that is - still being processed by the server, using the functions described in - this section. + + Functions for Sending Cancel Requests + + + PQcancelConnPQcancelConn + + + + Prepares a connection over which a cancel request can be sent. + +PGcancelConn *PQcancelConn(PGconn *conn); + + + + + creates a + PGcancelConnPGcancelConn + object, but it won't instantly start sending a cancel request over this + connection. A cancel request can be sent over this connection in a + blocking manner using and in a + non-blocking manner using . + The return value can be passed to + to check if the PGcancelConn object was + created successfully. The PGcancelConn object + is an opaque structure that is not meant to be accessed directly by the + application. This PGcancelConn object can be + used to cancel the query that's running on the original connection in a + thread-safe way. + + + + If the original connection is encrypted (using TLS or GSS), then the + connection for the cancel request is encrypted in the same way. Any + connection options that are only used during authentication or after + authentication of the client are ignored though, because cancellation + requests do not require authentication and the connection is closed right + after the cancellation request is submitted. + + + + Note that when PQcancelConn returns a non-null + pointer, you must call when you + are finished with it, in order to dispose of the structure and any + associated memory blocks. This must be done even if the cancel request + failed or was abandoned. + + + + + + PQcancelBlockingPQcancelBlocking + + + + Requests that the server abandons processing of the current command in a blocking manner. + +int PQcancelBlocking(PGcancelConn *cancelConn); + + + + + The request is made over the given PGcancelConn, + which needs to be created with . + The return value of + is 1 if the cancel request was successfully + dispatched and 0 if not. If it was unsuccessful, the error message can be + retrieved using . + + + + Successful dispatch of the cancellation is no guarantee that the request + will have any effect, however. If the cancellation is effective, the + command being canceled will terminate early and return an error result. + If the cancellation fails (say, because the server was already done + processing the command), then there will be no visible result at all. + + + + + + + PQcancelStartPQcancelStart + PQcancelPollPQcancelPoll + + + + Requests that the server abandons processing of the current command in a non-blocking manner. + +int PQcancelStart(PGcancelConn *cancelConn); + +PostgresPollingStatusType PQcancelPoll(PGcancelConn *cancelConn); + + + + + The request is made over the given PGcancelConn, + which needs to be created with . + The return value of + is 1 if the cancellation request could be started and 0 if not. + If it was unsuccessful, the error message can be + retrieved using . + + + + If PQcancelStart succeeds, the next stage + is to poll libpq so that it can proceed with + the connection sequence. + Use to obtain the descriptor of the + socket underlying the database connection. + (Caution: do not assume that the socket remains the same + across PQconnectPoll calls.) + Loop thus: If PQcancelPoll(cancelConn) last returned + PGRES_POLLING_READING, wait until the socket is ready to + read (as indicated by select(), poll(), or + similar system function). + Then call PQcancelPoll(cancelConn) again. + Conversely, if PQcancelPoll(cancelConn) last returned + PGRES_POLLING_WRITING, wait until the socket is ready + to write, then call PQcancelPoll(cancelConn) again. + On the first iteration, i.e., if you have yet to call + PQcancelPoll(cancelConn), behave as if it last returned + PGRES_POLLING_WRITING. Continue this loop until + PQcancelPoll(cancelConn) returns + PGRES_POLLING_FAILED, indicating the connection procedure + has failed, or PGRES_POLLING_OK, indicating the connection + has been successfully made. + + + + At any time during connection, the status of the connection can be + checked by calling . If this call returns CONNECTION_BAD, then the + connection procedure has failed; if the call returns CONNECTION_OK, then the + connection is ready. Both of these states are equally detectable + from the return value of PQcancelPoll, described above. Other states might also occur + during (and only during) an asynchronous connection procedure. These + indicate the current stage of the connection procedure and might be useful + to provide feedback to the user for example. These statuses are: + + + + + + + PQcancelStatusPQcancelStatus + + + + Returns the status of the cancel connection. + +ConnStatusType PQcancelStatus(const PGcancelConn *cancelConn); + + + + This function is almost the same as , only + for a PGcancelConn instead of a + PGconn. In addition to all the statuses that + can return for a + PGconn, PQcancelStatus + can return one additional status: + + + + CONNECTION_ALLOCATED + + + Waiting for a call to or + , to actually open the + socket. This is the connection state right after + calling + or . No connection to the + server has been initiated yet at this point. To actually start + sending the cancel request use or + . + + + + + + + + One final note about the returned statuses is that + CONNECTION_OK has a slightly different meaning for a + PGcancelConn than what it has for a + PGconn. When + returns CONNECTION_OK for a PGcancelConn + it means that that the dispatch of the cancel request has completed (although + this is no promise that the query was actually canceled) and that the + cancel connection is now closed, while a CONNECTION_OK + result for PGconn means that queries can be + sent over the connection. + + + + + + + PQcancelSocketPQcancelSocket + + + + Obtains the file descriptor number of the cancel connection socket to + the server. A valid descriptor will be greater than or equal + to 0; a result of -1 indicates that no server connection is + currently open. This might change as a result of calling all of the + functions in this section on the (except for + and + PQcancelSocket itself). + +int PQcancelSocket(const PGcancelConn *cancelConn); + + + + + + + PQcancelErrorMessagePQcancelErrorMessage + + + + A version of that can be used for + cancellation connections. If + returns CONNECTION_BAD, then this function can be + called on the PGcancelConn to retrieve the + error message. + +char *PQcancelErrorMessage(const PGcancelConn *cancelconn); + + + + + + + PQcancelFinishPQcancelFinish + + + Closes the cancel connection (if it did not finish sending the cancel + request yet). Also frees memory used by the PGcancelConn + object. + +void PQcancelFinish(PGcancelConn *cancelConn); + + + + + Note that even if the cancel attempt fails (as + indicated by ), the application should call + to free the memory used by the PGcancelConn object. + The PGcancelConn pointer must not be used again after + has been called. + + + + + + PQcancelResetPQcancelReset + + + Resets the PGcancelConn so it can be reused for a new + cancel connection. + +void PQcancelReset(PGcancelConn *cancelConn); + + + + + If the PGcancelConn is currently used to send a cancel + request, then this connection is closed. It will then prepare the + PGcancelConn object such that it can be used to send a + new cancel request. This can be used to create one PGcancelConn + for a PGconn and reuse that multiple times throughout + the lifetime of the original PGconn. + + + + + + + + Obsolete Functions for Sending Cancel Requests + + + These functions represent older methods of sending cancel requests. + Although they still work, they are deprecated due to not sending the cancel + requests in an encrypted manner, even when the original connection + specified sslmode or gssencmode to + require encryption. Thus these older methods are heavily discouraged from + being used in new code, and it is recommended to change existing code to + use the new functions instead. + @@ -6037,7 +6322,7 @@ int PQsetSingleRowMode(PGconn *conn); Creates a data structure containing the information needed to cancel - a command issued through a particular database connection. + a command using . PGcancel *PQgetCancel(PGconn *conn); @@ -6079,36 +6364,37 @@ void PQfreeCancel(PGcancel *cancel); - Requests that the server abandon processing of the current command. - + is a deprecated and insecure + variant of , but one that can be + used safely from within a signal handler. int PQcancel(PGcancel *cancel, char *errbuf, int errbufsize); - The return value is 1 if the cancel request was successfully - dispatched and 0 if not. If not, errbuf is filled - with an explanatory error message. errbuf - must be a char array of size errbufsize (the - recommended size is 256 bytes). + only exists because of backwards + compatibility reasons. should be + used instead. The only benefit that has + is that it can be safely invoked from a signal handler, if the + errbuf is a local variable in the signal handler. + However, this is generally not considered a big enough benefit to be + worth the security issues that this function has. - Successful dispatch is no guarantee that the request will have - any effect, however. If the cancellation is effective, the current - command will terminate early and return an error result. If the - cancellation fails (say, because the server was already done - processing the command), then there will be no visible result at - all. + The PGcancel object is read-only as far as + is concerned, so it can also be invoked + from a thread that is separate from the one manipulating the + PGconn object. - can safely be invoked from a signal - handler, if the errbuf is a local variable in the - signal handler. The PGcancel object is read-only - as far as is concerned, so it can - also be invoked from a thread that is separate from the one - manipulating the PGconn object. + The return value of + is 1 if the cancel request was successfully + dispatched and 0 if not. If not, errbuf is filled + with an explanatory error message. errbuf + must be a char array of size errbufsize (the + recommended size is 256 bytes). @@ -6120,13 +6406,21 @@ int PQcancel(PGcancel *cancel, char *errbuf, int errbufsize); - is a deprecated variant of - . + is a deprecated and insecure + variant of . int PQrequestCancel(PGconn *conn); + + only exists because of backwards + compatibility reasons. should be + used instead. There is no benefit to using + over + . + + Requests that the server abandon processing of the current command. It operates directly on the @@ -6141,7 +6435,7 @@ int PQrequestCancel(PGconn *conn); - + @@ -9353,7 +9647,7 @@ int PQisthreadsafe(); The deprecated functions and are not thread-safe and should not be used in multithread programs. - can be replaced by . + can be replaced by . can be replaced by . diff --git a/src/interfaces/libpq/exports.txt b/src/interfaces/libpq/exports.txt index 088592deb16..0ae814490e6 100644 --- a/src/interfaces/libpq/exports.txt +++ b/src/interfaces/libpq/exports.txt @@ -193,3 +193,12 @@ PQsendClosePrepared 190 PQsendClosePortal 191 PQchangePassword 192 PQsendPipelineSync 193 +PQcancelBlocking 194 +PQcancelStart 195 +PQcancelConn 196 +PQcancelPoll 197 +PQcancelStatus 198 +PQcancelSocket 199 +PQcancelErrorMessage 200 +PQcancelReset 201 +PQcancelFinish 202 diff --git a/src/interfaces/libpq/fe-cancel.c b/src/interfaces/libpq/fe-cancel.c index 51f8d8a78c4..e66b8819ee0 100644 --- a/src/interfaces/libpq/fe-cancel.c +++ b/src/interfaces/libpq/fe-cancel.c @@ -21,6 +21,288 @@ #include "libpq-int.h" #include "port/pg_bswap.h" + +/* + * PQcancelConn + * + * Asynchronously cancel a query on the given connection. This requires polling + * the returned PGcancelConn to actually complete the cancellation of the + * query. + */ +PGcancelConn * +PQcancelConn(PGconn *conn) +{ + PGconn *cancelConn = pqMakeEmptyPGconn(); + pg_conn_host originalHost; + + if (cancelConn == NULL) + return NULL; + + /* Check we have an open connection */ + if (!conn) + { + libpq_append_conn_error(cancelConn, "passed connection was NULL"); + return (PGcancelConn *) cancelConn; + } + + if (conn->sock == PGINVALID_SOCKET) + { + libpq_append_conn_error(cancelConn, "passed connection is not open"); + return (PGcancelConn *) cancelConn; + } + + /* + * Indicate that this connection is used to send a cancellation + */ + cancelConn->cancelRequest = true; + + if (!pqCopyPGconn(conn, cancelConn)) + return (PGcancelConn *) cancelConn; + + /* + * Compute derived options + */ + if (!pqConnectOptions2(cancelConn)) + return (PGcancelConn *) cancelConn; + + /* + * Copy cancellation token data from the original connnection + */ + cancelConn->be_pid = conn->be_pid; + cancelConn->be_key = conn->be_key; + + /* + * Cancel requests should not iterate over all possible hosts. The request + * needs to be sent to the exact host and address that the original + * connection used. So we manually create the host and address arrays with + * a single element after freeing the host array that we generated from + * the connection options. + */ + pqReleaseConnHosts(cancelConn); + cancelConn->nconnhost = 1; + cancelConn->naddr = 1; + + cancelConn->connhost = calloc(cancelConn->nconnhost, sizeof(pg_conn_host)); + if (!cancelConn->connhost) + goto oom_error; + + originalHost = conn->connhost[conn->whichhost]; + if (originalHost.host) + { + cancelConn->connhost[0].host = strdup(originalHost.host); + if (!cancelConn->connhost[0].host) + goto oom_error; + } + if (originalHost.hostaddr) + { + cancelConn->connhost[0].hostaddr = strdup(originalHost.hostaddr); + if (!cancelConn->connhost[0].hostaddr) + goto oom_error; + } + if (originalHost.port) + { + cancelConn->connhost[0].port = strdup(originalHost.port); + if (!cancelConn->connhost[0].port) + goto oom_error; + } + if (originalHost.password) + { + cancelConn->connhost[0].password = strdup(originalHost.password); + if (!cancelConn->connhost[0].password) + goto oom_error; + } + + cancelConn->addr = calloc(cancelConn->naddr, sizeof(AddrInfo)); + if (!cancelConn->connhost) + goto oom_error; + + cancelConn->addr[0].addr = conn->raddr; + cancelConn->addr[0].family = conn->raddr.addr.ss_family; + + cancelConn->status = CONNECTION_ALLOCATED; + return (PGcancelConn *) cancelConn; + +oom_error: + conn->status = CONNECTION_BAD; + libpq_append_conn_error(cancelConn, "out of memory"); + return (PGcancelConn *) cancelConn; +} + + +/* + * PQcancelBlocking + * + * Send a cancellation request in a blocking fashion. + * Returns 1 if successful 0 if not. + */ +int +PQcancelBlocking(PGcancelConn *cancelConn) +{ + if (!PQcancelStart(cancelConn)) + return 0; + return pqConnectDBComplete(&cancelConn->conn); +} + +/* + * PQcancelStart + * + * Starts sending a cancellation request in a blocking fashion. Returns + * 1 if successful 0 if not. + */ +int +PQcancelStart(PGcancelConn *cancelConn) +{ + if (!cancelConn || cancelConn->conn.status == CONNECTION_BAD) + return 0; + + if (cancelConn->conn.status != CONNECTION_ALLOCATED) + { + libpq_append_conn_error(&cancelConn->conn, + "cancel request is already being sent on this connection"); + cancelConn->conn.status = CONNECTION_BAD; + return 0; + } + + return pqConnectDBStart(&cancelConn->conn); +} + +/* + * PQcancelPoll + * + * Poll a cancel connection. For usage details see PQconnectPoll. + */ +PostgresPollingStatusType +PQcancelPoll(PGcancelConn *cancelConn) +{ + PGconn *conn = (PGconn *) cancelConn; + int n; + + /* + * We leave most of the connection establishement to PQconnectPoll, since + * it's very similar to normal connection establishment. But once we get + * to the CONNECTION_AWAITING_RESPONSE we need to start doing our own + * thing. + */ + if (conn->status != CONNECTION_AWAITING_RESPONSE) + { + return PQconnectPoll(conn); + } + + /* + * At this point we are waiting on the server to close the connection, + * which is its way of communicating that the cancel has been handled. + */ + + n = pqReadData(conn); + + if (n == 0) + return PGRES_POLLING_READING; + +#ifndef WIN32 + + /* + * If we receive an error report it, but only if errno is non-zero. + * Otherwise we assume it's an EOF, which is what we expect from the + * server. + * + * We skip this for Windows, because Windows is a bit special in its EOF + * behaviour for TCP. Sometimes it will error with an ECONNRESET when + * there is a clean connection closure. See these threads for details: + * https://www.postgresql.org/message-id/flat/90b34057-4176-7bb0-0dbb-9822a5f6425b%40greiz-reinsdorf.de + * + * https://www.postgresql.org/message-id/flat/CA%2BhUKG%2BOeoETZQ%3DQw5Ub5h3tmwQhBmDA%3DnuNO3KG%3DzWfUypFAw%40mail.gmail.com + * + * PQcancel ignores such errors and reports success for the cancellation + * anyway, so even if this is not always correct we do the same here. + */ + if (n < 0 && errno != 0) + { + conn->status = CONNECTION_BAD; + return PGRES_POLLING_FAILED; + } +#endif + + /* + * We don't expect any data, only connection closure. So if we strangely + * do receive some data we consider that an error. + */ + if (n > 0) + { + libpq_append_conn_error(conn, "received unexpected response from server"); + conn->status = CONNECTION_BAD; + return PGRES_POLLING_FAILED; + } + + /* + * Getting here means that we received an EOF, which is what we were + * expecting -- the cancel request has completed. + */ + cancelConn->conn.status = CONNECTION_OK; + resetPQExpBuffer(&conn->errorMessage); + return PGRES_POLLING_OK; +} + +/* + * PQcancelStatus + * + * Get the status of a cancel connection. + */ +ConnStatusType +PQcancelStatus(const PGcancelConn *cancelConn) +{ + return PQstatus((const PGconn *) cancelConn); +} + +/* + * PQcancelSocket + * + * Get the socket of the cancel connection. + */ +int +PQcancelSocket(const PGcancelConn *cancelConn) +{ + return PQsocket((const PGconn *) cancelConn); +} + +/* + * PQcancelErrorMessage + * + * Get the socket of the cancel connection. + */ +char * +PQcancelErrorMessage(const PGcancelConn *cancelConn) +{ + return PQerrorMessage((const PGconn *) cancelConn); +} + +/* + * PQcancelReset + * + * Resets the cancel connection, so it can be reused to send a new cancel + * request. + */ +void +PQcancelReset(PGcancelConn *cancelConn) +{ + pqClosePGconn((PGconn *) cancelConn); + cancelConn->conn.status = CONNECTION_ALLOCATED; + cancelConn->conn.whichhost = 0; + cancelConn->conn.whichaddr = 0; + cancelConn->conn.try_next_host = false; + cancelConn->conn.try_next_addr = false; +} + +/* + * PQcancelFinish + * + * Closes and frees the cancel connection. + */ +void +PQcancelFinish(PGcancelConn *cancelConn) +{ + PQfinish((PGconn *) cancelConn); +} + /* * PQgetCancel: get a PGcancel structure corresponding to a connection. * diff --git a/src/interfaces/libpq/fe-connect.c b/src/interfaces/libpq/fe-connect.c index d4e10a0c4f3..b4e7394314f 100644 --- a/src/interfaces/libpq/fe-connect.c +++ b/src/interfaces/libpq/fe-connect.c @@ -616,8 +616,17 @@ pqDropServerData(PGconn *conn) conn->write_failed = false; free(conn->write_err_msg); conn->write_err_msg = NULL; - conn->be_pid = 0; - conn->be_key = 0; + + /* + * Cancel connections need to retain their be_pid and be_key across + * PQcancelReset invocations, otherwise they would not have access to the + * secret token of the connection they are supposed to cancel. + */ + if (!conn->cancelRequest) + { + conn->be_pid = 0; + conn->be_key = 0; + } } @@ -923,6 +932,45 @@ fillPGconn(PGconn *conn, PQconninfoOption *connOptions) return true; } +/* + * Copy over option values from srcConn to dstConn + * + * Don't put anything cute here --- intelligence should be in + * connectOptions2 ... + * + * Returns true on success. On failure, returns false and sets error message of + * dstConn. + */ +bool +pqCopyPGconn(PGconn *srcConn, PGconn *dstConn) +{ + const internalPQconninfoOption *option; + + /* copy over connection options */ + for (option = PQconninfoOptions; option->keyword; option++) + { + if (option->connofs >= 0) + { + const char **tmp = (const char **) ((char *) srcConn + option->connofs); + + if (*tmp) + { + char **dstConnmember = (char **) ((char *) dstConn + option->connofs); + + if (*dstConnmember) + free(*dstConnmember); + *dstConnmember = strdup(*tmp); + if (*dstConnmember == NULL) + { + libpq_append_conn_error(dstConn, "out of memory"); + return false; + } + } + } + } + return true; +} + /* * connectOptions1 * @@ -2308,10 +2356,18 @@ pqConnectDBStart(PGconn *conn) * Set up to try to connect to the first host. (Setting whichhost = -1 is * a bit of a cheat, but PQconnectPoll will advance it to 0 before * anything else looks at it.) + * + * Cancel requests are special though, they should only try one host and + * address, and these fields have already been set up in PQcancelConn, so + * leave these fields alone for cancel requests. */ - conn->whichhost = -1; - conn->try_next_addr = false; - conn->try_next_host = true; + if (!conn->cancelRequest) + { + conn->whichhost = -1; + conn->try_next_host = true; + conn->try_next_addr = false; + } + conn->status = CONNECTION_NEEDED; /* Also reset the target_server_type state if needed */ @@ -2453,7 +2509,10 @@ pqConnectDBComplete(PGconn *conn) /* * Now try to advance the state machine. */ - flag = PQconnectPoll(conn); + if (conn->cancelRequest) + flag = PQcancelPoll((PGcancelConn *) conn); + else + flag = PQconnectPoll(conn); } } @@ -2578,13 +2637,17 @@ keep_going: /* We will come back to here until there is * Oops, no more hosts. * * If we are trying to connect in "prefer-standby" mode, then drop - * the standby requirement and start over. + * the standby requirement and start over. Don't do this for + * cancel requests though, since we are certain the list of + * servers won't change as the target_server_type option is not + * applicable to those connections. * * Otherwise, an appropriate error message is already set up, so * we just need to set the right status. */ if (conn->target_server_type == SERVER_TYPE_PREFER_STANDBY && - conn->nconnhost > 0) + conn->nconnhost > 0 && + !conn->cancelRequest) { conn->target_server_type = SERVER_TYPE_PREFER_STANDBY_PASS2; conn->whichhost = 0; @@ -3226,6 +3289,29 @@ keep_going: /* We will come back to here until there is } #endif /* USE_SSL */ + /* + * For cancel requests this is as far as we need to go in the + * connection establishment. Now we can actually send our + * cancellation request. + */ + if (conn->cancelRequest) + { + CancelRequestPacket cancelpacket; + + packetlen = sizeof(cancelpacket); + cancelpacket.cancelRequestCode = (MsgType) pg_hton32(CANCEL_REQUEST_CODE); + cancelpacket.backendPID = pg_hton32(conn->be_pid); + cancelpacket.cancelAuthCode = pg_hton32(conn->be_key); + if (pqPacketSend(conn, 0, &cancelpacket, packetlen) != STATUS_OK) + { + libpq_append_conn_error(conn, "could not send cancel packet: %s", + SOCK_STRERROR(SOCK_ERRNO, sebuf, sizeof(sebuf))); + goto error_return; + } + conn->status = CONNECTION_AWAITING_RESPONSE; + return PGRES_POLLING_READING; + } + /* * Build the startup packet. */ @@ -3975,8 +4061,14 @@ keep_going: /* We will come back to here until there is } } - /* We can release the address list now. */ - release_conn_addrinfo(conn); + /* + * For non cancel requests we can release the address list + * now. For cancel requests we never actually resolve + * addresses and instead the addrinfo exists for the lifetime + * of the connection. + */ + if (!conn->cancelRequest) + release_conn_addrinfo(conn); /* * Contents of conn->errorMessage are no longer interesting @@ -4344,6 +4436,7 @@ freePGconn(PGconn *conn) free(conn->events[i].name); } + release_conn_addrinfo(conn); pqReleaseConnHosts(conn); free(conn->client_encoding_initial); @@ -4495,6 +4588,15 @@ release_conn_addrinfo(PGconn *conn) static void sendTerminateConn(PGconn *conn) { + /* + * The Postgres cancellation protocol does not have a notion of a + * Terminate message, so don't send one. + */ + if (conn->cancelRequest) + { + return; + } + /* * Note that the protocol doesn't allow us to send Terminate messages * during the startup phase. @@ -4548,7 +4650,13 @@ pqClosePGconn(PGconn *conn) conn->pipelineStatus = PQ_PIPELINE_OFF; pqClearAsyncResult(conn); /* deallocate result */ pqClearConnErrorState(conn); - release_conn_addrinfo(conn); + + /* + * Since cancel requests never change their addrinfo we don't free it + * here. Otherwise we would have to rebuild it during a PQcancelReset. + */ + if (!conn->cancelRequest) + release_conn_addrinfo(conn); /* Reset all state obtained from server, too */ pqDropServerData(conn); diff --git a/src/interfaces/libpq/libpq-fe.h b/src/interfaces/libpq/libpq-fe.h index defc415fa3f..523ea6535f3 100644 --- a/src/interfaces/libpq/libpq-fe.h +++ b/src/interfaces/libpq/libpq-fe.h @@ -78,7 +78,9 @@ typedef enum CONNECTION_CONSUME, /* Consuming any extra messages. */ CONNECTION_GSS_STARTUP, /* Negotiating GSSAPI. */ CONNECTION_CHECK_TARGET, /* Checking target server properties. */ - CONNECTION_CHECK_STANDBY /* Checking if server is in standby mode. */ + CONNECTION_CHECK_STANDBY, /* Checking if server is in standby mode. */ + CONNECTION_ALLOCATED /* Waiting for connection attempt to be + * started. */ } ConnStatusType; typedef enum @@ -165,6 +167,11 @@ typedef enum */ typedef struct pg_conn PGconn; +/* PGcancelConn encapsulates a cancel connection to the backend. + * The contents of this struct are not supposed to be known to applications. + */ +typedef struct pg_cancel_conn PGcancelConn; + /* PGresult encapsulates the result of a query (or more precisely, of a single * SQL command --- a query string given to PQsendQuery can contain multiple * commands and thus return multiple PGresult objects). @@ -321,16 +328,34 @@ extern PostgresPollingStatusType PQresetPoll(PGconn *conn); /* Synchronous (blocking) */ extern void PQreset(PGconn *conn); +/* Create a PGcancelConn that's used to cancel a query on the given PGconn */ +extern PGcancelConn *PQcancelConn(PGconn *conn); + +/* issue a cancel request in a non-blocking manner */ +extern int PQcancelStart(PGcancelConn *cancelConn); + +/* issue a blocking cancel request */ +extern int PQcancelBlocking(PGcancelConn *cancelConn); + +/* poll a non-blocking cancel request */ +extern PostgresPollingStatusType PQcancelPoll(PGcancelConn *cancelConn); +extern ConnStatusType PQcancelStatus(const PGcancelConn *cancelConn); +extern int PQcancelSocket(const PGcancelConn *cancelConn); +extern char *PQcancelErrorMessage(const PGcancelConn *cancelConn); +extern void PQcancelReset(PGcancelConn *cancelConn); +extern void PQcancelFinish(PGcancelConn *cancelConn); + + /* request a cancel structure */ extern PGcancel *PQgetCancel(PGconn *conn); /* free a cancel structure */ extern void PQfreeCancel(PGcancel *cancel); -/* issue a cancel request */ +/* deprecated version of PQcancelBlocking, but one which is signal-safe */ extern int PQcancel(PGcancel *cancel, char *errbuf, int errbufsize); -/* backwards compatible version of PQcancel; not thread-safe */ +/* deprecated version of PQcancel; not thread-safe */ extern int PQrequestCancel(PGconn *conn); /* Accessor functions for PGconn objects */ diff --git a/src/interfaces/libpq/libpq-int.h b/src/interfaces/libpq/libpq-int.h index 82c18f870d2..1982cd4ded2 100644 --- a/src/interfaces/libpq/libpq-int.h +++ b/src/interfaces/libpq/libpq-int.h @@ -409,6 +409,10 @@ struct pg_conn char *require_auth; /* name of the expected auth method */ char *load_balance_hosts; /* load balance over hosts */ + bool cancelRequest; /* true if this connection is used to send a + * cancel request, instead of being a normal + * connection that's used for queries */ + /* Optional file to write trace info to */ FILE *Pfdebug; int traceFlags; @@ -621,6 +625,11 @@ struct pg_conn PQExpBufferData workBuffer; /* expansible string */ }; +struct pg_cancel_conn +{ + PGconn conn; +}; + /* PGcancel stores all data necessary to cancel a connection. A copy of this * data is required to safely cancel a connection running on a different * thread. @@ -687,6 +696,7 @@ extern void pqClosePGconn(PGconn *conn); extern int pqPacketSend(PGconn *conn, char pack_type, const void *buf, size_t buf_len); extern bool pqGetHomeDirectory(char *buf, int bufsize); +extern bool pqCopyPGconn(PGconn *srcConn, PGconn *dstConn); extern bool pqParseIntParam(const char *value, int *result, PGconn *conn, const char *context); diff --git a/src/test/modules/libpq_pipeline/libpq_pipeline.c b/src/test/modules/libpq_pipeline/libpq_pipeline.c index 5f43aa40de4..7f32b61df40 100644 --- a/src/test/modules/libpq_pipeline/libpq_pipeline.c +++ b/src/test/modules/libpq_pipeline/libpq_pipeline.c @@ -86,6 +86,264 @@ pg_fatal_impl(int line, const char *fmt,...) exit(1); } +/* + * Check that the query on the given connection got canceled. + * + * This is a function wrapped in a macro to make the reported line number + * in an error match the line number of the invocation. + */ +#define confirm_query_canceled(conn) confirm_query_canceled_impl(__LINE__, conn) +static void +confirm_query_canceled_impl(int line, PGconn *conn) +{ + PGresult *res = NULL; + + res = PQgetResult(conn); + if (res == NULL) + pg_fatal_impl(line, "PQgetResult returned null: %s", + PQerrorMessage(conn)); + if (PQresultStatus(res) != PGRES_FATAL_ERROR) + pg_fatal_impl(line, "query did not fail when it was expected"); + if (strcmp(PQresultErrorField(res, PG_DIAG_SQLSTATE), "57014") != 0) + pg_fatal_impl(line, "query failed with a different error than cancellation: %s", + PQerrorMessage(conn)); + PQclear(res); + while (PQisBusy(conn)) + { + PQconsumeInput(conn); + } +} + +#define send_cancellable_query(conn, monitorConn) send_cancellable_query_impl(__LINE__, conn, monitorConn) +static void +send_cancellable_query_impl(int line, PGconn *conn, PGconn *monitorConn) +{ + const char *env_wait; + const Oid paramTypes[1] = {INT4OID}; + + env_wait = getenv("PG_TEST_TIMEOUT_DEFAULT"); + if (env_wait == NULL) + env_wait = "180"; + + if (PQsendQueryParams(conn, "SELECT pg_sleep($1)", 1, paramTypes, &env_wait, NULL, NULL, 0) != 1) + pg_fatal_impl(line, "failed to send query: %s", PQerrorMessage(conn)); + + /* + * Wait until the query is actually running. Otherwise sending a + * cancellation request might not cancel the query due to race conditions. + */ + while (true) + { + char *value = NULL; + PGresult *res = PQexec( + monitorConn, + "SELECT count(*) FROM pg_stat_activity WHERE " + "query = 'SELECT pg_sleep($1)' " + "AND state = 'active'"); + + if (PQresultStatus(res) != PGRES_TUPLES_OK) + { + pg_fatal("Connection to database failed: %s", PQerrorMessage(monitorConn)); + } + if (PQntuples(res) != 1) + { + pg_fatal("unexpected number of rows received: %d", PQntuples(res)); + } + if (PQnfields(res) != 1) + { + pg_fatal("unexpected number of columns received: %d", PQnfields(res)); + } + value = PQgetvalue(res, 0, 0); + if (*value != '0') + { + PQclear(res); + break; + } + PQclear(res); + + /* + * wait 10ms before polling again + */ + pg_usleep(10000); + } +} + +static void +test_cancel(PGconn *conn, const char *conninfo) +{ + PGcancel *cancel = NULL; + PGcancelConn *cancelConn = NULL; + PGconn *monitorConn = NULL; + char errorbuf[256]; + + fprintf(stderr, "test cancellations... "); + + if (PQsetnonblocking(conn, 1) != 0) + pg_fatal("failed to set nonblocking mode: %s", PQerrorMessage(conn)); + + /* + * Make a connection to the database to monitor the query on the main + * connection. + */ + monitorConn = PQconnectdb(conninfo); + if (PQstatus(conn) != CONNECTION_OK) + { + pg_fatal("Connection to database failed: %s", + PQerrorMessage(conn)); + } + + /* test PQcancel */ + send_cancellable_query(conn, monitorConn); + cancel = PQgetCancel(conn); + if (!PQcancel(cancel, errorbuf, sizeof(errorbuf))) + { + pg_fatal("failed to run PQcancel: %s", errorbuf); + }; + confirm_query_canceled(conn); + + /* PGcancel object can be reused for the next query */ + send_cancellable_query(conn, monitorConn); + if (!PQcancel(cancel, errorbuf, sizeof(errorbuf))) + { + pg_fatal("failed to run PQcancel: %s", errorbuf); + }; + confirm_query_canceled(conn); + + PQfreeCancel(cancel); + + /* test PQrequestCancel */ + send_cancellable_query(conn, monitorConn); + if (!PQrequestCancel(conn)) + pg_fatal("failed to run PQrequestCancel: %s", PQerrorMessage(conn)); + confirm_query_canceled(conn); + + /* test PQcancelSend */ + send_cancellable_query(conn, monitorConn); + cancelConn = PQcancelConn(conn); + if (!PQcancelSend(cancelConn)) + pg_fatal("failed to run PQcancelSend: %s", PQcancelErrorMessage(cancelConn)); + confirm_query_canceled(conn); + PQcancelFinish(cancelConn); + + /* test PQcancelConn and then polling with PQcancelPoll */ + send_cancellable_query(conn, monitorConn); + cancelConn = PQcancelConn(conn); + if (!PQcancelStart(cancelConn)) + pg_fatal("bad cancel connection: %s", PQcancelErrorMessage(cancelConn)); + while (true) + { + struct timeval tv; + fd_set input_mask; + fd_set output_mask; + PostgresPollingStatusType pollres = PQcancelPoll(cancelConn); + int sock = PQcancelSocket(cancelConn); + + if (pollres == PGRES_POLLING_OK) + { + break; + } + + FD_ZERO(&input_mask); + FD_ZERO(&output_mask); + switch (pollres) + { + case PGRES_POLLING_READING: + pg_debug("polling for reads\n"); + FD_SET(sock, &input_mask); + break; + case PGRES_POLLING_WRITING: + pg_debug("polling for writes\n"); + FD_SET(sock, &output_mask); + break; + default: + pg_fatal("bad cancel connection: %s", PQcancelErrorMessage(cancelConn)); + } + + if (sock < 0) + pg_fatal("sock did not exist: %s", PQcancelErrorMessage(cancelConn)); + + tv.tv_sec = 3; + tv.tv_usec = 0; + + while (true) + { + if (select(sock + 1, &input_mask, &output_mask, NULL, &tv) < 0) + { + if (errno == EINTR) + continue; + pg_fatal("select() failed: %m"); + } + break; + } + } + if (PQcancelStatus(cancelConn) != CONNECTION_OK) + pg_fatal("unexpected cancel connection status: %s", PQcancelErrorMessage(cancelConn)); + confirm_query_canceled(conn); + + /* + * test PQcancelReset works on the cancel connection and it can be reused + * after + */ + PQcancelReset(cancelConn); + + send_cancellable_query(conn, monitorConn); + if (!PQcancelStart(cancelConn)) + pg_fatal("bad cancel connection: %s", PQcancelErrorMessage(cancelConn)); + while (true) + { + struct timeval tv; + fd_set input_mask; + fd_set output_mask; + PostgresPollingStatusType pollres = PQcancelPoll(cancelConn); + int sock = PQcancelSocket(cancelConn); + + if (pollres == PGRES_POLLING_OK) + { + break; + } + + FD_ZERO(&input_mask); + FD_ZERO(&output_mask); + switch (pollres) + { + case PGRES_POLLING_READING: + pg_debug("polling for reads\n"); + FD_SET(sock, &input_mask); + break; + case PGRES_POLLING_WRITING: + pg_debug("polling for writes\n"); + FD_SET(sock, &output_mask); + break; + default: + pg_fatal("bad cancel connection: %s", PQcancelErrorMessage(cancelConn)); + } + + if (sock < 0) + pg_fatal("sock did not exist: %s", PQcancelErrorMessage(cancelConn)); + + tv.tv_sec = 3; + tv.tv_usec = 0; + + while (true) + { + if (select(sock + 1, &input_mask, &output_mask, NULL, &tv) < 0) + { + if (errno == EINTR) + continue; + pg_fatal("select() failed: %m"); + } + break; + } + } + if (PQcancelStatus(cancelConn) != CONNECTION_OK) + pg_fatal("unexpected cancel connection status: %s", PQcancelErrorMessage(cancelConn)); + confirm_query_canceled(conn); + + PQcancelFinish(cancelConn); + + fprintf(stderr, "ok\n"); +} + static void test_disallowed_in_pipeline(PGconn *conn) { @@ -1789,6 +2047,7 @@ usage(const char *progname) static void print_test_list(void) { + printf("cancel\n"); printf("disallowed_in_pipeline\n"); printf("multi_pipelines\n"); printf("nosync\n"); @@ -1890,7 +2149,9 @@ main(int argc, char **argv) PQTRACE_SUPPRESS_TIMESTAMPS | PQTRACE_REGRESS_MODE); } - if (strcmp(testname, "disallowed_in_pipeline") == 0) + if (strcmp(testname, "cancel") == 0) + test_cancel(conn, conninfo); + else if (strcmp(testname, "disallowed_in_pipeline") == 0) test_disallowed_in_pipeline(conn); else if (strcmp(testname, "multi_pipelines") == 0) test_multi_pipelines(conn); diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index d808aad8b05..b2b83b69c2f 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -1760,6 +1760,7 @@ PG_Locale_Strategy PG_Lock_Status PG_init_t PGcancel +PGcancelConn PGcmdQueueEntry PGconn PGdataValue base-commit: bd8fc1677b88ed80e4e00e0e46401ec537952482 -- 2.34.1