From: "Maciej S. Szmigiero" <maciej.szmigiero@oracle.com>

Add a basic support for receiving device state via multifd channels -
channels that are shared with RAM transfers.

Depending whether MULTIFD_FLAG_DEVICE_STATE flag is present or not in the
packet header either device state (MultiFDPacketDeviceState_t) or RAM
data (existing MultiFDPacket_t) is read.

The received device state data is provided to
qemu_loadvm_load_state_buffer() function for processing in the
device's load_state_buffer handler.

Signed-off-by: Maciej S. Szmigiero <maciej.szmigiero@oracle.com>
---
 migration/multifd.c | 87 +++++++++++++++++++++++++++++++++++++++++----
 migration/multifd.h | 26 +++++++++++++-
 2 files changed, 105 insertions(+), 8 deletions(-)

diff --git a/migration/multifd.c b/migration/multifd.c
index 999b88b7ebcb..9578a985449b 100644
--- a/migration/multifd.c
+++ b/migration/multifd.c
@@ -21,6 +21,7 @@
 #include "file.h"
 #include "migration.h"
 #include "migration-stats.h"
+#include "savevm.h"
 #include "socket.h"
 #include "tls.h"
 #include "qemu-file.h"
@@ -252,14 +253,24 @@ static int multifd_recv_unfill_packet_header(MultiFDRecvParams *p,
     return 0;
 }
 
-static int multifd_recv_unfill_packet(MultiFDRecvParams *p, Error **errp)
+static int multifd_recv_unfill_packet_device_state(MultiFDRecvParams *p,
+                                                   Error **errp)
+{
+    MultiFDPacketDeviceState_t *packet = p->packet_dev_state;
+
+    packet->instance_id = be32_to_cpu(packet->instance_id);
+    p->next_packet_size = be32_to_cpu(packet->next_packet_size);
+
+    return 0;
+}
+
+static int multifd_recv_unfill_packet_ram(MultiFDRecvParams *p, Error **errp)
 {
     const MultiFDPacket_t *packet = p->packet;
     int ret = 0;
 
     p->next_packet_size = be32_to_cpu(packet->next_packet_size);
     p->packet_num = be64_to_cpu(packet->packet_num);
-    p->packets_recved++;
 
     if (!(p->flags & MULTIFD_FLAG_SYNC)) {
         ret = multifd_ram_unfill_packet(p, errp);
@@ -271,6 +282,17 @@ static int multifd_recv_unfill_packet(MultiFDRecvParams *p, Error **errp)
     return ret;
 }
 
+static int multifd_recv_unfill_packet(MultiFDRecvParams *p, Error **errp)
+{
+    p->packets_recved++;
+
+    if (p->flags & MULTIFD_FLAG_DEVICE_STATE) {
+        return multifd_recv_unfill_packet_device_state(p, errp);
+    }
+
+    return multifd_recv_unfill_packet_ram(p, errp);
+}
+
 static bool multifd_send_should_exit(void)
 {
     return qatomic_read(&multifd_send_state->exiting);
@@ -1023,6 +1045,7 @@ static void multifd_recv_cleanup_channel(MultiFDRecvParams *p)
     p->packet_len = 0;
     g_free(p->packet);
     p->packet = NULL;
+    g_clear_pointer(&p->packet_dev_state, g_free);
     g_free(p->normal);
     p->normal = NULL;
     g_free(p->zero);
@@ -1124,6 +1147,28 @@ void multifd_recv_sync_main(void)
     trace_multifd_recv_sync_main(multifd_recv_state->packet_num);
 }
 
+static int multifd_device_state_recv(MultiFDRecvParams *p, Error **errp)
+{
+    g_autofree char *idstr = NULL;
+    g_autofree char *dev_state_buf = NULL;
+    int ret;
+
+    dev_state_buf = g_malloc(p->next_packet_size);
+
+    ret = qio_channel_read_all(p->c, dev_state_buf, p->next_packet_size, errp);
+    if (ret != 0) {
+        return ret;
+    }
+
+    idstr = g_strndup(p->packet_dev_state->idstr,
+                      sizeof(p->packet_dev_state->idstr));
+
+    return qemu_loadvm_load_state_buffer(idstr,
+                                         p->packet_dev_state->instance_id,
+                                         dev_state_buf, p->next_packet_size,
+                                         errp);
+}
+
 static void *multifd_recv_thread(void *opaque)
 {
     MultiFDRecvParams *p = opaque;
@@ -1137,6 +1182,7 @@ static void *multifd_recv_thread(void *opaque)
     while (true) {
         MultiFDPacketHdr_t hdr;
         uint32_t flags = 0;
+        bool is_device_state = false;
         bool has_data = false;
         uint8_t *pkt_buf;
         size_t pkt_len;
@@ -1159,8 +1205,14 @@ static void *multifd_recv_thread(void *opaque)
                 break;
             }
 
-            pkt_buf = (uint8_t *)p->packet + sizeof(hdr);
-            pkt_len = p->packet_len - sizeof(hdr);
+            is_device_state = p->flags & MULTIFD_FLAG_DEVICE_STATE;
+            if (is_device_state) {
+                pkt_buf = (uint8_t *)p->packet_dev_state + sizeof(hdr);
+                pkt_len = sizeof(*p->packet_dev_state) - sizeof(hdr);
+            } else {
+                pkt_buf = (uint8_t *)p->packet + sizeof(hdr);
+                pkt_len = p->packet_len - sizeof(hdr);
+            }
 
             ret = qio_channel_read_all_eof(p->c, (char *)pkt_buf, pkt_len,
                                            &local_err);
@@ -1178,9 +1230,14 @@ static void *multifd_recv_thread(void *opaque)
             flags = p->flags;
             /* recv methods don't know how to handle the SYNC flag */
             p->flags &= ~MULTIFD_FLAG_SYNC;
-            if (!(flags & MULTIFD_FLAG_SYNC)) {
-                has_data = p->normal_num || p->zero_num;
+
+            if (is_device_state) {
+                has_data = p->next_packet_size > 0;
+            } else {
+                has_data = !(flags & MULTIFD_FLAG_SYNC) &&
+                    (p->normal_num || p->zero_num);
             }
+
             qemu_mutex_unlock(&p->mutex);
         } else {
             /*
@@ -1209,14 +1266,29 @@ static void *multifd_recv_thread(void *opaque)
         }
 
         if (has_data) {
-            ret = multifd_recv_state->ops->recv(p, &local_err);
+            if (is_device_state) {
+                assert(use_packets);
+                ret = multifd_device_state_recv(p, &local_err);
+            } else {
+                ret = multifd_recv_state->ops->recv(p, &local_err);
+            }
             if (ret != 0) {
                 break;
             }
+        } else if (is_device_state) {
+            error_setg(&local_err,
+                       "multifd: received empty device state packet");
+            break;
         }
 
         if (use_packets) {
             if (flags & MULTIFD_FLAG_SYNC) {
+                if (is_device_state) {
+                    error_setg(&local_err,
+                               "multifd: received SYNC device state packet");
+                    break;
+                }
+
                 qemu_sem_post(&multifd_recv_state->sem_sync);
                 qemu_sem_wait(&p->sem_sync);
             }
@@ -1285,6 +1357,7 @@ int multifd_recv_setup(Error **errp)
             p->packet_len = sizeof(MultiFDPacket_t)
                 + sizeof(uint64_t) * page_count;
             p->packet = g_malloc0(p->packet_len);
+            p->packet_dev_state = g_malloc0(sizeof(*p->packet_dev_state));
         }
         p->name = g_strdup_printf(MIGRATION_THREAD_DST_MULTIFD, i);
         p->normal = g_new0(ram_addr_t, page_count);
diff --git a/migration/multifd.h b/migration/multifd.h
index 106a48496dc6..026b653057e2 100644
--- a/migration/multifd.h
+++ b/migration/multifd.h
@@ -46,6 +46,12 @@ MultiFDRecvData *multifd_get_recv_data(void);
 #define MULTIFD_FLAG_UADK (8 << 1)
 #define MULTIFD_FLAG_QATZIP (16 << 1)
 
+/*
+ * If set it means that this packet contains device state
+ * (MultiFDPacketDeviceState_t), not RAM data (MultiFDPacket_t).
+ */
+#define MULTIFD_FLAG_DEVICE_STATE (1 << 6)
+
 /* This value needs to be a multiple of qemu_target_page_size() */
 #define MULTIFD_PACKET_SIZE (512 * 1024)
 
@@ -78,6 +84,16 @@ typedef struct {
     uint64_t offset[];
 } __attribute__((packed)) MultiFDPacket_t;
 
+typedef struct {
+    MultiFDPacketHdr_t hdr;
+
+    char idstr[256] QEMU_NONSTRING;
+    uint32_t instance_id;
+
+    /* size of the next packet that contains the actual data */
+    uint32_t next_packet_size;
+} __attribute__((packed)) MultiFDPacketDeviceState_t;
+
 typedef struct {
     /* number of used pages */
     uint32_t num;
@@ -95,6 +111,13 @@ struct MultiFDRecvData {
     off_t file_offset;
 };
 
+typedef struct {
+    char *idstr;
+    uint32_t instance_id;
+    char *buf;
+    size_t buf_len;
+} MultiFDDeviceState_t;
+
 typedef enum {
     MULTIFD_PAYLOAD_NONE,
     MULTIFD_PAYLOAD_RAM,
@@ -210,8 +233,9 @@ typedef struct {
 
     /* thread local variables. No locking required */
 
-    /* pointer to the packet */
+    /* pointers to the possible packet types */
     MultiFDPacket_t *packet;
+    MultiFDPacketDeviceState_t *packet_dev_state;
     /* size of the next packet that contains pages */
     uint32_t next_packet_size;
     /* packets received through this channel */

On Sun, Nov 17, 2024 at 08:20:05PM +0100, Maciej S. Szmigiero wrote:
> From: "Maciej S. Szmigiero" <maciej.szmigiero@oracle.com>
> 
> Add a basic support for receiving device state via multifd channels -
> channels that are shared with RAM transfers.
> 
> Depending whether MULTIFD_FLAG_DEVICE_STATE flag is present or not in the
> packet header either device state (MultiFDPacketDeviceState_t) or RAM
> data (existing MultiFDPacket_t) is read.
> 
> The received device state data is provided to
> qemu_loadvm_load_state_buffer() function for processing in the
> device's load_state_buffer handler.
> 
> Signed-off-by: Maciej S. Szmigiero <maciej.szmigiero@oracle.com>

Reviewed-by: Peter Xu <peterx@redhat.com>

Only a few nitpicks:

> ---
>  migration/multifd.c | 87 +++++++++++++++++++++++++++++++++++++++++----
>  migration/multifd.h | 26 +++++++++++++-
>  2 files changed, 105 insertions(+), 8 deletions(-)
> 
> diff --git a/migration/multifd.c b/migration/multifd.c
> index 999b88b7ebcb..9578a985449b 100644
> --- a/migration/multifd.c
> +++ b/migration/multifd.c
> @@ -21,6 +21,7 @@
>  #include "file.h"
>  #include "migration.h"
>  #include "migration-stats.h"
> +#include "savevm.h"
>  #include "socket.h"
>  #include "tls.h"
>  #include "qemu-file.h"
> @@ -252,14 +253,24 @@ static int multifd_recv_unfill_packet_header(MultiFDRecvParams *p,
>      return 0;
>  }
>  
> -static int multifd_recv_unfill_packet(MultiFDRecvParams *p, Error **errp)
> +static int multifd_recv_unfill_packet_device_state(MultiFDRecvParams *p,
> +                                                   Error **errp)
> +{
> +    MultiFDPacketDeviceState_t *packet = p->packet_dev_state;
> +
> +    packet->instance_id = be32_to_cpu(packet->instance_id);
> +    p->next_packet_size = be32_to_cpu(packet->next_packet_size);
> +
> +    return 0;
> +}
> +
> +static int multifd_recv_unfill_packet_ram(MultiFDRecvParams *p, Error **errp)
>  {
>      const MultiFDPacket_t *packet = p->packet;
>      int ret = 0;
>  
>      p->next_packet_size = be32_to_cpu(packet->next_packet_size);
>      p->packet_num = be64_to_cpu(packet->packet_num);
> -    p->packets_recved++;
>  
>      if (!(p->flags & MULTIFD_FLAG_SYNC)) {
>          ret = multifd_ram_unfill_packet(p, errp);
> @@ -271,6 +282,17 @@ static int multifd_recv_unfill_packet(MultiFDRecvParams *p, Error **errp)
>      return ret;
>  }
>  
> +static int multifd_recv_unfill_packet(MultiFDRecvParams *p, Error **errp)
> +{
> +    p->packets_recved++;
> +
> +    if (p->flags & MULTIFD_FLAG_DEVICE_STATE) {
> +        return multifd_recv_unfill_packet_device_state(p, errp);
> +    }
> +
> +    return multifd_recv_unfill_packet_ram(p, errp);
> +}
> +
>  static bool multifd_send_should_exit(void)
>  {
>      return qatomic_read(&multifd_send_state->exiting);
> @@ -1023,6 +1045,7 @@ static void multifd_recv_cleanup_channel(MultiFDRecvParams *p)
>      p->packet_len = 0;
>      g_free(p->packet);
>      p->packet = NULL;
> +    g_clear_pointer(&p->packet_dev_state, g_free);
>      g_free(p->normal);
>      p->normal = NULL;
>      g_free(p->zero);
> @@ -1124,6 +1147,28 @@ void multifd_recv_sync_main(void)
>      trace_multifd_recv_sync_main(multifd_recv_state->packet_num);
>  }
>  
> +static int multifd_device_state_recv(MultiFDRecvParams *p, Error **errp)
> +{
> +    g_autofree char *idstr = NULL;
> +    g_autofree char *dev_state_buf = NULL;
> +    int ret;
> +
> +    dev_state_buf = g_malloc(p->next_packet_size);
> +
> +    ret = qio_channel_read_all(p->c, dev_state_buf, p->next_packet_size, errp);
> +    if (ret != 0) {
> +        return ret;
> +    }
> +
> +    idstr = g_strndup(p->packet_dev_state->idstr,
> +                      sizeof(p->packet_dev_state->idstr));
> +
> +    return qemu_loadvm_load_state_buffer(idstr,
> +                                         p->packet_dev_state->instance_id,
> +                                         dev_state_buf, p->next_packet_size,
> +                                         errp);
> +}
> +
>  static void *multifd_recv_thread(void *opaque)
>  {
>      MultiFDRecvParams *p = opaque;
> @@ -1137,6 +1182,7 @@ static void *multifd_recv_thread(void *opaque)
>      while (true) {
>          MultiFDPacketHdr_t hdr;
>          uint32_t flags = 0;
> +        bool is_device_state = false;
>          bool has_data = false;
>          uint8_t *pkt_buf;
>          size_t pkt_len;
> @@ -1159,8 +1205,14 @@ static void *multifd_recv_thread(void *opaque)
>                  break;
>              }
>  
> -            pkt_buf = (uint8_t *)p->packet + sizeof(hdr);
> -            pkt_len = p->packet_len - sizeof(hdr);
> +            is_device_state = p->flags & MULTIFD_FLAG_DEVICE_STATE;
> +            if (is_device_state) {
> +                pkt_buf = (uint8_t *)p->packet_dev_state + sizeof(hdr);
> +                pkt_len = sizeof(*p->packet_dev_state) - sizeof(hdr);
> +            } else {
> +                pkt_buf = (uint8_t *)p->packet + sizeof(hdr);
> +                pkt_len = p->packet_len - sizeof(hdr);
> +            }
>  
>              ret = qio_channel_read_all_eof(p->c, (char *)pkt_buf, pkt_len,
>                                             &local_err);
> @@ -1178,9 +1230,14 @@ static void *multifd_recv_thread(void *opaque)
>              flags = p->flags;
>              /* recv methods don't know how to handle the SYNC flag */
>              p->flags &= ~MULTIFD_FLAG_SYNC;
> -            if (!(flags & MULTIFD_FLAG_SYNC)) {
> -                has_data = p->normal_num || p->zero_num;
> +
> +            if (is_device_state) {
> +                has_data = p->next_packet_size > 0;
> +            } else {
> +                has_data = !(flags & MULTIFD_FLAG_SYNC) &&
> +                    (p->normal_num || p->zero_num);
>              }
> +
>              qemu_mutex_unlock(&p->mutex);
>          } else {
>              /*
> @@ -1209,14 +1266,29 @@ static void *multifd_recv_thread(void *opaque)
>          }
>  
>          if (has_data) {
> -            ret = multifd_recv_state->ops->recv(p, &local_err);
> +            if (is_device_state) {
> +                assert(use_packets);
> +                ret = multifd_device_state_recv(p, &local_err);
> +            } else {
> +                ret = multifd_recv_state->ops->recv(p, &local_err);
> +            }
>              if (ret != 0) {
>                  break;
>              }
> +        } else if (is_device_state) {
> +            error_setg(&local_err,
> +                       "multifd: received empty device state packet");
> +            break;

You used assert anyway elsewhere, and this also smells like programming
error.  We could stick with assert above and reduce "if / elif ...":

    if (is_device_state) {
        assert(p->next_packet_size > 0);
        has_data = true;
    }

Then drop else if.

>          }
>  
>          if (use_packets) {
>              if (flags & MULTIFD_FLAG_SYNC) {
> +                if (is_device_state) {
> +                    error_setg(&local_err,
> +                               "multifd: received SYNC device state packet");
> +                    break;
> +                }

Same here. I'd use assert().

> +
>                  qemu_sem_post(&multifd_recv_state->sem_sync);
>                  qemu_sem_wait(&p->sem_sync);
>              }
> @@ -1285,6 +1357,7 @@ int multifd_recv_setup(Error **errp)
>              p->packet_len = sizeof(MultiFDPacket_t)
>                  + sizeof(uint64_t) * page_count;
>              p->packet = g_malloc0(p->packet_len);
> +            p->packet_dev_state = g_malloc0(sizeof(*p->packet_dev_state));
>          }
>          p->name = g_strdup_printf(MIGRATION_THREAD_DST_MULTIFD, i);
>          p->normal = g_new0(ram_addr_t, page_count);
> diff --git a/migration/multifd.h b/migration/multifd.h
> index 106a48496dc6..026b653057e2 100644
> --- a/migration/multifd.h
> +++ b/migration/multifd.h
> @@ -46,6 +46,12 @@ MultiFDRecvData *multifd_get_recv_data(void);
>  #define MULTIFD_FLAG_UADK (8 << 1)
>  #define MULTIFD_FLAG_QATZIP (16 << 1)
>  
> +/*
> + * If set it means that this packet contains device state
> + * (MultiFDPacketDeviceState_t), not RAM data (MultiFDPacket_t).
> + */
> +#define MULTIFD_FLAG_DEVICE_STATE (1 << 6)
> +
>  /* This value needs to be a multiple of qemu_target_page_size() */
>  #define MULTIFD_PACKET_SIZE (512 * 1024)
>  
> @@ -78,6 +84,16 @@ typedef struct {
>      uint64_t offset[];
>  } __attribute__((packed)) MultiFDPacket_t;
>  
> +typedef struct {
> +    MultiFDPacketHdr_t hdr;
> +
> +    char idstr[256] QEMU_NONSTRING;
> +    uint32_t instance_id;
> +
> +    /* size of the next packet that contains the actual data */
> +    uint32_t next_packet_size;
> +} __attribute__((packed)) MultiFDPacketDeviceState_t;
> +
>  typedef struct {
>      /* number of used pages */
>      uint32_t num;
> @@ -95,6 +111,13 @@ struct MultiFDRecvData {
>      off_t file_offset;
>  };
>  
> +typedef struct {
> +    char *idstr;
> +    uint32_t instance_id;
> +    char *buf;
> +    size_t buf_len;
> +} MultiFDDeviceState_t;
> +
>  typedef enum {
>      MULTIFD_PAYLOAD_NONE,
>      MULTIFD_PAYLOAD_RAM,
> @@ -210,8 +233,9 @@ typedef struct {
>  
>      /* thread local variables. No locking required */
>  
> -    /* pointer to the packet */
> +    /* pointers to the possible packet types */
>      MultiFDPacket_t *packet;
> +    MultiFDPacketDeviceState_t *packet_dev_state;
>      /* size of the next packet that contains pages */
>      uint32_t next_packet_size;
>      /* packets received through this channel */
> 

-- 
Peter Xu

On 5.12.2024 17:06, Peter Xu wrote:
> On Sun, Nov 17, 2024 at 08:20:05PM +0100, Maciej S. Szmigiero wrote:
>> From: "Maciej S. Szmigiero" <maciej.szmigiero@oracle.com>
>>
>> Add a basic support for receiving device state via multifd channels -
>> channels that are shared with RAM transfers.
>>
>> Depending whether MULTIFD_FLAG_DEVICE_STATE flag is present or not in the
>> packet header either device state (MultiFDPacketDeviceState_t) or RAM
>> data (existing MultiFDPacket_t) is read.
>>
>> The received device state data is provided to
>> qemu_loadvm_load_state_buffer() function for processing in the
>> device's load_state_buffer handler.
>>
>> Signed-off-by: Maciej S. Szmigiero <maciej.szmigiero@oracle.com>
> 
> Reviewed-by: Peter Xu <peterx@redhat.com>
> 
> Only a few nitpicks:
> 
>> ---
>>   migration/multifd.c | 87 +++++++++++++++++++++++++++++++++++++++++----
>>   migration/multifd.h | 26 +++++++++++++-
>>   2 files changed, 105 insertions(+), 8 deletions(-)
>>
>> diff --git a/migration/multifd.c b/migration/multifd.c
>> index 999b88b7ebcb..9578a985449b 100644
>> --- a/migration/multifd.c
>> +++ b/migration/multifd.c
>> @@ -21,6 +21,7 @@
>>   #include "file.h"
>>   #include "migration.h"
>>   #include "migration-stats.h"
>> +#include "savevm.h"
>>   #include "socket.h"
>>   #include "tls.h"
>>   #include "qemu-file.h"
>> @@ -252,14 +253,24 @@ static int multifd_recv_unfill_packet_header(MultiFDRecvParams *p,
>>       return 0;
>>   }
>>   
>> -static int multifd_recv_unfill_packet(MultiFDRecvParams *p, Error **errp)
>> +static int multifd_recv_unfill_packet_device_state(MultiFDRecvParams *p,
>> +                                                   Error **errp)
>> +{
>> +    MultiFDPacketDeviceState_t *packet = p->packet_dev_state;
>> +
>> +    packet->instance_id = be32_to_cpu(packet->instance_id);
>> +    p->next_packet_size = be32_to_cpu(packet->next_packet_size);
>> +
>> +    return 0;
>> +}
>> +
>> +static int multifd_recv_unfill_packet_ram(MultiFDRecvParams *p, Error **errp)
>>   {
>>       const MultiFDPacket_t *packet = p->packet;
>>       int ret = 0;
>>   
>>       p->next_packet_size = be32_to_cpu(packet->next_packet_size);
>>       p->packet_num = be64_to_cpu(packet->packet_num);
>> -    p->packets_recved++;
>>   
>>       if (!(p->flags & MULTIFD_FLAG_SYNC)) {
>>           ret = multifd_ram_unfill_packet(p, errp);
>> @@ -271,6 +282,17 @@ static int multifd_recv_unfill_packet(MultiFDRecvParams *p, Error **errp)
>>       return ret;
>>   }
>>   
>> +static int multifd_recv_unfill_packet(MultiFDRecvParams *p, Error **errp)
>> +{
>> +    p->packets_recved++;
>> +
>> +    if (p->flags & MULTIFD_FLAG_DEVICE_STATE) {
>> +        return multifd_recv_unfill_packet_device_state(p, errp);
>> +    }
>> +
>> +    return multifd_recv_unfill_packet_ram(p, errp);
>> +}
>> +
>>   static bool multifd_send_should_exit(void)
>>   {
>>       return qatomic_read(&multifd_send_state->exiting);
>> @@ -1023,6 +1045,7 @@ static void multifd_recv_cleanup_channel(MultiFDRecvParams *p)
>>       p->packet_len = 0;
>>       g_free(p->packet);
>>       p->packet = NULL;
>> +    g_clear_pointer(&p->packet_dev_state, g_free);
>>       g_free(p->normal);
>>       p->normal = NULL;
>>       g_free(p->zero);
>> @@ -1124,6 +1147,28 @@ void multifd_recv_sync_main(void)
>>       trace_multifd_recv_sync_main(multifd_recv_state->packet_num);
>>   }
>>   
>> +static int multifd_device_state_recv(MultiFDRecvParams *p, Error **errp)
>> +{
>> +    g_autofree char *idstr = NULL;
>> +    g_autofree char *dev_state_buf = NULL;
>> +    int ret;
>> +
>> +    dev_state_buf = g_malloc(p->next_packet_size);
>> +
>> +    ret = qio_channel_read_all(p->c, dev_state_buf, p->next_packet_size, errp);
>> +    if (ret != 0) {
>> +        return ret;
>> +    }
>> +
>> +    idstr = g_strndup(p->packet_dev_state->idstr,
>> +                      sizeof(p->packet_dev_state->idstr));
>> +
>> +    return qemu_loadvm_load_state_buffer(idstr,
>> +                                         p->packet_dev_state->instance_id,
>> +                                         dev_state_buf, p->next_packet_size,
>> +                                         errp);
>> +}
>> +
>>   static void *multifd_recv_thread(void *opaque)
>>   {
>>       MultiFDRecvParams *p = opaque;
>> @@ -1137,6 +1182,7 @@ static void *multifd_recv_thread(void *opaque)
>>       while (true) {
>>           MultiFDPacketHdr_t hdr;
>>           uint32_t flags = 0;
>> +        bool is_device_state = false;
>>           bool has_data = false;
>>           uint8_t *pkt_buf;
>>           size_t pkt_len;
>> @@ -1159,8 +1205,14 @@ static void *multifd_recv_thread(void *opaque)
>>                   break;
>>               }
>>   
>> -            pkt_buf = (uint8_t *)p->packet + sizeof(hdr);
>> -            pkt_len = p->packet_len - sizeof(hdr);
>> +            is_device_state = p->flags & MULTIFD_FLAG_DEVICE_STATE;
>> +            if (is_device_state) {
>> +                pkt_buf = (uint8_t *)p->packet_dev_state + sizeof(hdr);
>> +                pkt_len = sizeof(*p->packet_dev_state) - sizeof(hdr);
>> +            } else {
>> +                pkt_buf = (uint8_t *)p->packet + sizeof(hdr);
>> +                pkt_len = p->packet_len - sizeof(hdr);
>> +            }
>>   
>>               ret = qio_channel_read_all_eof(p->c, (char *)pkt_buf, pkt_len,
>>                                              &local_err);
>> @@ -1178,9 +1230,14 @@ static void *multifd_recv_thread(void *opaque)
>>               flags = p->flags;
>>               /* recv methods don't know how to handle the SYNC flag */
>>               p->flags &= ~MULTIFD_FLAG_SYNC;
>> -            if (!(flags & MULTIFD_FLAG_SYNC)) {
>> -                has_data = p->normal_num || p->zero_num;
>> +
>> +            if (is_device_state) {
>> +                has_data = p->next_packet_size > 0;
>> +            } else {
>> +                has_data = !(flags & MULTIFD_FLAG_SYNC) &&
>> +                    (p->normal_num || p->zero_num);
>>               }
>> +
>>               qemu_mutex_unlock(&p->mutex);
>>           } else {
>>               /*
>> @@ -1209,14 +1266,29 @@ static void *multifd_recv_thread(void *opaque)
>>           }
>>   
>>           if (has_data) {
>> -            ret = multifd_recv_state->ops->recv(p, &local_err);
>> +            if (is_device_state) {
>> +                assert(use_packets);
>> +                ret = multifd_device_state_recv(p, &local_err);
>> +            } else {
>> +                ret = multifd_recv_state->ops->recv(p, &local_err);
>> +            }
>>               if (ret != 0) {
>>                   break;
>>               }
>> +        } else if (is_device_state) {
>> +            error_setg(&local_err,
>> +                       "multifd: received empty device state packet");
>> +            break;
> 
> You used assert anyway elsewhere, and this also smells like programming
> error.  We could stick with assert above and reduce "if / elif ...":
> 
>      if (is_device_state) {
>          assert(p->next_packet_size > 0);
>          has_data = true;
>      }
> 
> Then drop else if.

It's not necessarily a programming error, but rather a problem with the
received bit stream or its incompatibility with the receiving QEMU version.

So I think returning an error is more appropriate than triggering
an assert() failure for that.

>>           }
>>   
>>           if (use_packets) {
>>               if (flags & MULTIFD_FLAG_SYNC) {
>> +                if (is_device_state) {
>> +                    error_setg(&local_err,
>> +                               "multifd: received SYNC device state packet");
>> +                    break;
>> +                }
> 
> Same here. I'd use assert().
> 

Same here :) - the sender sent us possibly wrong packet or packet of
incompatible version, we should handle this gracefully rather than
assert()/abort() QEMU.

Thanks,
Maciej

On Fri, Dec 06, 2024 at 10:12:27PM +0100, Maciej S. Szmigiero wrote:
> Same here :) - the sender sent us possibly wrong packet or packet of
> incompatible version, we should handle this gracefully rather than
> assert()/abort() QEMU.

Ah, sure.  Feel free to keep them.  My R-b can keep.

-- 
Peter Xu