From e7e1f0b2a5275ac88626f03db2535fa8323f3999 Mon Sep 17 00:00:00 2001 From: Anthonin Bonnefoy Date: Mon, 24 Aug 2026 08:54:08 +0200 Subject: [PATCH v2 1/2] Create a new msg_buffer struct for PGconn's inBuffer Create a new struct to store buffer, size, start, cursor and end of PGconn's inBuffer and pass it as a parameter to all functions reading from inBuffer. The goal is to be able to provide a different msg_buffer to read from, which is going to be necessary for the decompression support as the decompressed messages will be stored in a dedicated buffer. --- src/interfaces/libpq/fe-auth.c | 20 +- src/interfaces/libpq/fe-connect.c | 57 +++-- src/interfaces/libpq/fe-misc.c | 177 ++++++++-------- src/interfaces/libpq/fe-protocol3.c | 253 ++++++++++++----------- src/interfaces/libpq/fe-secure-openssl.c | 3 +- src/interfaces/libpq/libpq-int.h | 36 ++-- src/tools/pgindent/typedefs.list | 1 + 7 files changed, 281 insertions(+), 266 deletions(-) diff --git a/src/interfaces/libpq/fe-auth.c b/src/interfaces/libpq/fe-auth.c index e3bddf91203..8df5bad92ab 100644 --- a/src/interfaces/libpq/fe-auth.c +++ b/src/interfaces/libpq/fe-auth.c @@ -81,7 +81,7 @@ pg_GSS_continue(PGconn *conn, int payloadlen) payloadlen); return STATUS_ERROR; } - if (pqGetnchar(ginbuf.value, payloadlen, conn)) + if (pqGetnchar(ginbuf.value, payloadlen, conn, &conn->inBuffer)) { /* * Shouldn't happen, because the caller should've ensured that the @@ -99,7 +99,7 @@ pg_GSS_continue(PGconn *conn, int payloadlen) /* finished parsing, trace server-to-client message */ if (conn->Pfdebug) - pqTraceOutputMessage(conn, conn->inBuffer + conn->inStart, false); + pqTraceOutputMessage(conn, conn->inBuffer.buffer + conn->inBuffer.start, false); /* Only try to acquire credentials if GSS delegation isn't disabled. */ if (!pg_GSS_have_cred_cache(&conn->gcred)) @@ -247,7 +247,7 @@ pg_SSPI_continue(PGconn *conn, int payloadlen) payloadlen); return STATUS_ERROR; } - if (pqGetnchar(inputbuf, payloadlen, conn)) + if (pqGetnchar(inputbuf, payloadlen, conn, &conn->inBuffer)) { /* * Shouldn't happen, because the caller should've ensured that the @@ -267,7 +267,7 @@ pg_SSPI_continue(PGconn *conn, int payloadlen) /* finished parsing, trace server-to-client message */ if (conn->Pfdebug) - pqTraceOutputMessage(conn, conn->inBuffer + conn->inStart, false); + pqTraceOutputMessage(conn, conn->inBuffer.buffer + conn->inBuffer.start, false); OutBuffers[0].pvBuffer = NULL; OutBuffers[0].BufferType = SECBUFFER_TOKEN; @@ -464,7 +464,7 @@ pg_SASL_init(PGconn *conn, int payloadlen, bool *async) selected_mechanism = NULL; for (;;) { - if (pqGets(&mechanism_buf, conn)) + if (pqGets(&mechanism_buf, conn, &conn->inBuffer)) { appendPQExpBufferStr(&conn->errorMessage, "fe_sendauth: invalid authentication request from server: invalid list of authentication mechanisms\n"); @@ -605,7 +605,7 @@ pg_SASL_init(PGconn *conn, int payloadlen, bool *async) /* finished parsing, trace server-to-client message */ if (conn->Pfdebug) - pqTraceOutputMessage(conn, conn->inBuffer + conn->inStart, false); + pqTraceOutputMessage(conn, conn->inBuffer.buffer + conn->inBuffer.start, false); Assert(conn->sasl); @@ -718,7 +718,7 @@ pg_SASL_continue(PGconn *conn, int payloadlen, bool final, bool *async) return STATUS_ERROR; } - if (pqGetnchar(challenge, payloadlen, conn)) + if (pqGetnchar(challenge, payloadlen, conn, &conn->inBuffer)) { free(challenge); return STATUS_ERROR; @@ -726,7 +726,7 @@ pg_SASL_continue(PGconn *conn, int payloadlen, bool final, bool *async) /* finished parsing, trace server-to-client message */ if (conn->Pfdebug) - pqTraceOutputMessage(conn, conn->inBuffer + conn->inStart, false); + pqTraceOutputMessage(conn, conn->inBuffer.buffer + conn->inBuffer.start, false); /* For safety and convenience, ensure the buffer is NULL-terminated. */ challenge[payloadlen] = '\0'; @@ -803,13 +803,13 @@ pg_password_sendauth(PGconn *conn, const char *password, AuthRequest areq) /* Read the salt from the AuthenticationMD5Password message. */ if (areq == AUTH_REQ_MD5) { - if (pqGetnchar(md5Salt, 4, conn)) + if (pqGetnchar(md5Salt, 4, conn, &conn->inBuffer)) return STATUS_ERROR; /* shouldn't happen */ } /* finished parsing, trace server-to-client message */ if (conn->Pfdebug) - pqTraceOutputMessage(conn, conn->inBuffer + conn->inStart, false); + pqTraceOutputMessage(conn, conn->inBuffer.buffer + conn->inBuffer.start, false); /* Encrypt the password if needed. */ diff --git a/src/interfaces/libpq/fe-connect.c b/src/interfaces/libpq/fe-connect.c index 811cb0f2a50..0941ae4b6ca 100644 --- a/src/interfaces/libpq/fe-connect.c +++ b/src/interfaces/libpq/fe-connect.c @@ -541,7 +541,7 @@ pqDropConnection(PGconn *conn, bool flushInput) /* Optionally discard any unread data */ if (flushInput) - conn->inStart = conn->inCursor = conn->inEnd = 0; + conn->inBuffer.start = conn->inBuffer.cursor = conn->inBuffer.end = 0; /* Always discard any unsent data */ conn->outCount = 0; @@ -2740,7 +2740,7 @@ pqConnectDBStart(PGconn *conn) } /* Ensure our buffers are empty */ - conn->inStart = conn->inCursor = conn->inEnd = 0; + conn->inBuffer.start = conn->inBuffer.cursor = conn->inBuffer.end = 0; conn->outCount = 0; /* @@ -3799,7 +3799,7 @@ keep_going: /* We will come back to here until there is /* caller failed to wait for data */ return PGRES_POLLING_READING; } - if (pqGetc(&SSLok, conn) < 0) + if (pqGetc(&SSLok, conn, &conn->inBuffer) < 0) { /* should not happen really */ return PGRES_POLLING_READING; @@ -3810,7 +3810,7 @@ keep_going: /* We will come back to here until there is pqTraceOutputCharResponse(conn, "SSLResponse", SSLok); /* mark byte consumed */ - conn->inStart = conn->inCursor; + conn->inBuffer.start = conn->inBuffer.cursor; } else if (SSLok == 'N') { @@ -3818,7 +3818,7 @@ keep_going: /* We will come back to here until there is pqTraceOutputCharResponse(conn, "SSLResponse", SSLok); /* mark byte consumed */ - conn->inStart = conn->inCursor; + conn->inBuffer.start = conn->inBuffer.cursor; /* * The connection is still valid, so if it's OK to @@ -3858,7 +3858,7 @@ keep_going: /* We will come back to here until there is * handshake, so it wasn't encrypted and indeed may have * been injected by a man-in-the-middle. */ - if (conn->inCursor != conn->inEnd) + if (conn->inBuffer.cursor != conn->inBuffer.end) { libpq_append_conn_error(conn, "received unencrypted data after SSL response"); goto error_return; @@ -3904,7 +3904,7 @@ keep_going: /* We will come back to here until there is else if (rdresult == 0) /* caller failed to wait for data */ return PGRES_POLLING_READING; - if (pqGetc(&gss_ok, conn) < 0) + if (pqGetc(&gss_ok, conn, &conn->inBuffer) < 0) /* shouldn't happen... */ return PGRES_POLLING_READING; @@ -3928,7 +3928,7 @@ keep_going: /* We will come back to here until there is } /* mark byte consumed */ - conn->inStart = conn->inCursor; + conn->inBuffer.start = conn->inBuffer.cursor; if (gss_ok == 'N') { @@ -3965,7 +3965,7 @@ keep_going: /* We will come back to here until there is * handshake, so it wasn't encrypted and indeed may have * been injected by a man-in-the-middle. */ - if (conn->inCursor != conn->inEnd) + if (conn->inBuffer.cursor != conn->inBuffer.end) { libpq_append_conn_error(conn, "received unencrypted data after GSSAPI encryption response"); goto error_return; @@ -4009,10 +4009,10 @@ keep_going: /* We will come back to here until there is * the message is incomplete, we will return without advancing * inStart, and resume here next time). */ - conn->inCursor = conn->inStart; + conn->inBuffer.cursor = conn->inBuffer.start; /* Read type byte */ - if (pqGetc(&beresp, conn)) + if (pqGetc(&beresp, conn, &conn->inBuffer)) { /* We'll come back when there is more data */ return PGRES_POLLING_READING; @@ -4034,7 +4034,7 @@ keep_going: /* We will come back to here until there is } /* Read message length word */ - if (pqGetInt(&msgLength, 4, conn)) + if (pqGetInt(&msgLength, 4, conn, &conn->inBuffer)) { /* We'll come back when there is more data */ return PGRES_POLLING_READING; @@ -4071,14 +4071,14 @@ keep_going: /* We will come back to here until there is (msgLength < 8 || msgLength > MAX_ERRLEN)) { /* Handle error from a pre-3.0 server */ - conn->inCursor = conn->inStart + 1; /* reread data */ - if (pqGets_append(&conn->errorMessage, conn)) + conn->inBuffer.cursor = conn->inBuffer.start + 1; /* reread data */ + if (pqGets_append(&conn->errorMessage, conn, &conn->inBuffer)) { /* * We may not have authenticated the server yet, so * don't let the buffer grow forever. */ - avail = conn->inEnd - conn->inCursor; + avail = conn->inBuffer.end - conn->inBuffer.cursor; if (avail > MAX_ERRLEN) { libpq_append_conn_error(conn, "received invalid error message"); @@ -4089,7 +4089,7 @@ keep_going: /* We will come back to here until there is return PGRES_POLLING_READING; } /* OK, we read the message; mark data consumed */ - pqParseDone(conn, conn->inCursor); + pqParseDone(conn, &conn->inBuffer, conn->inBuffer.cursor); /* * Before 7.2, the postmaster didn't always end its @@ -4115,7 +4115,7 @@ keep_going: /* We will come back to here until there is * PGRES_POLLING_READING after this point. */ msgLength -= 4; - avail = conn->inEnd - conn->inCursor; + avail = conn->inBuffer.end - conn->inBuffer.cursor; if (avail < msgLength) { /* @@ -4123,8 +4123,7 @@ keep_going: /* We will come back to here until there is * needed to hold the whole message; see notes in * pqParseInput3. */ - if (pqCheckInBufferSpace(conn->inCursor + (size_t) msgLength, - conn)) + if (pqCheckMsgBufferSpace(conn->inBuffer.cursor + (size_t) msgLength, &conn->inBuffer, conn)) goto error_return; /* We'll come back when there is more data */ return PGRES_POLLING_READING; @@ -4133,13 +4132,13 @@ keep_going: /* We will come back to here until there is /* Handle errors. */ if (beresp == PqMsg_ErrorResponse) { - if (pqGetErrorNotice3(conn, true)) + if (pqGetErrorNotice3(conn, true, &conn->inBuffer)) { libpq_append_conn_error(conn, "received invalid error message"); goto error_return; } /* OK, we read the message; mark data consumed */ - pqParseDone(conn, conn->inCursor); + pqParseDone(conn, &conn->inBuffer, conn->inBuffer.cursor); /* * If error is "cannot connect now", try the next host if @@ -4168,7 +4167,7 @@ keep_going: /* We will come back to here until there is libpq_append_conn_error(conn, "received duplicate protocol negotiation message"); goto error_return; } - if (pqGetNegotiateProtocolVersion3(conn)) + if (pqGetNegotiateProtocolVersion3(conn, &conn->inBuffer)) { /* pqGetNegotiateProtocolVersion3 set error already */ goto error_return; @@ -4176,7 +4175,7 @@ keep_going: /* We will come back to here until there is conn->pversion_negotiated = true; /* OK, we read the message; mark data consumed */ - pqParseDone(conn, conn->inCursor); + pqParseDone(conn, &conn->inBuffer, conn->inBuffer.cursor); goto keep_going; } @@ -4185,7 +4184,7 @@ keep_going: /* We will come back to here until there is conn->auth_req_received = true; /* Get the type of request. */ - if (pqGetInt((int *) &areq, 4, conn)) + if (pqGetInt((int *) &areq, 4, conn, &conn->inBuffer)) { /* can't happen because we checked the length already */ libpq_append_conn_error(conn, "received invalid authentication request"); @@ -4217,7 +4216,7 @@ keep_going: /* We will come back to here until there is * don't call pqParseDone here because we already traced this * message inside pg_fe_sendauth. */ - conn->inStart = conn->inCursor; + conn->inBuffer.start = conn->inBuffer.cursor; if (res != STATUS_OK) { @@ -5031,8 +5030,8 @@ pqMakeEmptyPGconn(void) * be enlarged anytime it has less than 8K free, so we initially allocate * twice that. */ - conn->inBufSize = 16 * 1024; - conn->inBuffer = (char *) malloc(conn->inBufSize); + conn->inBuffer.bufSize = 16 * 1024; + conn->inBuffer.buffer = (char *) malloc(conn->inBuffer.bufSize); conn->outBufSize = 16 * 1024; conn->outBuffer = (char *) malloc(conn->outBufSize); conn->rowBufLen = 32; @@ -5040,7 +5039,7 @@ pqMakeEmptyPGconn(void) initPQExpBuffer(&conn->errorMessage); initPQExpBuffer(&conn->workBuffer); - if (conn->inBuffer == NULL || + if (conn->inBuffer.buffer == NULL || conn->outBuffer == NULL || conn->rowBuf == NULL || PQExpBufferBroken(&conn->errorMessage) || @@ -5148,7 +5147,7 @@ freePGconn(PGconn *conn) free(conn->scram_server_key_binary); /* if this is a cancel connection, be_cancel_key may still be allocated */ free(conn->be_cancel_key); - free(conn->inBuffer); + free(conn->inBuffer.buffer); free(conn->outBuffer); free(conn->rowBuf); termPQExpBuffer(&conn->errorMessage); diff --git a/src/interfaces/libpq/fe-misc.c b/src/interfaces/libpq/fe-misc.c index 73d10e5cae5..2bb8b6e67f4 100644 --- a/src/interfaces/libpq/fe-misc.c +++ b/src/interfaces/libpq/fe-misc.c @@ -76,12 +76,12 @@ PQlibVersion(void) * data in the buffer, not that there is necessarily a hard error. */ int -pqGetc(char *result, PGconn *conn) +pqGetc(char *result, PGconn *conn, msg_buffer *msgBuf) { - if (conn->inCursor >= conn->inEnd) + if (msgBuf->cursor >= msgBuf->end) return EOF; - *result = conn->inBuffer[conn->inCursor++]; + *result = msgBuf->buffer[msgBuf->cursor++]; return 0; } @@ -108,12 +108,12 @@ pqPutc(char c, PGconn *conn) * but the excess characters are silently discarded. */ static int -pqGets_internal(PQExpBuffer buf, PGconn *conn, bool resetbuffer) +pqGets_internal(PQExpBuffer buf, PGconn *conn, msg_buffer *msgBuf, bool resetbuffer) { /* Copy conn data to locals for faster search loop */ - char *inBuffer = conn->inBuffer; - int inCursor = conn->inCursor; - int inEnd = conn->inEnd; + char *inBuffer = msgBuf->buffer; + int inCursor = msgBuf->cursor; + int inEnd = msgBuf->end; int slen; while (inCursor < inEnd && inBuffer[inCursor]) @@ -122,28 +122,28 @@ pqGets_internal(PQExpBuffer buf, PGconn *conn, bool resetbuffer) if (inCursor >= inEnd) return EOF; - slen = inCursor - conn->inCursor; + slen = inCursor - msgBuf->cursor; if (resetbuffer) resetPQExpBuffer(buf); - appendBinaryPQExpBuffer(buf, inBuffer + conn->inCursor, slen); + appendBinaryPQExpBuffer(buf, inBuffer + msgBuf->cursor, slen); - conn->inCursor = ++inCursor; + msgBuf->cursor = ++inCursor; return 0; } int -pqGets(PQExpBuffer buf, PGconn *conn) +pqGets(PQExpBuffer buf, PGconn *conn, msg_buffer *msgBuf) { - return pqGets_internal(buf, conn, true); + return pqGets_internal(buf, conn, msgBuf, true); } int -pqGets_append(PQExpBuffer buf, PGconn *conn) +pqGets_append(PQExpBuffer buf, PGconn *conn, msg_buffer *msgBuf) { - return pqGets_internal(buf, conn, false); + return pqGets_internal(buf, conn, msgBuf, false); } @@ -164,15 +164,15 @@ pqPuts(const char *s, PGconn *conn) * read exactly len bytes in buffer s, no null termination */ int -pqGetnchar(void *s, size_t len, PGconn *conn) +pqGetnchar(void *s, size_t len, PGconn *conn, msg_buffer *msgBuf) { - if (len > (size_t) (conn->inEnd - conn->inCursor)) + if (len > (size_t) (msgBuf->end - msgBuf->cursor)) return EOF; - memcpy(s, conn->inBuffer + conn->inCursor, len); + memcpy(s, msgBuf->buffer + msgBuf->cursor, len); /* no terminating null */ - conn->inCursor += len; + msgBuf->cursor += len; return 0; } @@ -186,12 +186,12 @@ pqGetnchar(void *s, size_t len, PGconn *conn) * will actually be used, but just isn't getting copied anywhere as yet. */ int -pqSkipnchar(size_t len, PGconn *conn) +pqSkipnchar(size_t len, PGconn *conn, msg_buffer *msgBuf) { - if (len > (size_t) (conn->inEnd - conn->inCursor)) + if (len > (size_t) (msgBuf->end - msgBuf->cursor)) return EOF; - conn->inCursor += len; + msgBuf->cursor += len; return 0; } @@ -215,7 +215,7 @@ pqPutnchar(const void *s, size_t len, PGconn *conn) * to local byte order */ int -pqGetInt(int *result, size_t bytes, PGconn *conn) +pqGetInt(int *result, size_t bytes, PGconn *conn, msg_buffer *msgBuf) { uint16 tmp2; uint32 tmp4; @@ -223,17 +223,17 @@ pqGetInt(int *result, size_t bytes, PGconn *conn) switch (bytes) { case 2: - if (conn->inCursor + 2 > conn->inEnd) + if (msgBuf->cursor + 2 > msgBuf->end) return EOF; - memcpy(&tmp2, conn->inBuffer + conn->inCursor, 2); - conn->inCursor += 2; + memcpy(&tmp2, msgBuf->buffer + msgBuf->cursor, 2); + msgBuf->cursor += 2; *result = (int) pg_ntoh16(tmp2); break; case 4: - if (conn->inCursor + 4 > conn->inEnd) + if (msgBuf->cursor + 4 > msgBuf->end) return EOF; - memcpy(&tmp4, conn->inBuffer + conn->inCursor, 4); - conn->inCursor += 4; + memcpy(&tmp4, msgBuf->buffer + msgBuf->cursor, 4); + msgBuf->cursor += 4; *result = (int) pg_ntoh32(tmp4); break; default: @@ -344,15 +344,15 @@ pqCheckOutBufferSpace(size_t bytes_needed, PGconn *conn) } /* - * Make sure conn's input buffer can hold bytes_needed bytes (caller must - * include already-stored data into the value!) + * Make sure msgBuf can hold bytes_needed bytes (caller must include + * already-stored data into the value!) * * Returns 0 on success, EOF if failed to enlarge buffer */ int -pqCheckInBufferSpace(size_t bytes_needed, PGconn *conn) +pqCheckMsgBufferSpace(size_t bytes_needed, msg_buffer *msgBuf, PGconn *conn) { - int newsize = conn->inBufSize; + int newsize = msgBuf->bufSize; char *newbuf; /* Quick exit if we have enough space */ @@ -360,29 +360,29 @@ pqCheckInBufferSpace(size_t bytes_needed, PGconn *conn) return 0; /* - * Before concluding that we need to enlarge the buffer, left-justify + * Before concluding that we need to enlarge the msgBuf, left-justify * whatever is in it and recheck. The caller's value of bytes_needed * includes any data to the left of inStart, but we can delete that in - * preference to enlarging the buffer. It's slightly ugly to have this + * preference to enlarging the msgBuf. It's slightly ugly to have this * function do this, but it's better than making callers worry about it. */ - bytes_needed -= conn->inStart; + bytes_needed -= msgBuf->start; - if (conn->inStart < conn->inEnd) + if (msgBuf->start < msgBuf->end) { - if (conn->inStart > 0) + if (msgBuf->start > 0) { - memmove(conn->inBuffer, conn->inBuffer + conn->inStart, - conn->inEnd - conn->inStart); - conn->inEnd -= conn->inStart; - conn->inCursor -= conn->inStart; - conn->inStart = 0; + memmove(msgBuf->buffer, msgBuf->buffer + msgBuf->start, + msgBuf->end - msgBuf->start); + msgBuf->end -= msgBuf->start; + msgBuf->cursor -= msgBuf->start; + msgBuf->start = 0; } } else { - /* buffer is logically empty, reset it */ - conn->inStart = conn->inCursor = conn->inEnd = 0; + /* msgBuf is logically empty, reset it */ + msgBuf->start = msgBuf->cursor = msgBuf->end = 0; } /* Recheck whether we have enough space */ @@ -390,7 +390,7 @@ pqCheckInBufferSpace(size_t bytes_needed, PGconn *conn) return 0; /* - * If we need to enlarge the buffer, we first try to double it in size; if + * If we need to enlarge the msgBuf, we first try to double it in size; if * that doesn't work, enlarge in multiples of 8K. This avoids thrashing * the malloc pool by repeated small enlargements. * @@ -403,17 +403,17 @@ pqCheckInBufferSpace(size_t bytes_needed, PGconn *conn) if (newsize > 0 && bytes_needed <= (size_t) newsize) { - newbuf = realloc(conn->inBuffer, newsize); + newbuf = realloc(msgBuf->buffer, newsize); if (newbuf) { /* realloc succeeded */ - conn->inBuffer = newbuf; - conn->inBufSize = newsize; + msgBuf->buffer = newbuf; + msgBuf->bufSize = newsize; return 0; } } - newsize = conn->inBufSize; + newsize = msgBuf->bufSize; do { newsize += 8192; @@ -421,35 +421,35 @@ pqCheckInBufferSpace(size_t bytes_needed, PGconn *conn) if (newsize > 0 && bytes_needed <= (size_t) newsize) { - newbuf = realloc(conn->inBuffer, newsize); + newbuf = realloc(msgBuf->buffer, newsize); if (newbuf) { /* realloc succeeded */ - conn->inBuffer = newbuf; - conn->inBufSize = newsize; + msgBuf->buffer = newbuf; + msgBuf->bufSize = newsize; return 0; } } /* realloc failed. Probably out of memory */ appendPQExpBufferStr(&conn->errorMessage, - "cannot allocate memory for input buffer\n"); + "cannot allocate memory for message buffer\n"); return EOF; } /* * pqParseDone: after a server-to-client message has successfully - * been parsed, advance conn->inStart to account for it. + * been parsed, advance msgBuf->start to account for it. */ void -pqParseDone(PGconn *conn, int newInStart) +pqParseDone(PGconn *conn, msg_buffer *msgBuf, int newInStart) { /* trace server-to-client message */ if (conn->Pfdebug) - pqTraceOutputMessage(conn, conn->inBuffer + conn->inStart, false); + pqTraceOutputMessage(conn, msgBuf->buffer + msgBuf->start, false); /* Mark message as done */ - conn->inStart = newInStart; + msgBuf->start = newInStart; } /* @@ -607,7 +607,7 @@ pqPutMsgEnd(PGconn *conn) * 0: no data is presently available, but no error detected * -1: error detected (including EOF = connection closure); * conn->errorMessage set - * NOTE: callers must not assume that pointers or indexes into conn->inBuffer + * NOTE: callers must not assume that pointers or indexes into conn->inBuffer.buffer * remain valid across this call! * ---------- */ @@ -628,8 +628,9 @@ pqReadData(PGconn *conn) else if (available > 0) { /* - * Make sure there are no bytes stuck in layers between conn->inBuffer - * and the socket, to make it safe for clients to poll on PQsocket(). + * Make sure there are no bytes stuck in layers between + * conn->inBuffer.buffer and the socket, to make it safe for clients + * to poll on PQsocket(). */ if (pqDrainPending(conn)) return -1; @@ -657,21 +658,21 @@ pqReadData_internal(PGconn *conn) ssize_t nread; /* Left-justify any data in the buffer to make room */ - if (conn->inStart < conn->inEnd) + if (conn->inBuffer.start < conn->inBuffer.end) { - if (conn->inStart > 0) + if (conn->inBuffer.start > 0) { - memmove(conn->inBuffer, conn->inBuffer + conn->inStart, - conn->inEnd - conn->inStart); - conn->inEnd -= conn->inStart; - conn->inCursor -= conn->inStart; - conn->inStart = 0; + memmove(conn->inBuffer.buffer, conn->inBuffer.buffer + conn->inBuffer.start, + conn->inBuffer.end - conn->inBuffer.start); + conn->inBuffer.end -= conn->inBuffer.start; + conn->inBuffer.cursor -= conn->inBuffer.start; + conn->inBuffer.start = 0; } } else { /* buffer is logically empty, reset it */ - conn->inStart = conn->inCursor = conn->inEnd = 0; + conn->inBuffer.start = conn->inBuffer.cursor = conn->inBuffer.end = 0; } /* @@ -682,22 +683,22 @@ pqReadData_internal(PGconn *conn) * enough for a TCP packet or Unix pipe bufferload. 8K is the usual pipe * buffer size, so... */ - if (conn->inBufSize - conn->inEnd < 8192) + if (conn->inBuffer.bufSize - conn->inBuffer.end < 8192) { - if (pqCheckInBufferSpace(conn->inEnd + (size_t) 8192, conn)) + if (pqCheckMsgBufferSpace(conn->inBuffer.end + (size_t) 8192, &conn->inBuffer, conn)) { /* * We don't insist that the enlarge worked, but we need some room */ - if (conn->inBufSize - conn->inEnd < 100) + if (conn->inBuffer.bufSize - conn->inBuffer.end < 100) return -1; /* errorMessage already set */ } } /* OK, try to read some data */ retry3: - nread = pqsecure_read(conn, conn->inBuffer + conn->inEnd, - conn->inBufSize - conn->inEnd); + nread = pqsecure_read(conn, conn->inBuffer.buffer + conn->inBuffer.end, + conn->inBuffer.bufSize - conn->inBuffer.end); if (nread < 0) { switch (SOCK_ERRNO) @@ -726,7 +727,7 @@ retry3: } if (nread > 0) { - conn->inEnd += nread; + conn->inBuffer.end += nread; /* * Hack to deal with the fact that some kernels will only give us back @@ -736,12 +737,12 @@ retry3: * buffer space. Without this, the block-and-restart behavior of * libpq's higher levels leads to O(N^2) performance on long messages. * - * Since we left-justified the data above, conn->inEnd gives the - * amount of data already read in the current message. We consider - * the message "long" once we have acquired 32k ... + * Since we left-justified the data above, conn->inBuffer.end gives + * the amount of data already read in the current message. We + * consider the message "long" once we have acquired 32k ... */ - if (conn->inEnd > 32768 && - (conn->inBufSize - conn->inEnd) >= 8192) + if (conn->inBuffer.end > 32768 && + (conn->inBuffer.bufSize - conn->inBuffer.end) >= 8192) { someread = 1; goto retry3; @@ -791,8 +792,8 @@ retry3: * arrived. */ retry4: - nread = pqsecure_read(conn, conn->inBuffer + conn->inEnd, - conn->inBufSize - conn->inEnd); + nread = pqsecure_read(conn, conn->inBuffer.buffer + conn->inBuffer.end, + conn->inBuffer.bufSize - conn->inBuffer.end); if (nread < 0) { switch (SOCK_ERRNO) @@ -821,7 +822,7 @@ retry4: } if (nread > 0) { - conn->inEnd += nread; + conn->inBuffer.end += nread; return 1; } @@ -844,7 +845,7 @@ definitelyFailed: /*--- * Drain any transport data that is already buffered in userspace and add it - * to conn->inBuffer, enlarging inBuffer if necessary. The drain fails if + * to conn->inBuffer.buffer, enlarging inBuffer if necessary. The drain fails if * inBuffer cannot be made to hold all available transport data. * * We assume that the underlying secure transport implementation does not @@ -909,7 +910,7 @@ definitelyFailed: * remaining data from the SSL buffer to the libpq buffer. * * The function returns 0 on success and -1 on error. Success means that - * there was no data pending or it was successfully drained to conn->inBuffer. + * there was no data pending or it was successfully drained to conn->inBuffer.buffer. * On error, conn->errorMessage is set. */ static int @@ -923,10 +924,10 @@ pqDrainPending(PGconn *conn) return bytes_pending; /* Expand the input buffer if necessary. */ - if (pqCheckInBufferSpace(conn->inEnd + (size_t) bytes_pending, conn)) + if (pqCheckMsgBufferSpace(conn->inBuffer.end + (size_t) bytes_pending, &conn->inBuffer, conn)) return -1; /* errorMessage already set */ - nread = pqsecure_read(conn, conn->inBuffer + conn->inEnd, + nread = pqsecure_read(conn, conn->inBuffer.buffer + conn->inBuffer.end, bytes_pending); /* @@ -935,7 +936,7 @@ pqDrainPending(PGconn *conn) */ if (nread < 0) return -1; - conn->inEnd += nread; + conn->inBuffer.end += nread; if (nread != bytes_pending) { libpq_append_conn_error(conn, @@ -956,7 +957,7 @@ pqDrainPending(PGconn *conn) * because the socket would block and the connection is non-blocking. * * Note that this is also responsible for consuming data from the socket - * (putting it in conn->inBuffer) in any situation where we can't send + * (putting it in conn->inBuffer.buffer) in any situation where we can't send * all the specified data immediately. * * If a socket-level write failure occurs, conn->write_failed is set and the diff --git a/src/interfaces/libpq/fe-protocol3.c b/src/interfaces/libpq/fe-protocol3.c index 1fd427cb9be..63a308915c9 100644 --- a/src/interfaces/libpq/fe-protocol3.c +++ b/src/interfaces/libpq/fe-protocol3.c @@ -48,14 +48,14 @@ static void handleFatalError(PGconn *conn); static void handleSyncLoss(PGconn *conn, char id, int msgLength); -static int getRowDescriptions(PGconn *conn, int msgLength); -static int getParamDescriptions(PGconn *conn, int msgLength); -static int getAnotherTuple(PGconn *conn, int msgLength); -static int getParameterStatus(PGconn *conn); -static int getBackendKeyData(PGconn *conn, int msgLength); -static int getNotify(PGconn *conn); -static int getCopyStart(PGconn *conn, ExecStatusType copytype); -static int getReadyForQuery(PGconn *conn); +static int getRowDescriptions(PGconn *conn, int msgLength, msg_buffer *msgBuf); +static int getParamDescriptions(PGconn *conn, int msgLength, msg_buffer *msgBuf); +static int getAnotherTuple(PGconn *conn, int msgLength, msg_buffer *msgBuf); +static int getParameterStatus(PGconn *conn, msg_buffer *msgBuf); +static int getBackendKeyData(PGconn *conn, int msgLength, msg_buffer *msgBuf); +static int getNotify(PGconn *conn, msg_buffer *msgBuf); +static int getCopyStart(PGconn *conn, ExecStatusType copytype, msg_buffer *msgBuf); +static int getReadyForQuery(PGconn *conn, msg_buffer *msgBuf); static void reportErrorPosition(PQExpBuffer msg, const char *query, int loc, int encoding); static size_t build_startup_packet(const PGconn *conn, char *packet, @@ -73,6 +73,7 @@ pqParseInput3(PGconn *conn) char id; int msgLength; int avail; + msg_buffer *msgBuf = &conn->inBuffer; /* * Loop to parse successive complete messages available in the buffer. @@ -83,10 +84,10 @@ pqParseInput3(PGconn *conn) * Try to read a message. First get the type code and length. Return * if not enough data. */ - conn->inCursor = conn->inStart; - if (pqGetc(&id, conn)) + msgBuf->cursor = msgBuf->start; + if (pqGetc(&id, conn, msgBuf)) return; - if (pqGetInt(&msgLength, 4, conn)) + if (pqGetInt(&msgLength, 4, conn, msgBuf)) return; /* @@ -109,7 +110,7 @@ pqParseInput3(PGconn *conn) * Can't process if message body isn't all here yet. */ msgLength -= 4; - avail = conn->inEnd - conn->inCursor; + avail = msgBuf->end - msgBuf->cursor; if (avail < msgLength) { /* @@ -120,14 +121,14 @@ pqParseInput3(PGconn *conn) * recovery strategy if we are unable to make the buffer big * enough. */ - if (pqCheckInBufferSpace(conn->inCursor + (size_t) msgLength, - conn)) + if (pqCheckMsgBufferSpace(msgBuf->cursor + (size_t) msgLength, + msgBuf, conn)) { /* * Abandon the connection. There's not much else we can * safely do; we can't just ignore the message or we could * miss important changes to the connection state. - * pqCheckInBufferSpace() already reported the error. + * pqCheckMsgBufferSpace() already reported the error. */ handleFatalError(conn); } @@ -152,12 +153,12 @@ pqParseInput3(PGconn *conn) */ if (id == PqMsg_NotificationResponse) { - if (getNotify(conn)) + if (getNotify(conn, msgBuf)) return; } else if (id == PqMsg_NoticeResponse) { - if (pqGetErrorNotice3(conn, false)) + if (pqGetErrorNotice3(conn, false, msgBuf)) return; } else if (conn->asyncStatus != PGASYNC_BUSY) @@ -177,12 +178,12 @@ pqParseInput3(PGconn *conn) */ if (id == PqMsg_ErrorResponse) { - if (pqGetErrorNotice3(conn, false /* treat as notice */ )) + if (pqGetErrorNotice3(conn, false /* treat as notice */ , msgBuf)) return; } else if (id == PqMsg_ParameterStatus) { - if (getParameterStatus(conn)) + if (getParameterStatus(conn, msgBuf)) return; } else @@ -192,7 +193,7 @@ pqParseInput3(PGconn *conn) "message type 0x%02x arrived from server while idle", id); /* Discard the unexpected message */ - conn->inCursor += msgLength; + msgBuf->cursor += msgLength; } } else @@ -203,7 +204,7 @@ pqParseInput3(PGconn *conn) switch (id) { case PqMsg_CommandComplete: - if (pqGets(&conn->workBuffer, conn)) + if (pqGets(&conn->workBuffer, conn, msgBuf)) return; if (!pgHavePendingResult(conn)) { @@ -221,12 +222,12 @@ pqParseInput3(PGconn *conn) conn->asyncStatus = PGASYNC_READY; break; case PqMsg_ErrorResponse: - if (pqGetErrorNotice3(conn, true)) + if (pqGetErrorNotice3(conn, true, msgBuf)) return; conn->asyncStatus = PGASYNC_READY; break; case PqMsg_ReadyForQuery: - if (getReadyForQuery(conn)) + if (getReadyForQuery(conn, msgBuf)) return; if (conn->pipelineStatus != PQ_PIPELINE_OFF) { @@ -303,7 +304,7 @@ pqParseInput3(PGconn *conn) } break; case PqMsg_ParameterStatus: - if (getParameterStatus(conn)) + if (getParameterStatus(conn, msgBuf)) return; break; case PqMsg_BackendKeyData: @@ -313,7 +314,7 @@ pqParseInput3(PGconn *conn) * just as easy to handle it as part of the main loop. * Save the data and continue processing. */ - if (getBackendKeyData(conn, msgLength)) + if (getBackendKeyData(conn, msgLength, msgBuf)) return; break; case PqMsg_RowDescription: @@ -325,14 +326,14 @@ pqParseInput3(PGconn *conn) * We've already choked for some reason. Just discard * the data till we get to the end of the query. */ - conn->inCursor += msgLength; + msgBuf->cursor += msgLength; } else if (conn->result == NULL || (conn->cmd_queue_head && conn->cmd_queue_head->queryclass == PGQUERY_DESCRIBE)) { /* First 'T' in a query sequence */ - if (getRowDescriptions(conn, msgLength)) + if (getRowDescriptions(conn, msgLength, msgBuf)) return; } else @@ -377,7 +378,7 @@ pqParseInput3(PGconn *conn) } break; case PqMsg_ParameterDescription: - if (getParamDescriptions(conn, msgLength)) + if (getParamDescriptions(conn, msgLength, msgBuf)) return; break; case PqMsg_DataRow: @@ -386,7 +387,7 @@ pqParseInput3(PGconn *conn) conn->result->resultStatus == PGRES_TUPLES_CHUNK)) { /* Read another tuple of a normal query response */ - if (getAnotherTuple(conn, msgLength)) + if (getAnotherTuple(conn, msgLength, msgBuf)) return; } else if (conn->error_result || @@ -397,7 +398,7 @@ pqParseInput3(PGconn *conn) * We've already choked for some reason. Just discard * tuples till we get to the end of the query. */ - conn->inCursor += msgLength; + msgBuf->cursor += msgLength; } else { @@ -405,22 +406,22 @@ pqParseInput3(PGconn *conn) libpq_append_conn_error(conn, "server sent data (\"D\" message) without prior row description (\"T\" message)"); pqSaveErrorResult(conn); /* Discard the unexpected message */ - conn->inCursor += msgLength; + msgBuf->cursor += msgLength; } break; case PqMsg_CopyInResponse: - if (getCopyStart(conn, PGRES_COPY_IN)) + if (getCopyStart(conn, PGRES_COPY_IN, msgBuf)) return; conn->asyncStatus = PGASYNC_COPY_IN; break; case PqMsg_CopyOutResponse: - if (getCopyStart(conn, PGRES_COPY_OUT)) + if (getCopyStart(conn, PGRES_COPY_OUT, msgBuf)) return; conn->asyncStatus = PGASYNC_COPY_OUT; conn->copy_already_done = 0; break; case PqMsg_CopyBothResponse: - if (getCopyStart(conn, PGRES_COPY_BOTH)) + if (getCopyStart(conn, PGRES_COPY_BOTH, msgBuf)) return; conn->asyncStatus = PGASYNC_COPY_BOTH; conn->copy_already_done = 0; @@ -432,7 +433,7 @@ pqParseInput3(PGconn *conn) * only occur if application exits COPY OUT mode too * early. */ - conn->inCursor += msgLength; + msgBuf->cursor += msgLength; break; case PqMsg_CopyDone: @@ -450,15 +451,15 @@ pqParseInput3(PGconn *conn) /* not sure if we will see more, so go to ready state */ conn->asyncStatus = PGASYNC_READY; /* Discard the unexpected message */ - conn->inCursor += msgLength; + msgBuf->cursor += msgLength; break; } /* switch on protocol character */ } /* Successfully consumed this message */ - if (conn->inCursor == conn->inStart + 5 + msgLength) + if (msgBuf->cursor == msgBuf->start + 5 + msgLength) { /* Normal case: parsing agrees with specified length */ - pqParseDone(conn, conn->inCursor); + pqParseDone(conn, msgBuf, msgBuf->cursor); } else if (conn->error_result && conn->status == CONNECTION_BAD) { @@ -473,7 +474,7 @@ pqParseInput3(PGconn *conn) pqSaveErrorResult(conn); conn->asyncStatus = PGASYNC_READY; /* trust the specified message length as what to skip */ - conn->inStart += 5 + msgLength; + msgBuf->start += 5 + msgLength; } } } @@ -516,7 +517,7 @@ handleSyncLoss(PGconn *conn, char id, int msgLength) * (the latter case is not actually used currently). */ static int -getRowDescriptions(PGconn *conn, int msgLength) +getRowDescriptions(PGconn *conn, int msgLength, msg_buffer *msgBuf) { PGresult *result; int nfields; @@ -547,7 +548,7 @@ getRowDescriptions(PGconn *conn, int msgLength) /* parseInput already read the 'T' label and message length. */ /* the next two bytes are the number of fields */ - if (pqGetInt(&(result->numAttributes), 2, conn)) + if (pqGetInt(&(result->numAttributes), 2, conn, msgBuf)) { /* We should not run out of data here, so complain */ errmsg = libpq_gettext("insufficient data in \"T\" message"); @@ -581,13 +582,13 @@ getRowDescriptions(PGconn *conn, int msgLength) int atttypmod; int format; - if (pqGets(&conn->workBuffer, conn) || - pqGetInt(&tableid, 4, conn) || - pqGetInt(&columnid, 2, conn) || - pqGetInt(&typid, 4, conn) || - pqGetInt(&typlen, 2, conn) || - pqGetInt(&atttypmod, 4, conn) || - pqGetInt(&format, 2, conn)) + if (pqGets(&conn->workBuffer, conn, msgBuf) || + pqGetInt(&tableid, 4, conn, msgBuf) || + pqGetInt(&columnid, 2, conn, msgBuf) || + pqGetInt(&typid, 4, conn, msgBuf) || + pqGetInt(&typlen, 2, conn, msgBuf) || + pqGetInt(&atttypmod, 4, conn, msgBuf) || + pqGetInt(&format, 2, conn, msgBuf)) { /* We should not run out of data here, so complain */ errmsg = libpq_gettext("insufficient data in \"T\" message"); @@ -670,7 +671,7 @@ advance_and_error: * Show the message as fully consumed, else pqParseInput3 will overwrite * our error with a complaint about that. */ - conn->inCursor = conn->inStart + 5 + msgLength; + msgBuf->cursor = msgBuf->start + 5 + msgLength; /* * Return zero to allow input parsing to continue. Subsequent "D" @@ -687,7 +688,7 @@ advance_and_error: * (the latter case is not actually used currently). */ static int -getParamDescriptions(PGconn *conn, int msgLength) +getParamDescriptions(PGconn *conn, int msgLength, msg_buffer *msgBuf) { PGresult *result; const char *errmsg = NULL; /* means "out of memory", see below */ @@ -700,7 +701,7 @@ getParamDescriptions(PGconn *conn, int msgLength) /* parseInput already read the 't' label and message length. */ /* the next two bytes are the number of parameters */ - if (pqGetInt(&(result->numParameters), 2, conn)) + if (pqGetInt(&(result->numParameters), 2, conn, msgBuf)) goto not_enough_data; nparams = result->numParameters; @@ -719,7 +720,7 @@ getParamDescriptions(PGconn *conn, int msgLength) { int typid; - if (pqGetInt(&typid, 4, conn)) + if (pqGetInt(&typid, 4, conn, msgBuf)) goto not_enough_data; result->paramDescs[i].typid = typid; } @@ -758,7 +759,7 @@ advance_and_error: * Show the message as fully consumed, else pqParseInput3 will overwrite * our error with a complaint about that. */ - conn->inCursor = conn->inStart + 5 + msgLength; + msgBuf->cursor = msgBuf->start + 5 + msgLength; /* * Return zero to allow input parsing to continue. Essentially, we've @@ -775,7 +776,7 @@ advance_and_error: * (the latter case is not actually used currently). */ static int -getAnotherTuple(PGconn *conn, int msgLength) +getAnotherTuple(PGconn *conn, int msgLength, msg_buffer *msgBuf) { PGresult *result = conn->result; int nfields = result->numAttributes; @@ -786,7 +787,7 @@ getAnotherTuple(PGconn *conn, int msgLength) int i; /* Get the field count and make sure it's what we expect */ - if (pqGetInt(&tupnfields, 2, conn)) + if (pqGetInt(&tupnfields, 2, conn, msgBuf)) { /* We should not run out of data here, so complain */ errmsg = libpq_gettext("insufficient data in \"D\" message"); @@ -818,7 +819,7 @@ getAnotherTuple(PGconn *conn, int msgLength) for (i = 0; i < nfields; i++) { /* get the value length */ - if (pqGetInt(&vlen, 4, conn)) + if (pqGetInt(&vlen, 4, conn, msgBuf)) { /* We should not run out of data here, so complain */ errmsg = libpq_gettext("insufficient data in \"D\" message"); @@ -831,12 +832,12 @@ getAnotherTuple(PGconn *conn, int msgLength) * buffer even if the value is NULL. This allows row processors to * estimate data sizes more easily. */ - rowbuf[i].value = conn->inBuffer + conn->inCursor; + rowbuf[i].value = msgBuf->buffer + msgBuf->cursor; /* Skip over the data value */ if (vlen > 0) { - if (pqSkipnchar(vlen, conn)) + if (pqSkipnchar(vlen, conn, msgBuf)) { /* We should not run out of data here, so complain */ errmsg = libpq_gettext("insufficient data in \"D\" message"); @@ -876,7 +877,7 @@ advance_and_error: * Show the message as fully consumed, else pqParseInput3 will overwrite * our error with a complaint about that. */ - conn->inCursor = conn->inStart + 5 + msgLength; + msgBuf->cursor = msgBuf->start + 5 + msgLength; /* * Return zero to allow input parsing to continue. Subsequent "D" @@ -896,7 +897,7 @@ advance_and_error: * returns EOF if not enough data. */ int -pqGetErrorNotice3(PGconn *conn, bool isError) +pqGetErrorNotice3(PGconn *conn, bool isError, msg_buffer *msgBuf) { PGresult *res = NULL; bool have_position = false; @@ -944,11 +945,11 @@ pqGetErrorNotice3(PGconn *conn, bool isError) */ for (;;) { - if (pqGetc(&id, conn)) + if (pqGetc(&id, conn, msgBuf)) goto fail; if (id == '\0') break; /* terminator found */ - if (pqGets(&workBuf, conn)) + if (pqGets(&workBuf, conn, msgBuf)) goto fail; pqSaveMessageField(res, id, workBuf.data); if (id == PG_DIAG_SQLSTATE) @@ -1242,7 +1243,7 @@ reportErrorPosition(PQExpBuffer msg, const char *query, int loc, int encoding) * * The only caller of reportErrorPosition() is pqBuildErrorMessage3(); it * gets its query from either a PQresultErrorField() or a PGcmdQueueEntry, - * both of which must have fit into conn->inBuffer/outBuffer. So slen fits + * both of which must have fit into inBuffer/outBuffer. So slen fits * inside an int, but we can't assume that (slen * sizeof(int)) fits * inside a size_t. */ @@ -1441,15 +1442,15 @@ reportErrorPosition(PQExpBuffer msg, const char *query, int loc, int encoding) * returns 1 on failure. The error message is filled in. */ int -pqGetNegotiateProtocolVersion3(PGconn *conn) +pqGetNegotiateProtocolVersion3(PGconn *conn, msg_buffer *msgBuf) { int their_version; int num; - if (pqGetInt(&their_version, 4, conn) != 0) + if (pqGetInt(&their_version, 4, conn, msgBuf) != 0) goto eof; - if (pqGetInt(&num, 4, conn) != 0) + if (pqGetInt(&num, 4, conn, msgBuf) != 0) goto eof; /* @@ -1517,7 +1518,7 @@ pqGetNegotiateProtocolVersion3(PGconn *conn) */ for (int i = 0; i < num; i++) { - if (pqGets(&conn->workBuffer, conn)) + if (pqGets(&conn->workBuffer, conn, msgBuf)) { goto eof; } @@ -1551,16 +1552,16 @@ failure: * returns EOF if not enough data. */ static int -getParameterStatus(PGconn *conn) +getParameterStatus(PGconn *conn, msg_buffer *msgBuf) { PQExpBufferData valueBuf; /* Get the parameter name */ - if (pqGets(&conn->workBuffer, conn)) + if (pqGets(&conn->workBuffer, conn, msgBuf)) return EOF; /* Get the parameter value (could be large) */ initPQExpBuffer(&valueBuf); - if (pqGets(&valueBuf, conn)) + if (pqGets(&valueBuf, conn, msgBuf)) { termPQExpBuffer(&valueBuf); return EOF; @@ -1582,7 +1583,7 @@ getParameterStatus(PGconn *conn) * returns EOF if not enough data. */ static int -getBackendKeyData(PGconn *conn, int msgLength) +getBackendKeyData(PGconn *conn, int msgLength, msg_buffer *msgBuf) { int cancel_key_len; @@ -1593,10 +1594,10 @@ getBackendKeyData(PGconn *conn, int msgLength) conn->be_cancel_key_len = 0; } - if (pqGetInt(&(conn->be_pid), 4, conn)) + if (pqGetInt(&(conn->be_pid), 4, conn, msgBuf)) return EOF; - cancel_key_len = 5 + msgLength - (conn->inCursor - conn->inStart); + cancel_key_len = 5 + msgLength - (msgBuf->cursor - msgBuf->start); if (cancel_key_len != 4 && conn->pversion == PG_PROTOCOL(3, 0)) { @@ -1626,7 +1627,7 @@ getBackendKeyData(PGconn *conn, int msgLength) handleFatalError(conn); return 0; } - if (pqGetnchar(conn->be_cancel_key, cancel_key_len, conn)) + if (pqGetnchar(conn->be_cancel_key, cancel_key_len, conn, msgBuf)) { free(conn->be_cancel_key); conn->be_cancel_key = NULL; @@ -1646,7 +1647,7 @@ getBackendKeyData(PGconn *conn, int msgLength) * returns EOF if not enough data. */ static int -getNotify(PGconn *conn) +getNotify(PGconn *conn, msg_buffer *msgBuf) { int be_pid; char *svname; @@ -1654,9 +1655,9 @@ getNotify(PGconn *conn) int extralen; PGnotify *newNotify; - if (pqGetInt(&be_pid, 4, conn)) + if (pqGetInt(&be_pid, 4, conn, msgBuf)) return EOF; - if (pqGets(&conn->workBuffer, conn)) + if (pqGets(&conn->workBuffer, conn, msgBuf)) return EOF; /* must save name while getting extra string */ svname = strdup(conn->workBuffer.data); @@ -1672,7 +1673,7 @@ getNotify(PGconn *conn) handleFatalError(conn); return 0; } - if (pqGets(&conn->workBuffer, conn)) + if (pqGets(&conn->workBuffer, conn, msgBuf)) { free(svname); return EOF; @@ -1717,7 +1718,7 @@ getNotify(PGconn *conn) * parseInput already read the message type and length. */ static int -getCopyStart(PGconn *conn, ExecStatusType copytype) +getCopyStart(PGconn *conn, ExecStatusType copytype, msg_buffer *msgBuf) { PGresult *result; int nfields; @@ -1727,11 +1728,11 @@ getCopyStart(PGconn *conn, ExecStatusType copytype) if (!result) goto failure; - if (pqGetc(&conn->copy_is_binary, conn)) + if (pqGetc(&conn->copy_is_binary, conn, msgBuf)) goto failure; result->binary = conn->copy_is_binary; /* the next two bytes are the number of fields */ - if (pqGetInt(&(result->numAttributes), 2, conn)) + if (pqGetInt(&(result->numAttributes), 2, conn, msgBuf)) goto failure; nfields = result->numAttributes; @@ -1749,7 +1750,7 @@ getCopyStart(PGconn *conn, ExecStatusType copytype) { int format; - if (pqGetInt(&format, 2, conn)) + if (pqGetInt(&format, 2, conn, msgBuf)) goto failure; /* @@ -1773,11 +1774,11 @@ failure: * getReadyForQuery - process ReadyForQuery message */ static int -getReadyForQuery(PGconn *conn) +getReadyForQuery(PGconn *conn, msg_buffer *msgBuf) { char xact_status; - if (pqGetc(&xact_status, conn)) + if (pqGetc(&xact_status, conn, msgBuf)) return EOF; switch (xact_status) { @@ -1805,11 +1806,14 @@ getReadyForQuery(PGconn *conn) * message available, -1 if end of copy, -2 if error. */ static int -getCopyDataMessage(PGconn *conn) +getCopyDataMessage(PGconn *conn, msg_buffer **out_buf) { char id; int msgLength; int avail; + msg_buffer *msgBuf = &conn->inBuffer; + + *out_buf = msgBuf; for (;;) { @@ -1818,31 +1822,31 @@ getCopyDataMessage(PGconn *conn) * callers, we keep returning 0 until the next message is fully * available, even if it is not Copy Data. */ - conn->inCursor = conn->inStart; - if (pqGetc(&id, conn)) + msgBuf->cursor = msgBuf->start; + if (pqGetc(&id, conn, msgBuf)) return 0; - if (pqGetInt(&msgLength, 4, conn)) + if (pqGetInt(&msgLength, 4, conn, msgBuf)) return 0; if (msgLength < 4) { handleSyncLoss(conn, id, msgLength); return -2; } - avail = conn->inEnd - conn->inCursor; + avail = msgBuf->end - msgBuf->cursor; if (avail < msgLength - 4) { /* * Before returning, enlarge the input buffer if needed to hold * the whole message. See notes in parseInput. */ - if (pqCheckInBufferSpace(conn->inCursor + (size_t) msgLength - 4, - conn)) + if (pqCheckMsgBufferSpace(msgBuf->cursor + (size_t) msgLength - 4, + msgBuf, conn)) { /* * Abandon the connection. There's not much else we can * safely do; we can't just ignore the message or we could * miss important changes to the connection state. - * pqCheckInBufferSpace() already reported the error. + * pqCheckMsgBufferSpace() already reported the error. */ handleFatalError(conn); return -2; @@ -1859,15 +1863,15 @@ getCopyDataMessage(PGconn *conn) switch (id) { case PqMsg_NotificationResponse: - if (getNotify(conn)) + if (getNotify(conn, msgBuf)) return 0; break; case PqMsg_NoticeResponse: - if (pqGetErrorNotice3(conn, false)) + if (pqGetErrorNotice3(conn, false, msgBuf)) return 0; break; case PqMsg_ParameterStatus: - if (getParameterStatus(conn)) + if (getParameterStatus(conn, msgBuf)) return 0; break; case PqMsg_CopyData: @@ -1902,7 +1906,7 @@ getCopyDataMessage(PGconn *conn) return -2; /* Drop the processed message and loop around for another */ - pqParseDone(conn, conn->inCursor); + pqParseDone(conn, msgBuf, msgBuf->cursor); } } @@ -1920,6 +1924,7 @@ int pqGetCopyData3(PGconn *conn, char **buffer, int async) { int msgLength; + msg_buffer *msgBuf = &conn->inBuffer; for (;;) { @@ -1928,7 +1933,7 @@ pqGetCopyData3(PGconn *conn, char **buffer, int async) * callers, we keep returning 0 until the next message is fully * available, even if it is not Copy Data. */ - msgLength = getCopyDataMessage(conn); + msgLength = getCopyDataMessage(conn, &msgBuf); if (msgLength < 0) return msgLength; /* end-of-copy or error */ if (msgLength == 0) @@ -1956,17 +1961,17 @@ pqGetCopyData3(PGconn *conn, char **buffer, int async) libpq_append_conn_error(conn, "out of memory"); return -2; } - memcpy(*buffer, &conn->inBuffer[conn->inCursor], msgLength); + memcpy(*buffer, &msgBuf->buffer[msgBuf->cursor], msgLength); (*buffer)[msgLength] = '\0'; /* Add terminating null */ /* Mark message consumed */ - pqParseDone(conn, conn->inCursor + msgLength); + pqParseDone(conn, msgBuf, msgBuf->cursor + msgLength); return msgLength; } /* Empty, so drop it and loop around for another */ - pqParseDone(conn, conn->inCursor); + pqParseDone(conn, msgBuf, msgBuf->cursor); } } @@ -2031,6 +2036,7 @@ pqGetlineAsync3(PGconn *conn, char *buffer, int bufsize) { int msgLength; int avail; + msg_buffer *msgBuf = &conn->inBuffer; if (conn->asyncStatus != PGASYNC_COPY_OUT && conn->asyncStatus != PGASYNC_COPY_BOTH) @@ -2042,7 +2048,7 @@ pqGetlineAsync3(PGconn *conn, char *buffer, int bufsize) * even if it is not Copy Data. This should keep PQendcopy from blocking. * (Note: unlike pqGetCopyData3, we do not change asyncStatus here.) */ - msgLength = getCopyDataMessage(conn); + msgLength = getCopyDataMessage(conn, &msgBuf); if (msgLength < 0) return -1; /* end-of-copy or error */ if (msgLength == 0) @@ -2054,14 +2060,14 @@ pqGetlineAsync3(PGconn *conn, char *buffer, int bufsize) * conn->copy_already_done to remember how much of the row was already * returned to the caller. */ - conn->inCursor += conn->copy_already_done; + msgBuf->cursor += conn->copy_already_done; avail = msgLength - 4 - conn->copy_already_done; if (avail <= bufsize) { /* Able to consume the whole message */ - memcpy(buffer, &conn->inBuffer[conn->inCursor], avail); + memcpy(buffer, &msgBuf->buffer[msgBuf->cursor], avail); /* Mark message consumed */ - conn->inStart = conn->inCursor + avail; + msgBuf->start = msgBuf->cursor + avail; /* Reset state for next time */ conn->copy_already_done = 0; return avail; @@ -2069,7 +2075,7 @@ pqGetlineAsync3(PGconn *conn, char *buffer, int bufsize) else { /* We must return a partial message */ - memcpy(buffer, &conn->inBuffer[conn->inCursor], bufsize); + memcpy(buffer, &msgBuf->buffer[msgBuf->cursor], bufsize); /* The message is NOT consumed from libpq's buffer */ conn->copy_already_done += bufsize; return bufsize; @@ -2186,6 +2192,7 @@ pqFunctionCall3(PGconn *conn, Oid fnid, int msgLength; int avail; int i; + msg_buffer *msgBuf = &conn->inBuffer; /* already validated by PQnfn */ Assert(conn->pipelineStatus == PQ_PIPELINE_OFF); @@ -2243,10 +2250,10 @@ pqFunctionCall3(PGconn *conn, Oid fnid, */ needInput = true; - conn->inCursor = conn->inStart; - if (pqGetc(&id, conn)) + msgBuf->cursor = msgBuf->start; + if (pqGetc(&id, conn, msgBuf)) continue; - if (pqGetInt(&msgLength, 4, conn)) + if (pqGetInt(&msgLength, 4, conn, msgBuf)) continue; /* @@ -2269,21 +2276,21 @@ pqFunctionCall3(PGconn *conn, Oid fnid, * Can't process if message body isn't all here yet. */ msgLength -= 4; - avail = conn->inEnd - conn->inCursor; + avail = msgBuf->end - msgBuf->cursor; if (avail < msgLength) { /* * Before looping, enlarge the input buffer if needed to hold the * whole message. See notes in parseInput. */ - if (pqCheckInBufferSpace(conn->inCursor + (size_t) msgLength, - conn)) + if (pqCheckMsgBufferSpace(msgBuf->cursor + (size_t) msgLength, + msgBuf, conn)) { /* * Abandon the connection. There's not much else we can * safely do; we can't just ignore the message or we could * miss important changes to the connection state. - * pqCheckInBufferSpace() already reported the error. + * pqCheckMsgBufferSpace() already reported the error. */ handleFatalError(conn); break; @@ -2299,13 +2306,13 @@ pqFunctionCall3(PGconn *conn, Oid fnid, switch (id) { case PqMsg_FunctionCallResponse: - if (pqGetInt(actual_result_len, 4, conn)) + if (pqGetInt(actual_result_len, 4, conn, msgBuf)) continue; if (*actual_result_len != -1) { if (result_is_int) { - if (pqGetInt(result_buf, *actual_result_len, conn)) + if (pqGetInt(result_buf, *actual_result_len, conn, msgBuf)) continue; } else @@ -2323,7 +2330,7 @@ pqFunctionCall3(PGconn *conn, Oid fnid, if (pqGetnchar(result_buf, *actual_result_len, - conn)) + conn, msgBuf)) continue; } } @@ -2331,26 +2338,26 @@ pqFunctionCall3(PGconn *conn, Oid fnid, status = PGRES_COMMAND_OK; break; case PqMsg_ErrorResponse: - if (pqGetErrorNotice3(conn, true)) + if (pqGetErrorNotice3(conn, true, msgBuf)) continue; status = PGRES_FATAL_ERROR; break; case PqMsg_NotificationResponse: /* handle notify and go back to processing return values */ - if (getNotify(conn)) + if (getNotify(conn, msgBuf)) continue; break; case PqMsg_NoticeResponse: /* handle notice and go back to processing return values */ - if (pqGetErrorNotice3(conn, false)) + if (pqGetErrorNotice3(conn, false, msgBuf)) continue; break; case PqMsg_ReadyForQuery: - if (getReadyForQuery(conn)) + if (getReadyForQuery(conn, msgBuf)) continue; /* consume the message */ - pqParseDone(conn, conn->inStart + 5 + msgLength); + pqParseDone(conn, msgBuf, msgBuf->start + 5 + msgLength); /* * If we already have a result object (probably an error), use @@ -2378,7 +2385,7 @@ pqFunctionCall3(PGconn *conn, Oid fnid, /* and we're out */ return pqPrepareAsyncResult(conn); case PqMsg_ParameterStatus: - if (getParameterStatus(conn)) + if (getParameterStatus(conn, msgBuf)) continue; break; default: @@ -2391,7 +2398,7 @@ pqFunctionCall3(PGconn *conn, Oid fnid, * (so message tracing wouldn't work), but trust the specified * message length as what to skip. */ - conn->inStart += 5 + msgLength; + msgBuf->start += 5 + msgLength; return pqPrepareAsyncResult(conn); } @@ -2403,7 +2410,7 @@ pqFunctionCall3(PGconn *conn, Oid fnid, return pqPrepareAsyncResult(conn); /* Completed parsing this message, keep going */ - pqParseDone(conn, conn->inStart + 5 + msgLength); + pqParseDone(conn, msgBuf, msgBuf->start + 5 + msgLength); needInput = false; } diff --git a/src/interfaces/libpq/fe-secure-openssl.c b/src/interfaces/libpq/fe-secure-openssl.c index 243ed240087..85115643f17 100644 --- a/src/interfaces/libpq/fe-secure-openssl.c +++ b/src/interfaces/libpq/fe-secure-openssl.c @@ -258,7 +258,8 @@ pgtls_bytes_pending(PGconn *conn) * around in the caller to call pgtls_bytes_pending() again. Throw an * error rather than complicate the code in that way, because * SSL_read() should be bounded to the size of a single TLS record, - * and conn->inBuffer can't currently go past INT_MAX in size anyway. + * and conn->inBuffer.buffer can't currently go past INT_MAX in size + * anyway. */ libpq_append_conn_error(conn, "OpenSSL reports INT_MAX bytes pending"); return -1; diff --git a/src/interfaces/libpq/libpq-int.h b/src/interfaces/libpq/libpq-int.h index 9261fe0d309..673648fa154 100644 --- a/src/interfaces/libpq/libpq-int.h +++ b/src/interfaces/libpq/libpq-int.h @@ -366,6 +366,15 @@ typedef struct pg_conn_host * found in password file. */ } pg_conn_host; +typedef struct msg_buffer +{ + char *buffer; /* currently allocated buffer */ + int bufSize; /* allocated size of buffer */ + int start; /* offset to first unconsumed data in buffer */ + int cursor; /* next byte to tentatively consume */ + int end; /* offset to first position after avail data */ +} msg_buffer; + /* * PGconn stores all the state data associated with a single connection * to a backend. @@ -572,11 +581,8 @@ struct pg_conn * pqCheck{In,Out}BufferSpace(), but also a careful audit of all libpq * code that uses ints during size calculations. */ - char *inBuffer; /* currently allocated buffer */ - int inBufSize; /* allocated size of buffer */ - int inStart; /* offset to first unconsumed data in buffer */ - int inCursor; /* next byte to tentatively consume */ - int inEnd; /* offset to first position after avail data */ + msg_buffer inBuffer; + /* Buffer for data not yet sent to backend */ char *outBuffer; /* currently allocated buffer */ @@ -769,10 +775,10 @@ extern PGresult *PQnfn(PGconn *conn, int fnid, int *result_buf, int buf_size, extern char *pqBuildStartupPacket3(PGconn *conn, int *packetlen, const PQEnvironmentOption *options); extern void pqParseInput3(PGconn *conn); -extern int pqGetErrorNotice3(PGconn *conn, bool isError); +extern int pqGetErrorNotice3(PGconn *conn, bool isError, msg_buffer *msgBuf); extern void pqBuildErrorMessage3(PQExpBuffer msg, const PGresult *res, PGVerbosity verbosity, PGContextVisibility show_context); -extern int pqGetNegotiateProtocolVersion3(PGconn *conn); +extern int pqGetNegotiateProtocolVersion3(PGconn *conn, msg_buffer *msgBuf); extern int pqGetCopyData3(PGconn *conn, char **buffer, int async); extern int pqGetline3(PGconn *conn, char *s, int maxlen); extern int pqGetlineAsync3(PGconn *conn, char *buffer, int bufsize); @@ -795,17 +801,17 @@ extern int PQsendCancelRequest(PGconn *cancelConn); * necessarily any error. */ extern int pqCheckOutBufferSpace(size_t bytes_needed, PGconn *conn); -extern int pqCheckInBufferSpace(size_t bytes_needed, PGconn *conn); -extern void pqParseDone(PGconn *conn, int newInStart); -extern int pqGetc(char *result, PGconn *conn); +extern int pqCheckMsgBufferSpace(size_t bytes_needed, msg_buffer *msgBuffer, PGconn *conn); +extern void pqParseDone(PGconn *conn, msg_buffer *msgBuf, int newInStart); +extern int pqGetc(char *result, PGconn *conn, msg_buffer *msgBuf); extern int pqPutc(char c, PGconn *conn); -extern int pqGets(PQExpBuffer buf, PGconn *conn); -extern int pqGets_append(PQExpBuffer buf, PGconn *conn); +extern int pqGets(PQExpBuffer buf, PGconn *conn, msg_buffer *inBuf); +extern int pqGets_append(PQExpBuffer buf, PGconn *conn, msg_buffer *inBuf); extern int pqPuts(const char *s, PGconn *conn); -extern int pqGetnchar(void *s, size_t len, PGconn *conn); -extern int pqSkipnchar(size_t len, PGconn *conn); +extern int pqGetnchar(void *s, size_t len, PGconn *conn, msg_buffer *msgBuf); +extern int pqSkipnchar(size_t len, PGconn *conn, msg_buffer *msgBuf); extern int pqPutnchar(const void *s, size_t len, PGconn *conn); -extern int pqGetInt(int *result, size_t bytes, PGconn *conn); +extern int pqGetInt(int *result, size_t bytes, PGconn *conn, msg_buffer *msgBuf); extern int pqPutInt(int value, size_t bytes, PGconn *conn); extern int pqPutMsgStart(char msg_type, PGconn *conn); extern int pqPutMsgEnd(PGconn *conn); diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index a7a8dbdd23f..c42dd48bb23 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -3965,6 +3965,7 @@ missing_cache_key mix_data_t mode_t movedb_failure_params +msg_buffer multirange_bsearch_comparison multirange_unnest_fctx mxact -- That's all, folks. May the source be with you.