Split up IPC messages

cts/cli/regression.error_codes.exp

@@ -145,6 +145,7 @@ pcmk_rc_node_unknown - Node not found
 =#=#=#= End test: Get negative Pacemaker return code (with name) (XML) - OK (0) =#=#=#=
 * Passed: crm_error             - Get negative Pacemaker return code (with name) (XML)
 =#=#=#= Begin test: List Pacemaker return codes (non-positive) =#=#=#=
+-1041: More IPC message fragments to send
 -1040: DC is not yet elected
 -1039: Compression/decompression error
 -1038: Nameserver resolution error
@@ -190,6 +191,7 @@ pcmk_rc_node_unknown - Node not found
 * Passed: crm_error             - List Pacemaker return codes (non-positive)
 =#=#=#= Begin test: List Pacemaker return codes (non-positive) (XML) =#=#=#=
 <pacemaker-result api-version="X" request="crm_error -l -r --output-as=xml">
+  <result-code code="-1041" description="More IPC message fragments to send"/>
   <result-code code="-1040" description="DC is not yet elected"/>
   <result-code code="-1039" description="Compression/decompression error"/>
   <result-code code="-1038" description="Nameserver resolution error"/>
@@ -235,6 +237,7 @@ pcmk_rc_node_unknown - Node not found
 =#=#=#= End test: List Pacemaker return codes (non-positive) (XML) - OK (0) =#=#=#=
 * Passed: crm_error             - List Pacemaker return codes (non-positive) (XML)
 =#=#=#= Begin test: List Pacemaker return codes (non-positive) (with names) =#=#=#=
+-1041: pcmk_rc_ipc_more            More IPC message fragments to send
 -1040: pcmk_rc_no_dc               DC is not yet elected
 -1039: pcmk_rc_compression         Compression/decompression error
 -1038: pcmk_rc_ns_resolution       Nameserver resolution error
@@ -280,6 +283,7 @@ pcmk_rc_node_unknown - Node not found
 * Passed: crm_error             - List Pacemaker return codes (non-positive) (with names)
 =#=#=#= Begin test: List Pacemaker return codes (non-positive) (with names) (XML) =#=#=#=
 <pacemaker-result api-version="X" request="crm_error -n -l -r --output-as=xml">
+  <result-code code="-1041" name="pcmk_rc_ipc_more" description="More IPC message fragments to send"/>
   <result-code code="-1040" name="pcmk_rc_no_dc" description="DC is not yet elected"/>
   <result-code code="-1039" name="pcmk_rc_compression" description="Compression/decompression error"/>
   <result-code code="-1038" name="pcmk_rc_ns_resolution" description="Nameserver resolution error"/>

daemons/attrd/attrd_ipc.c

@@ -553,6 +553,7 @@ attrd_ipc_destroy(qb_ipcs_connection_t *c)
 static int32_t
 attrd_ipc_dispatch(qb_ipcs_connection_t * c, void *data, size_t size)
 {
+    int rc = pcmk_rc_ok;
     uint32_t id = 0;
     uint32_t flags = 0;
     pcmk__client_t *client = pcmk__find_client(c);
@@ -565,7 +566,33 @@ attrd_ipc_dispatch(qb_ipcs_connection_t * c, void *data, size_t size)
         return 0;
     }
 
-    xml = pcmk__client_data2xml(client, data, &id, &flags);
+    rc = pcmk__ipc_msg_append(&client->buffer, data);
+
+    if (rc == pcmk_rc_ipc_more) {
+        /* We haven't read the complete message yet, so just return. */
+        return 0;
+
+    } else if (rc == pcmk_rc_ok) {
+        /* We've read the complete message and there's already a header on
+         * the front.  Pass it off for processing.
+         */
+        xml = pcmk__client_data2xml(client, client->buffer->data, &id, &flags);
+        g_byte_array_free(client->buffer, TRUE);
+        client->buffer = NULL;
+
+    } else {
+        /* Some sort of error occurred reassembling the message.  All we can
+         * do is clean up, log an error and return.
+         */
+        crm_err("Error when reading IPC message: %s", pcmk_rc_str(rc));
+
+        if (client->buffer != NULL) {
+            g_byte_array_free(client->buffer, TRUE);
+            client->buffer = NULL;
+        }
+
+        return 0;
+    }
 
     if (xml == NULL) {
         crm_debug("Unrecognizable IPC data from PID %d", pcmk__client_pid(c));

daemons/based/based_callbacks.c

@@ -302,11 +302,46 @@ cib_common_callback_worker(uint32_t id, uint32_t flags, xmlNode * op_request,
 int32_t
 cib_common_callback(qb_ipcs_connection_t * c, void *data, size_t size, gboolean privileged)
 {
+    int rc = pcmk_rc_ok;
     uint32_t id = 0;
     uint32_t flags = 0;
     uint32_t call_options = cib_none;
     pcmk__client_t *cib_client = pcmk__find_client(c);
-    xmlNode *op_request = pcmk__client_data2xml(cib_client, data, &id, &flags);
+    xmlNode *op_request = NULL;
+
+    if (cib_client == NULL) {
+        crm_trace("Invalid client %p", c);
+        return 0;
+    }
+
+    rc = pcmk__ipc_msg_append(&cib_client->buffer, data);
+
+    if (rc == pcmk_rc_ipc_more) {
+        /* We haven't read the complete message yet, so just return. */
+        return 0;
+
+    } else if (rc == pcmk_rc_ok) {
+        /* We've read the complete message and there's already a header on
+         * the front.  Pass it off for processing.
+         */
+        op_request = pcmk__client_data2xml(cib_client, cib_client->buffer->data,
+                                           &id, &flags);
+        g_byte_array_free(cib_client->buffer, TRUE);
+        cib_client->buffer = NULL;
+
+    } else {
+        /* Some sort of error occurred reassembling the message.  All we can
+         * do is clean up, log an error and return.
+         */
+        crm_err("Error when reading IPC message: %s", pcmk_rc_str(rc));
+
+        if (cib_client->buffer != NULL) {
+            g_byte_array_free(cib_client->buffer, TRUE);
+            cib_client->buffer = NULL;
+        }
+
+        return 0;
+    }
 
     if (op_request) {
         int rc = pcmk_rc_ok;
@@ -324,10 +359,6 @@ cib_common_callback(qb_ipcs_connection_t * c, void *data, size_t size, gboolean
         pcmk__ipc_send_ack(cib_client, id, flags, PCMK__XE_NACK, NULL,
                            CRM_EX_PROTOCOL);
         return 0;
-
-    } else if(cib_client == NULL) {
-        crm_trace("Invalid client %p", c);
-        return 0;
     }
 
     if (pcmk_is_set(call_options, cib_sync_call)) {

daemons/based/based_notify.c

@@ -81,7 +81,14 @@ cib_notify_send_one(gpointer key, gpointer value, gpointer user_data)
             case pcmk__client_ipc:
                 rc = pcmk__ipc_send_iov(client, update->iov,
                                         crm_ipc_server_event);
-                if (rc != pcmk_rc_ok) {
+
+                /* EAGAIN isn't strictly an error for a server event.  The iov
+                 * was added to the send queue, but sending did fail with EAGAIN.
+                 * However, we will attempt to send the event the next time
+                 * pcmk__ipc_send_iov is called, or when crm_ipcs_flush_events_cb
+                 * happens.
+                 */
+                if ((rc != EAGAIN) && (rc != pcmk_rc_ok)) {
                     crm_warn("Could not notify client %s: %s " QB_XS " id=%s",
                              pcmk__client_name(client), pcmk_rc_str(rc),
                              client->id);
@@ -106,21 +113,37 @@ cib_notify_send(const xmlNode *xml)
 {
     struct iovec *iov;
     struct cib_notification_s update;
-
+    GString *iov_buffer = NULL;
     ssize_t bytes = 0;
-    int rc = pcmk__ipc_prepare_iov(0, xml, &iov, &bytes);
+    int rc = pcmk_rc_ok;
+    uint16_t index = 0;
+
+    iov_buffer = g_string_sized_new(1024);
+    pcmk__xml_string(xml, 0, iov_buffer, 0);
+
+    do {
+        rc = pcmk__ipc_prepare_iov(0, iov_buffer, index, &iov, &bytes);
+
+        if ((rc != pcmk_rc_ok) && (rc != pcmk_rc_ipc_more)) {
+            crm_notice("Could not notify clients: %s " QB_XS " rc=%d",
+                       pcmk_rc_str(rc), rc);
+            break;
+        }
 
-    if (rc == pcmk_rc_ok) {
         update.msg = xml;
         update.iov = iov;
         update.iov_size = bytes;
         pcmk__foreach_ipc_client(cib_notify_send_one, &update);
         pcmk_free_ipc_event(iov);
 
-    } else {
-        crm_notice("Could not notify clients: %s " QB_XS " rc=%d",
-                   pcmk_rc_str(rc), rc);
-    }
+        if (rc == pcmk_rc_ok) {
+            break;
+        }
+
+        index++;
+    } while (true);
+
+    g_string_free(iov_buffer, TRUE);
 }
 
 void

daemons/controld/controld_control.c

@@ -378,11 +378,39 @@ accept_controller_client(qb_ipcs_connection_t *c, uid_t uid, gid_t gid)
 static int32_t
 dispatch_controller_ipc(qb_ipcs_connection_t * c, void *data, size_t size)
 {
+    int rc = pcmk_rc_ok;
     uint32_t id = 0;
     uint32_t flags = 0;
     pcmk__client_t *client = pcmk__find_client(c);
+    xmlNode *msg = NULL;
+
+    rc = pcmk__ipc_msg_append(&client->buffer, data);
+
+    if (rc == pcmk_rc_ipc_more) {
+        /* We haven't read the complete message yet, so just return. */
+        return 0;
+
+    } else if (rc == pcmk_rc_ok) {
+        /* We've read the complete message and there's already a header on
+         * the front.  Pass it off for processing.
+         */
+        msg = pcmk__client_data2xml(client, client->buffer->data, &id, &flags);
+        g_byte_array_free(client->buffer, TRUE);
+        client->buffer = NULL;
 
-    xmlNode *msg = pcmk__client_data2xml(client, data, &id, &flags);
+    } else {
+        /* Some sort of error occurred reassembling the message.  All we can
+         * do is clean up, log an error and return.
+         */
+        crm_err("Error when reading IPC message: %s", pcmk_rc_str(rc));
+
+        if (client->buffer != NULL) {
+            g_byte_array_free(client->buffer, TRUE);
+            client->buffer = NULL;
+        }
+
+        return 0;
+    }
 
     if (msg == NULL) {
         pcmk__ipc_send_ack(client, id, flags, PCMK__XE_ACK, NULL,

daemons/execd/pacemaker-execd.c

@@ -123,16 +123,45 @@ lrmd_ipc_created(qb_ipcs_connection_t * c)
 static int32_t
 lrmd_ipc_dispatch(qb_ipcs_connection_t * c, void *data, size_t size)
 {
+    int rc = pcmk_rc_ok;
     uint32_t id = 0;
     uint32_t flags = 0;
     pcmk__client_t *client = pcmk__find_client(c);
-    xmlNode *request = pcmk__client_data2xml(client, data, &id, &flags);
+    xmlNode *request = NULL;
 
     CRM_CHECK(client != NULL, crm_err("Invalid client");
               return FALSE);
     CRM_CHECK(client->id != NULL, crm_err("Invalid client: %p", client);
               return FALSE);
 
+    rc = pcmk__ipc_msg_append(&client->buffer, data);
+
+    if (rc == pcmk_rc_ipc_more) {
+        /* We haven't read the complete message yet, so just return. */
+        return 0;
+
+    } else if (rc == pcmk_rc_ok) {
+        /* We've read the complete message and there's already a header on
+         * the front.  Pass it off for processing.
+         */
+        request = pcmk__client_data2xml(client, client->buffer->data, &id, &flags);
+        g_byte_array_free(client->buffer, TRUE);
+        client->buffer = NULL;
+
+    } else {
+        /* Some sort of error occurred reassembling the message.  All we can
+         * do is clean up, log an error and return.
+         */
+        crm_err("Error when reading IPC message: %s", pcmk_rc_str(rc));
+
+        if (client->buffer != NULL) {
+            g_byte_array_free(client->buffer, TRUE);
+            client->buffer = NULL;
+        }
+
+        return 0;
+    }
+
     CRM_CHECK(flags & crm_ipc_client_response, crm_err("Invalid client request: %p", client);
               return FALSE);
 

daemons/execd/remoted_proxy.c

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2024 the Pacemaker project contributors
+ * Copyright 2012-2025 the Pacemaker project contributors
  *
  * The version control history for this file may have further details.
  *
@@ -227,6 +227,7 @@ ipc_proxy_forward_client(pcmk__client_t *ipc_proxy, xmlNode *xml)
 static int32_t
 ipc_proxy_dispatch(qb_ipcs_connection_t * c, void *data, size_t size)
 {
+    int rc = pcmk_rc_ok;
     uint32_t id = 0;
     uint32_t flags = 0;
     pcmk__client_t *client = pcmk__find_client(c);
@@ -253,7 +254,33 @@ ipc_proxy_dispatch(qb_ipcs_connection_t * c, void *data, size_t size)
      * This function is receiving a request from connection
      * 1 and forwarding it to connection 2.
      */
-    request = pcmk__client_data2xml(client, data, &id, &flags);
+    rc = pcmk__ipc_msg_append(&client->buffer, data);
+
+    if (rc == pcmk_rc_ipc_more) {
+        /* We haven't read the complete message yet, so just return. */
+        return 0;
+
+    } else if (rc == pcmk_rc_ok) {
+        /* We've read the complete message and there's already a header on
+         * the front.  Pass it off for processing.
+         */
+        request = pcmk__client_data2xml(client, client->buffer->data, &id, &flags);
+        g_byte_array_free(client->buffer, TRUE);
+        client->buffer = NULL;
+
+    } else {
+        /* Some sort of error occurred reassembling the message.  All we can
+         * do is clean up, log an error and return.
+         */
+        crm_err("Error when reading IPC message: %s", pcmk_rc_str(rc));
+
+        if (client->buffer != NULL) {
+            g_byte_array_free(client->buffer, TRUE);
+            client->buffer = NULL;
+        }
+
+        return 0;
+    }
 
     if (!request) {
         return 0;

daemons/fenced/pacemaker-fenced.c

@@ -100,13 +100,39 @@ st_ipc_dispatch(qb_ipcs_connection_t * qbc, void *data, size_t size)
         return 0;
     }
 
-    request = pcmk__client_data2xml(c, data, &id, &flags);
+    rc = pcmk__ipc_msg_append(&c->buffer, data);
+
+    if (rc == pcmk_rc_ipc_more) {
+        /* We haven't read the complete message yet, so just return. */
+        return 0;
+
+    } else if (rc == pcmk_rc_ok) {
+        /* We've read the complete message and there's already a header on
+         * the front.  Pass it off for processing.
+         */
+        request = pcmk__client_data2xml(c, c->buffer->data, &id, &flags);
+        g_byte_array_free(c->buffer, TRUE);
+        c->buffer = NULL;
+
+    } else {
+        /* Some sort of error occurred reassembling the message.  All we can
+         * do is clean up, log an error and return.
+         */
+        crm_err("Error when reading IPC message: %s", pcmk_rc_str(rc));
+
+        if (c->buffer != NULL) {
+            g_byte_array_free(c->buffer, TRUE);
+            c->buffer = NULL;
+        }
+
+        return 0;
+    }
+
     if (request == NULL) {
         pcmk__ipc_send_ack(c, id, flags, PCMK__XE_NACK, NULL, CRM_EX_PROTOCOL);
         return 0;
     }
 
-
     op = crm_element_value(request, PCMK__XA_CRM_TASK);
     if(pcmk__str_eq(op, CRM_OP_RM_NODE_CACHE, pcmk__str_casei)) {
         crm_xml_add(request, PCMK__XA_T, PCMK__VALUE_STONITH_NG);