From c9acce9bc8e3f6baa940c8c91e49326ae6ad95d2 Mon Sep 17 00:00:00 2001 From: Jakub Wartak Date: Tue, 4 Aug 2026 12:00:23 +0200 Subject: [PATCH v06082026 08/10] libpq / pg_basebackup: add PQgetCopyDataInternalBuf() for heavy transfers At high data transfer rates(>10Gbit/s), the primary bottlneck in pg_basebackup becomes the 2-nd memory copy done in the userspace by libpq itself (the first one is the kernel itself copying the data from network to the userspace socket). Add optimized variant of PQgetCopyData(): PQgetCopyDataInternalBuf() which avoids repeated memory allocation and copying. The other major differences are: - we do not null terminate the string (as that is not used anyway) - the data is only valid untill the next call (so by that time it has to be consumed, but we can use internal libpq buffer until we receive more) Author: Jakub Wartak --- src/bin/pg_basebackup/pg_basebackup.c | 10 +++-- src/interfaces/libpq/exports.txt | 1 + src/interfaces/libpq/fe-exec.c | 26 ++++++++++++ src/interfaces/libpq/fe-protocol3.c | 58 +++++++++++++++++++++++++++ src/interfaces/libpq/libpq-fe.h | 1 + src/interfaces/libpq/libpq-int.h | 1 + 6 files changed, 94 insertions(+), 3 deletions(-) diff --git a/src/bin/pg_basebackup/pg_basebackup.c b/src/bin/pg_basebackup/pg_basebackup.c index af579a967a0..708b0727941 100644 --- a/src/bin/pg_basebackup/pg_basebackup.c +++ b/src/bin/pg_basebackup/pg_basebackup.c @@ -1034,7 +1034,13 @@ ReceiveCopyData(PGconn *conn, WriteDataCallback callback, int r; char *copybuf; - r = PQgetCopyData(conn, ©buf, 0); + /* + * Use the no-copy optimization: copybuf points directly into libpq's + * receive buffer and stays valid until the next call, which is all the + * callback needs (this avoids memory copy which hurts at high transfer + * rates). + */ + r = PQgetCopyDataInternalBuf(conn, ©buf, 0); if (r == -1) { /* End of chunk. */ @@ -1048,8 +1054,6 @@ ReceiveCopyData(PGconn *conn, WriteDataCallback callback, pg_fatal("background process terminated unexpectedly"); (*callback) (r, copybuf, callback_data); - - PQfreemem(copybuf); } } diff --git a/src/interfaces/libpq/exports.txt b/src/interfaces/libpq/exports.txt index 1e3d5bd5867..e02b92b285e 100644 --- a/src/interfaces/libpq/exports.txt +++ b/src/interfaces/libpq/exports.txt @@ -211,3 +211,4 @@ PQdefaultAuthDataHook 208 PQfullProtocolVersion 209 appendPQExpBufferVA 210 PQgetThreadLock 211 +PQgetCopyDataInternalBuf 212 diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c index 7b8edacbfde..9e584c222bf 100644 --- a/src/interfaces/libpq/fe-exec.c +++ b/src/interfaces/libpq/fe-exec.c @@ -2844,6 +2844,32 @@ PQgetCopyData(PGconn *conn, char **buffer, int async) return pqGetCopyData3(conn, buffer, async); } +/* + * PQgetCopyDataInternalBuf - variant of PQgetCopyData which bypasses memcpy() + * for high data transfer rates. + * + * Set *buffer to point directly into the internal receive buffer rather + * allocate and copy memory on every call. The *buffer is valid only until the + * next PQgetCopyDataInternalBuf/PQgetCopyData call on this connection, and must + * not be passed to PQfreemem(). The returned payload is NOT null-terminated. + * + * Otherwise it is pretty much the same as the original PQgetCopyData. + */ +int +PQgetCopyDataInternalBuf(PGconn *conn, char **buffer, int async) +{ + *buffer = NULL; /* for all failure cases */ + if (!conn) + return -2; + if (conn->asyncStatus != PGASYNC_COPY_OUT && + conn->asyncStatus != PGASYNC_COPY_BOTH) + { + libpq_append_conn_error(conn, "no COPY in progress"); + return -2; + } + return pqGetCopyDataInternalBuf3(conn, buffer, async); +} + /* * PQgetline - gets a newline-terminated string from the backend. * diff --git a/src/interfaces/libpq/fe-protocol3.c b/src/interfaces/libpq/fe-protocol3.c index 9d6a285fb28..18ba62b883f 100644 --- a/src/interfaces/libpq/fe-protocol3.c +++ b/src/interfaces/libpq/fe-protocol3.c @@ -2000,6 +2000,64 @@ pqGetCopyData3(PGconn *conn, char **buffer, int async) } } +/* + * pqGetCopyDataInternalBuf3 - like pqGetCopyData3, but avoids malloc and + * memory copying. + * + * Instead of allocating fresh buffer and copying the CopyData payload into + * that new memory, this sets *buffer directly into conn->inBuffer and returns + * its length. The message is marked consumed, and the returned pointer stays + * valid only until the next libpq call that reads from the socket (e.g. using + * pgReadData() or this call). Called must be done with the data processing + * before calling this again. + * + * Differences between thnis and the orginal pgGetCopyData3() are: + * - the caller must not free *buffer + * - the payload is not null-terminated + * + * The main advantage of this call is that it avoids memory copy for for + * high-throughput COPY. + */ +int +pqGetCopyDataInternalBuf3(PGconn *conn, char **buffer, int async) +{ + int msgLength; + + for (;;) + { + /* Collect the next input message; see pqGetCopyData3 for details. */ + msgLength = getCopyDataMessage(conn); + if (msgLength < 0) + return msgLength; /* end-of-copy or error */ + if (msgLength == 0) + { + /* Don't block if async read requested */ + if (async) + return 0; + /* Need to load more data */ + if (pqWait(true, false, conn) || + pqReadData(conn) < 0) + return -2; + continue; + } + + msgLength -= 4; + if (msgLength > 0) + { + /* Just use libpq internal input buffer */ + *buffer = &conn->inBuffer[conn->inCursor]; + + /* Mark message consumed */ + pqParseDone(conn, conn->inCursor + msgLength); + + return msgLength; + } + + /* Empty, so drop it and loop around for another */ + pqParseDone(conn, conn->inCursor); + } +} + /* * PQgetline - gets a newline-terminated string from the backend. * diff --git a/src/interfaces/libpq/libpq-fe.h b/src/interfaces/libpq/libpq-fe.h index 8ecb9b4a4c7..350019864a0 100644 --- a/src/interfaces/libpq/libpq-fe.h +++ b/src/interfaces/libpq/libpq-fe.h @@ -564,6 +564,7 @@ extern PGnotify *PQnotifies(PGconn *conn); extern int PQputCopyData(PGconn *conn, const char *buffer, int nbytes); extern int PQputCopyEnd(PGconn *conn, const char *errormsg); extern int PQgetCopyData(PGconn *conn, char **buffer, int async); +extern int PQgetCopyDataInternalBuf(PGconn *conn, char **buffer, int async); /* Deprecated routines for copy in/out */ extern int PQgetline(PGconn *conn, char *buffer, int length); diff --git a/src/interfaces/libpq/libpq-int.h b/src/interfaces/libpq/libpq-int.h index 3f921207a14..234524d947e 100644 --- a/src/interfaces/libpq/libpq-int.h +++ b/src/interfaces/libpq/libpq-int.h @@ -779,6 +779,7 @@ extern void pqBuildErrorMessage3(PQExpBuffer msg, const PGresult *res, PGVerbosity verbosity, PGContextVisibility show_context); extern int pqGetNegotiateProtocolVersion3(PGconn *conn); extern int pqGetCopyData3(PGconn *conn, char **buffer, int async); +extern int pqGetCopyDataInternalBuf3(PGconn *conn, char **buffer, int async); extern int pqGetline3(PGconn *conn, char *s, int maxlen); extern int pqGetlineAsync3(PGconn *conn, char *buffer, int bufsize); extern int pqEndcopy3(PGconn *conn); -- 2.43.0