[RFC PATCH cluster 12/12] allow large ipc responses by sending the response message in chunks

[Daniel Kral <d.kral@proxmox.com>]

If the response to a pmxcfs IPC request is larger than the client's
buffer size of MAX_MSG_SIZE (currently, 1 MiB), the IPC server responds
with an -EMSGSIZE. This can happen for some IPC requests in huge cluster
setups, such as CFS_IPC_GET_GUEST_LIST, CFS_IPC_GET_RRD_DUMP, and
CFS_IPC_GET_GUEST_CONFIG_PROPERTIES.

Allow the IPC client to receive and the IPC server to respond with
larger messages by sending the response body as several chunks to the
IPC client, which reconstructs them to the original response message.

This introduces a new request header, which makes the IPC server aware
that the IPC client can handle receiving multiple message chunks, and a
new response header, which tells the IPC client the response state.
Currently, the response state tells the IPC client when to expect more
message chunks (CFS_IPC_RES_FLAG_CHUNK) and the last message chunk
(CFS_IPC_RES_FLAG_CHUNK_LAST).

The new request header is introduced as a form to indicate which
response types the IPC client can handle from the IPC server. As an
upgrade of the pve-cluster package will not restart all PVE daemons,
which still use the old PVE::IPCC and therefore will neither send the
new request header nor handle the new response header, the IPC server
must respond without the new response header and send a maximum of one
message.

The ms_timeout variable has been replaced by the default timeout value
that corresponds to -1 with qb_ipcc_sendv_recv(), which is not handled
the same with qb_ipcc_recv().

Signed-off-by: Daniel Kral <d.kral@proxmox.com>
---
I've found that if only the pve-cluster package is upgraded, which
contains the `pmxcfs` binary (server) and `PVE::IPCC` module (client),
then this doesn't (always?) trigger e.g. pve-firewall's and
pve-ha-manager's postinst, which reloads/restarts the PVE daemon.
However, it seems like the other *pve-cluster* packages trigger that.

For safety, I have implemented the request header to handle these kind
of scenarios, so we ensure that the IPC server will never respond with
something the IPC client cannot handle. This can be reused if there's
any other changes to the pmxcfs IPC message protocol, e.g. compressed
messages, etc.

However, feedback is always appreciated about such designs.


 src/PVE/IPCC.xs             |  79 +++++++++++++++++++++-----
 src/PVE/Makefile            |   1 +
 src/pmxcfs/cfs-ipc-common.h |  56 ++++++++++++++++++
 src/pmxcfs/server.c         | 109 ++++++++++++++++++++++++++++--------
 4 files changed, 206 insertions(+), 39 deletions(-)
 create mode 100644 src/pmxcfs/cfs-ipc-common.h

diff --git a/src/PVE/IPCC.xs b/src/PVE/IPCC.xs
index 626971c..b9da1b6 100644
--- a/src/PVE/IPCC.xs
+++ b/src/PVE/IPCC.xs
@@ -29,6 +29,8 @@
 #include <qb/qblog.h>
 #include <qb/qbipcc.h>
 
+#include "cfs-ipc-common.h"
+
 #define RESTART_FLAG_FILE "/run/pve-cluster/cfs-restart-flag"
 #define RESTART_GRACE_PERIOD 10
 
@@ -37,6 +39,8 @@
 #define PCS_SERVICE1 1
 #define MAX_MSG_SIZE (8192*128)
 
+#define MAX_MSG_RECV_WAIT_MS 2000
+
 static qb_ipcc_connection_t *conn;
 static pid_t conn_pid;
 
@@ -125,18 +129,19 @@ recache_connection:
 	int iov_len = 2;
 	struct iovec iov[iov_len];
 
-	struct qb_ipc_request_header req_header;
+	struct cfs_ipc_request_header_t req_header;
 
-	req_header.id = msgid;
-	req_header.size = sizeof(req_header) + len;
+	req_header.qb_header.id = msgid;
+	req_header.qb_header.size = sizeof(req_header) + len;
+	req_header.magic = CFS_IPC_MAGIC;
+	req_header.version = CFS_IPC_CURRENT_VERSION;
 
 	iov[0].iov_base = (char *)&req_header;
 	iov[0].iov_len = sizeof(req_header);
 	iov[1].iov_base = dataptr;
 	iov[1].iov_len = len;
 
-	int32_t ms_timeout = -1; // fixme:     
-	int res = qb_ipcc_sendv_recv(conn, iov, iov_len, ipcbuffer, sizeof(ipcbuffer), ms_timeout);
+	int res = qb_ipcc_sendv(conn, iov, iov_len);
 	if (res < 0) {
 		qb_ipcc_disconnect(conn);
 		conn = NULL;
@@ -146,25 +151,69 @@ recache_connection:
 			goto recache_connection;
 		}
 		errno = -res;
-		XSRETURN_UNDEF;
+		goto out_err;
 	}
 
-	struct qb_ipc_response_header *res_header;
+	SV *result = newSVpvs("");
+	struct cfs_ipc_response_header_t *res_header = (struct cfs_ipc_response_header_t *)ipcbuffer;
+	struct qb_ipc_response_header *qb_header = &res_header->qb_header;
 
-	res_header = (struct qb_ipc_response_header *)ipcbuffer;
-	int dsize = res_header->size - sizeof(struct qb_ipc_response_header);
+	int index = 0;
+	do {
+		res = qb_ipcc_recv(conn, ipcbuffer, sizeof(ipcbuffer), MAX_MSG_RECV_WAIT_MS);
+		if (res < 0) {
+			errno = -res;
+			goto out_drop_conn;
+		}
 
-	if (res_header->error < 0) {
-		errno = -res_header->error;
-		XSRETURN_UNDEF;
+		if (qb_header->id != msgid ||
+		    qb_header->size != res ||
+		    res_header->index != index) {
+			errno = -EBADMSG;
+			goto out_drop_conn;
+		}
+
+		int dsize = qb_header->size - sizeof(*res_header);
+
+		sv_catpvn(result, ipcbuffer + sizeof(*res_header), dsize);
+
+		if (res_header->flags & CFS_IPC_RES_FLAG_CHUNK_LAST) {
+			/* Either the last part of a multi-message response or
+			 * a standalone message response. There are no more
+			 * incoming message chunks to expect. */
+			break;
+		} else if (res_header->flags & CFS_IPC_RES_FLAG_CHUNK) {
+			/* An intermediate part of a multi-message response.
+			 * More incoming message chunks are expected. */
+			index++;
+			continue;
+		} else {
+			/* This should not happen */
+			errno = -EPROTO;
+			goto out_drop_conn;
+		}
+	} while (1);
+
+	if (qb_header->error < 0) {
+		errno = -qb_header->error;
+		goto out_drop_result;
 	} else {
 		errno = 0;
-		if (dsize > 0) {
-			RETVAL = newSVpv(ipcbuffer + sizeof(struct qb_ipc_response_header), dsize);
+		if (SvCUR(result) > 0) {
+			RETVAL = result;
+			goto done;
 		} else {
-			XSRETURN_UNDEF;
+			goto out_drop_result;
 		}
 	}
+out_drop_conn:
+	qb_ipcc_disconnect(conn);
+	conn = NULL;
+out_drop_result:
+	SvREFCNT_dec(result);
+out_err:
+	XSRETURN_UNDEF;
+done:
 }
 OUTPUT: RETVAL
 
diff --git a/src/PVE/Makefile b/src/PVE/Makefile
index 0ccd2f8..636f494 100644
--- a/src/PVE/Makefile
+++ b/src/PVE/Makefile
@@ -45,6 +45,7 @@ CC=gcc
 CFLAGS += -fPIC -Wl,-z,relro -Wall -Werror -Wno-strict-aliasing -g -O2 -shared
 CFLAGS += $(shell pkg-config --cflags libqb)
 CFLAGS += $(shell perl -MExtUtils::Embed -e perl_inc)
+CFLAGS += -I../pmxcfs
 LDFLAGS = $(shell pkg-config --libs libqb)
 
 .c.o:
diff --git a/src/pmxcfs/cfs-ipc-common.h b/src/pmxcfs/cfs-ipc-common.h
new file mode 100644
index 0000000..5c9c226
--- /dev/null
+++ b/src/pmxcfs/cfs-ipc-common.h
@@ -0,0 +1,56 @@
+/*
+  Copyright (C) 2026 Proxmox Server Solutions GmbH
+
+  This program is free software: you can redistribute it and/or modify
+  it under the terms of the GNU Affero General Public License as published by
+  the Free Software Foundation, either version 3 of the License, or
+  (at your option) any later version.
+
+  This program is distributed in the hope that it will be useful,
+  but WITHOUT ANY WARRANTY; without even the implied warranty of
+  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+  GNU Affero General Public License for more details.
+
+  You should have received a copy of the GNU Affero General Public License
+  along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+  Author: Daniel Kral <d.kral@proxmox.com>
+
+*/
+
+#ifndef _PVE_IPC_COMMON_H_
+#define _PVE_IPC_COMMON_H_
+
+#include <qb/qbipc_common.h>
+
+#define CFS_IPC_MAGIC 0xBC24
+
+#define CFS_IPC_VERSION_NO_HEADER_SUPPORT 0
+
+#define CFS_IPC_MIN_VERSION_HEADER_SUPPORT 1
+#define CFS_IPC_MIN_VERSION_CHUNK_SUPPORT CFS_IPC_MIN_VERSION_HEADER_SUPPORT
+
+#define CFS_IPC_CURRENT_VERSION 1
+
+struct cfs_ipc_request_header_t {
+    struct qb_ipc_request_header qb_header;
+    uint32_t magic __attribute__((aligned(8)));
+    uint32_t version __attribute__((aligned(8)));
+} __attribute__((aligned(8)));
+
+enum cfs_ipc_response_flags {
+    /* An intermediate response message chunk. Receivers should expect to
+     * receive more messages afterwards. */
+    CFS_IPC_RES_FLAG_CHUNK = 1 << 1,
+    /* A standalone response message or the last response message chunk.
+     * Receivers can expect to not receive any message afterwards. */
+    CFS_IPC_RES_FLAG_CHUNK_LAST = 1 << 2,
+};
+
+struct cfs_ipc_response_header_t {
+    struct qb_ipc_response_header qb_header;
+    uint32_t flags __attribute__((aligned(8)));
+    uint32_t index __attribute__((aligned(8)));
+} __attribute__((aligned(8)));
+
+#endif /* _PVE_IPC_COMMON_H_ */
diff --git a/src/pmxcfs/server.c b/src/pmxcfs/server.c
index 46ad381..839327c 100644
--- a/src/pmxcfs/server.c
+++ b/src/pmxcfs/server.c
@@ -37,6 +37,7 @@
 
 #include <glib.h>
 
+#include "cfs-ipc-common.h"
 #include "cfs-ipc-ops.h"
 #include "cfs-utils.h"
 #include "logger.h"
@@ -431,17 +432,31 @@ static int32_t s1_msg_handle_request(
     return result;
 }
 
-static void s1_msg_send_response(qb_ipcs_connection_t *c, int32_t request_id, int32_t result) {
+static void s1_msg_send_response(
+    qb_ipcs_connection_t *c, int32_t request_id, uint32_t request_version, int32_t result
+) {
     int iov_len = 2;
     struct iovec iov[iov_len];
-    struct qb_ipc_response_header res_header;
+    struct cfs_ipc_response_header_t res_header;
     int32_t max_msg_size = qb_ipcs_connection_get_buffer_size(c);
 
-    if (sizeof(res_header) + outbuf->len > max_msg_size) {
-        cfs_critical(
-            "response for id %d is too large: %ld > %d", request_id, outbuf->len, max_msg_size
-        );
-        result = -EMSGSIZE;
+    /* Only send the extended response header if the client supports it */
+    if (request_version >= CFS_IPC_MIN_VERSION_HEADER_SUPPORT) {
+        iov[0].iov_base = &res_header;
+        iov[0].iov_len = sizeof(res_header);
+    } else {
+        /* Otherwise send only the QB response header... */
+        iov[0].iov_base = &res_header.qb_header;
+        iov[0].iov_len = sizeof(res_header.qb_header);
+
+        /* ...and restrict response length to the maximum buffer size as the
+         * client doesn't support receiving message chunks. */
+        if (sizeof(res_header) + outbuf->len > max_msg_size) {
+            cfs_critical(
+                "response for id %d is too large: %ld > %d", request_id, outbuf->len, max_msg_size
+            );
+            result = -EMSGSIZE;
+        }
     }
 
     /* Do not send the response body in case of an error */
@@ -449,24 +464,58 @@ static void s1_msg_send_response(qb_ipcs_connection_t *c, int32_t request_id, in
         g_string_truncate(outbuf, 0);
     }
 
-    iov[0].iov_base = (char *)&res_header;
-    iov[0].iov_len = sizeof(res_header);
-    iov[1].iov_base = outbuf->str;
-    iov[1].iov_len = outbuf->len;
+    res_header.qb_header.id = request_id;
+    res_header.qb_header.error = result;
+    res_header.flags = 0;
 
-    res_header.id = request_id;
-    res_header.size = iov[0].iov_len + iov[1].iov_len;
-    res_header.error = result;
+    int index = 0;
+    size_t msg_offset, msg_size;
+    size_t max_chunk_size = max_msg_size - iov[0].iov_len;
+    do {
+        res_header.index = index;
 
-    ssize_t res = qb_ipcs_response_sendv(c, iov, iov_len);
-    if (res < 0) {
-        cfs_critical("qb_ipcs_response_send: %s", strerror(errno));
-        qb_ipcs_disconnect(c);
-    }
+        msg_offset = max_chunk_size * index;
+        msg_size = outbuf->len - msg_offset;
+
+        if (msg_size > max_chunk_size) {
+            msg_size = max_chunk_size;
+            res_header.flags |= CFS_IPC_RES_FLAG_CHUNK;
+        } else {
+            res_header.flags |= CFS_IPC_RES_FLAG_CHUNK_LAST;
+        }
+
+        iov[1].iov_base = outbuf->str + msg_offset;
+        iov[1].iov_len = msg_size;
+
+        res_header.qb_header.size = iov[0].iov_len + iov[1].iov_len;
+
+        ssize_t res = qb_ipcs_response_sendv(c, iov, iov_len);
+        if (res == -EAGAIN) {
+            /* The client is busy (e.g. the previous message chunk hasn't been
+             * read yet). Try to send the same message again after some small
+             * backoff time */
+            cfs_debug("msg %d chunk %d send response is -EAGAIN, try again", request_id, index);
+            usleep(1000);
+            continue;
+        } else if (res < 0) {
+            cfs_debug("msg %d chunk %d send response is %ld, abort", request_id, index, res);
+            cfs_critical("qb_ipcs_response_send: %s", strerror(errno));
+            qb_ipcs_disconnect(c);
+            break;
+        } else {
+            cfs_debug("msg %d chunk %d with %ld bytes sent successfully", request_id, index, res);
+
+            if (res_header.flags & CFS_IPC_RES_FLAG_CHUNK_LAST) {
+                break;
+            }
+
+            index++;
+        }
+    } while (1);
 }
 
 static int32_t s1_msg_process_fn(qb_ipcs_connection_t *c, void *data, size_t size) {
-    struct qb_ipc_request_header *req_pt = (struct qb_ipc_request_header *)data;
+    struct cfs_ipc_request_header_t *req_pt = (struct cfs_ipc_request_header_t *)data;
 
     struct s1_context *ctx = (struct s1_context *)qb_ipcs_context_get(c);
 
@@ -476,15 +525,27 @@ static int32_t s1_msg_process_fn(qb_ipcs_connection_t *c, void *data, size_t siz
         return 0;
     }
 
-    int32_t request_id = req_pt->id;
-    int32_t request_size = req_pt->size - sizeof(struct qb_ipc_request_header);
-    data = (uint8_t *)data + sizeof(struct qb_ipc_request_header);
+    int32_t request_id = req_pt->qb_header.id;
+    int32_t request_size = req_pt->qb_header.size;
+    uint32_t request_version = CFS_IPC_VERSION_NO_HEADER_SUPPORT;
     cfs_debug("process msg:%d, size:%d", request_id, request_size);
 
+    /* There might be older IPC clients, which do not send the new, versioned
+     * request header yet, so we need to handle the request data here
+     * differently since the request headers have different fields and sizes. */
+    if (request_size >= sizeof(struct cfs_ipc_request_header_t) && req_pt->magic == CFS_IPC_MAGIC) {
+        data = (uint8_t *)data + sizeof(struct cfs_ipc_request_header_t);
+        request_size -= sizeof(struct cfs_ipc_request_header_t);
+        request_version = req_pt->version;
+    } else {
+        data = (uint8_t *)data + sizeof(struct qb_ipc_request_header);
+        request_size -= sizeof(struct qb_ipc_request_header);
+    }
+
     int32_t result = s1_msg_handle_request(ctx, data, request_id, request_size);
     cfs_debug("process result %d", result);
 
-    s1_msg_send_response(c, request_id, result);
+    s1_msg_send_response(c, request_id, request_version, result);
 
     return 0;
 }
-- 
2.47.3