migration/multifd.h

/*
 * Multifd common functions
 *
 * Copyright (c) 2019-2020 Red Hat Inc
 *
 * Authors:
 *  Juan Quintela <quintela@redhat.com>
 *
 * This work is licensed under the terms of the GNU GPL, version 2 or later.
 * See the COPYING file in the top-level directory.
 */

#ifndef QEMU_MIGRATION_MULTIFD_H
#define QEMU_MIGRATION_MULTIFD_H

#include "exec/target_page.h"
#include "ram.h"

typedef struct MultiFDRecvData MultiFDRecvData;
typedef struct MultiFDSendData MultiFDSendData;

bool multifd_send_setup(void);
void multifd_send_shutdown(void);
void multifd_send_channel_created(void);
int multifd_recv_setup(Error **errp);
void multifd_recv_cleanup(void);
void multifd_recv_shutdown(void);
bool multifd_recv_all_channels_created(void);
void multifd_recv_new_channel(QIOChannel *ioc, Error **errp);
void multifd_recv_sync_main(void);
int multifd_send_sync_main(void);
bool multifd_queue_page(RAMBlock *block, ram_addr_t offset);
bool multifd_recv(void);
MultiFDRecvData *multifd_get_recv_data(void);

/* Multifd Compression flags */
#define MULTIFD_FLAG_SYNC (1 << 0)

/* We reserve 4 bits for compression methods */
#define MULTIFD_FLAG_COMPRESSION_MASK (0xf << 1)
/* we need to be compatible. Before compression value was 0 */
#define MULTIFD_FLAG_NOCOMP (0 << 1)
#define MULTIFD_FLAG_ZLIB (1 << 1)
#define MULTIFD_FLAG_ZSTD (2 << 1)
#define MULTIFD_FLAG_QPL (4 << 1)
#define MULTIFD_FLAG_UADK (8 << 1)

/* This value needs to be a multiple of qemu_target_page_size() */
#define MULTIFD_PACKET_SIZE (512 * 1024)

typedef struct {
    uint32_t magic;
    uint32_t version;
    uint32_t flags;
    /* maximum number of allocated pages */
    uint32_t pages_alloc;
    /* non zero pages */
    uint32_t normal_pages;
    /* size of the next packet that contains pages */
    uint32_t next_packet_size;
    uint64_t packet_num;
    /* zero pages */
    uint32_t zero_pages;
    uint32_t unused32[1];    /* Reserved for future use */
    uint64_t unused64[3];    /* Reserved for future use */
    char ramblock[256];
    /*
     * This array contains the pointers to:
     *  - normal pages (initial normal_pages entries)
     *  - zero pages (following zero_pages entries)
     */
    uint64_t offset[];
} __attribute__((packed)) MultiFDPacket_t;

typedef struct {
    /* number of used pages */
    uint32_t num;
    /* number of normal pages */
    uint32_t normal_num;
    RAMBlock *block;
    /* offset of each page */
    ram_addr_t offset[];
} MultiFDPages_t;

struct MultiFDRecvData {
    void *opaque;
    size_t size;
    /* for preadv */
    off_t file_offset;
};

typedef enum {
    MULTIFD_PAYLOAD_NONE,
    MULTIFD_PAYLOAD_RAM,
} MultiFDPayloadType;

typedef union MultiFDPayload {
    MultiFDPages_t ram;
} MultiFDPayload;

struct MultiFDSendData {
    MultiFDPayloadType type;
    MultiFDPayload u;
};

static inline bool multifd_payload_empty(MultiFDSendData *data)
{
    return data->type == MULTIFD_PAYLOAD_NONE;
}

static inline void multifd_set_payload_type(MultiFDSendData *data,
                                            MultiFDPayloadType type)
{
    data->type = type;
}

typedef struct {
    /* Fields are only written at creating/deletion time */
    /* No lock required for them, they are read only */

    /* channel number */
    uint8_t id;
    /* channel thread name */
    char *name;
    /* channel thread id */
    QemuThread thread;
    bool thread_created;
    QemuThread tls_thread;
    bool tls_thread_created;
    /* communication channel */
    QIOChannel *c;
    /* packet allocated len */
    uint32_t packet_len;
    /* multifd flags for sending ram */
    int write_flags;

    /* sem where to wait for more work */
    QemuSemaphore sem;
    /* syncs main thread and channels */
    QemuSemaphore sem_sync;

    /* multifd flags for each packet */
    uint32_t flags;
    /*
     * The sender thread has work to do if either of below boolean is set.
     *
     * @pending_job:  a job is pending
     * @pending_sync: a sync request is pending
     *
     * For both of these fields, they're only set by the requesters, and
     * cleared by the multifd sender threads.
     */
    bool pending_job;
    bool pending_sync;
    MultiFDSendData *data;

    /* thread local variables. No locking required */

    /* pointer to the packet */
    MultiFDPacket_t *packet;
    /* size of the next packet that contains pages */
    uint32_t next_packet_size;
    /* packets sent through this channel */
    uint64_t packets_sent;
    /* buffers to send */
    struct iovec *iov;
    /* number of iovs used */
    uint32_t iovs_num;
    /* used for compression methods */
    void *compress_data;
}  MultiFDSendParams;

typedef struct {
    /* Fields are only written at creating/deletion time */
    /* No lock required for them, they are read only */

    /* channel number */
    uint8_t id;
    /* channel thread name */
    char *name;
    /* channel thread id */
    QemuThread thread;
    bool thread_created;
    /* communication channel */
    QIOChannel *c;
    /* packet allocated len */
    uint32_t packet_len;

    /* syncs main thread and channels */
    QemuSemaphore sem_sync;
    /* sem where to wait for more work */
    QemuSemaphore sem;

    /* this mutex protects the following parameters */
    QemuMutex mutex;
    /* should this thread finish */
    bool quit;
    /* multifd flags for each packet */
    uint32_t flags;
    /* global number of generated multifd packets */
    uint64_t packet_num;
    int pending_job;
    MultiFDRecvData *data;

    /* thread local variables. No locking required */

    /* pointer to the packet */
    MultiFDPacket_t *packet;
    /* size of the next packet that contains pages */
    uint32_t next_packet_size;
    /* packets received through this channel */
    uint64_t packets_recved;
    /* ramblock */
    RAMBlock *block;
    /* ramblock host address */
    uint8_t *host;
    /* buffers to recv */
    struct iovec *iov;
    /* Pages that are not zero */
    ram_addr_t *normal;
    /* num of non zero pages */
    uint32_t normal_num;
    /* Pages that are zero */
    ram_addr_t *zero;
    /* num of zero pages */
    uint32_t zero_num;
    /* used for de-compression methods */
    void *compress_data;
} MultiFDRecvParams;

typedef struct {
    /* Setup for sending side */
    int (*send_setup)(MultiFDSendParams *p, Error **errp);
    /* Cleanup for sending side */
    void (*send_cleanup)(MultiFDSendParams *p, Error **errp);
    /* Prepare the send packet */
    int (*send_prepare)(MultiFDSendParams *p, Error **errp);
    /* Setup for receiving side */
    int (*recv_setup)(MultiFDRecvParams *p, Error **errp);
    /* Cleanup for receiving side */
    void (*recv_cleanup)(MultiFDRecvParams *p);
    /* Read all data */
    int (*recv)(MultiFDRecvParams *p, Error **errp);
} MultiFDMethods;

void multifd_register_ops(int method, MultiFDMethods *ops);
void multifd_send_fill_packet(MultiFDSendParams *p);
bool multifd_send_prepare_common(MultiFDSendParams *p);
void multifd_send_zero_page_detect(MultiFDSendParams *p);
void multifd_recv_zero_page_process(MultiFDRecvParams *p);

static inline void multifd_send_prepare_header(MultiFDSendParams *p)
{
    p->iov[0].iov_len = p->packet_len;
    p->iov[0].iov_base = p->packet;
    p->iovs_num++;
}

void multifd_channel_connect(MultiFDSendParams *p, QIOChannel *ioc);

static inline uint32_t multifd_ram_page_size(void)
{
    return qemu_target_page_size();
}

static inline uint32_t multifd_ram_page_count(void)
{
    return MULTIFD_PACKET_SIZE / qemu_target_page_size();
}
#endif