[v13,08/12] migration: Create ram_save_multifd_page

The function still don't use multifd, but we have simplified
ram_save_page, xbzrle and RDMA stuff is gone.  We have added a new
counter.

Signed-off-by: Juan Quintela <quintela@redhat.com>

--
Add last_page parameter
Add commets for done and address
Remove multifd field, it is the same than normal pages
Merge next patch, now we send multiple pages at a time
Remove counter for multifd pages, it is identical to normal pages
Use iovec's instead of creating the equivalent.
Clear memory used by pages (dave)
Use g_new0(danp)
define MULTIFD_CONTINUE
now pages member is a pointer
Fix off-by-one in number of pages in one packet
Remove RAM_SAVE_FLAG_MULTIFD_PAGE
s/multifd_pages_t/MultiFDPages_t/
---
 migration/ram.c | 85 +++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 85 insertions(+)

* Juan Quintela (quintela@redhat.com) wrote:
> The function still don't use multifd, but we have simplified
> ram_save_page, xbzrle and RDMA stuff is gone.  We have added a new
> counter.
> 
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> 
> --
> Add last_page parameter
> Add commets for done and address
> Remove multifd field, it is the same than normal pages
> Merge next patch, now we send multiple pages at a time
> Remove counter for multifd pages, it is identical to normal pages
> Use iovec's instead of creating the equivalent.
> Clear memory used by pages (dave)
> Use g_new0(danp)
> define MULTIFD_CONTINUE
> now pages member is a pointer
> Fix off-by-one in number of pages in one packet
> Remove RAM_SAVE_FLAG_MULTIFD_PAGE
> s/multifd_pages_t/MultiFDPages_t/
> ---
>  migration/ram.c | 85 +++++++++++++++++++++++++++++++++++++++++++++++++
>  1 file changed, 85 insertions(+)
> 
> diff --git a/migration/ram.c b/migration/ram.c
> index 3e99d48123..28f5cea4d8 100644
> --- a/migration/ram.c
> +++ b/migration/ram.c
> @@ -55,6 +55,7 @@
>  #include "sysemu/sysemu.h"
>  #include "qemu/uuid.h"
>  #include "savevm.h"
> +#include "qemu/iov.h"
>  
>  /***********************************************************/
>  /* ram save/restore */
> @@ -804,8 +805,65 @@ struct {
>      QemuSemaphore sem_sync;
>      /* global number of generated multifd packets */
>      uint32_t seq;
> +    /* send channels ready */
> +    QemuSemaphore channels_ready;
>  } *multifd_send_state;
>  
> +static void multifd_send_pages(void)
> +{
> +    int i;
> +    static int next_channel;
> +    MultiFDSendParams *p = NULL; /* make happy gcc */
> +    MultiFDPages_t *pages = multifd_send_state->pages;
> +
> +    qemu_sem_wait(&multifd_send_state->channels_ready);
> +    for (i = next_channel;; i = (i + 1) % migrate_multifd_channels()) {
> +        p = &multifd_send_state->params[i];
> +
> +        qemu_mutex_lock(&p->mutex);
> +        if (!p->pending_job) {
> +            p->pending_job++;
> +            next_channel = (i + 1) % migrate_multifd_channels();
> +            break;
> +        }
> +        qemu_mutex_unlock(&p->mutex);
> +    }
> +    p->pages->used = 0;
> +    multifd_send_state->seq++;
> +    p->seq = multifd_send_state->seq;
> +    p->pages->block = NULL;
> +    multifd_send_state->pages = p->pages;
> +    p->pages = pages;

This function *really* needs a fat comment about the flow of ->pages
because it's pretty hard to follow.

Other than that;


Reviewed-by: Dr. David Alan Gilbert <dgilbert@redhat.com>

> +    qemu_mutex_unlock(&p->mutex);
> +    qemu_sem_post(&p->sem);
> +}
> +
> +static void multifd_queue_page(RAMBlock *block, ram_addr_t offset)
> +{
> +    MultiFDPages_t *pages = multifd_send_state->pages;
> +
> +    if (!pages->block) {
> +        pages->block = block;
> +    }
> +
> +    if (pages->block == block) {
> +        pages->offset[pages->used] = offset;
> +        pages->iov[pages->used].iov_base = block->host + offset;
> +        pages->iov[pages->used].iov_len = TARGET_PAGE_SIZE;
> +        pages->used++;
> +
> +        if (pages->used < pages->allocated) {
> +            return;
> +        }
> +    }
> +
> +    multifd_send_pages();
> +
> +    if (pages->block != block) {
> +        multifd_queue_page(block, offset);
> +    }
> +}
> +
>  static void multifd_send_terminate_threads(Error *err)
>  {
>      int i;
> @@ -860,6 +918,7 @@ int multifd_save_cleanup(Error **errp)
>          g_free(p->packet);
>          p->packet = NULL;
>      }
> +    qemu_sem_destroy(&multifd_send_state->channels_ready);
>      qemu_sem_destroy(&multifd_send_state->sem_sync);
>      g_free(multifd_send_state->params);
>      multifd_send_state->params = NULL;
> @@ -877,12 +936,17 @@ static void multifd_send_sync_main(void)
>      if (!migrate_use_multifd()) {
>          return;
>      }
> +    if (multifd_send_state->pages->used) {
> +        multifd_send_pages();
> +    }
>      for (i = 0; i < migrate_multifd_channels(); i++) {
>          MultiFDSendParams *p = &multifd_send_state->params[i];
>  
>          trace_multifd_send_sync_main_signal(p->id);
>  
>          qemu_mutex_lock(&p->mutex);
> +        multifd_send_state->seq++;
> +        p->seq = multifd_send_state->seq;
>          p->flags |= MULTIFD_FLAG_SYNC;
>          p->pending_job++;
>          qemu_mutex_unlock(&p->mutex);
> @@ -937,6 +1001,7 @@ static void *multifd_send_thread(void *opaque)
>              if (flags & MULTIFD_FLAG_SYNC) {
>                  qemu_sem_post(&multifd_send_state->sem_sync);
>              }
> +            qemu_sem_post(&multifd_send_state->channels_ready);
>          } else if (p->quit) {
>              qemu_mutex_unlock(&p->mutex);
>              break;
> @@ -996,6 +1061,7 @@ int multifd_save_setup(void)
>      atomic_set(&multifd_send_state->count, 0);
>      multifd_send_state->pages = multifd_pages_init(page_count);
>      qemu_sem_init(&multifd_send_state->sem_sync, 0);
> +    qemu_sem_init(&multifd_send_state->channels_ready, 0);
>  
>      for (i = 0; i < thread_count; i++) {
>          MultiFDSendParams *p = &multifd_send_state->params[i];
> @@ -1713,6 +1779,23 @@ static int ram_save_page(RAMState *rs, PageSearchStatus *pss, bool last_stage)
>      return pages;
>  }
>  
> +static int ram_save_multifd_page(RAMState *rs, RAMBlock *block,
> +                                 ram_addr_t offset)
> +{
> +    uint8_t *p;
> +
> +    p = block->host + offset;
> +
> +    ram_counters.transferred += save_page_header(rs, rs->f, block,
> +                                                 offset | RAM_SAVE_FLAG_PAGE);
> +    multifd_queue_page(block, offset);
> +    qemu_put_buffer(rs->f, p, TARGET_PAGE_SIZE);
> +    ram_counters.transferred += TARGET_PAGE_SIZE;
> +    ram_counters.normal++;
> +
> +    return 1;
> +}
> +
>  static int do_compress_ram_page(QEMUFile *f, z_stream *stream, RAMBlock *block,
>                                  ram_addr_t offset, uint8_t *source_buf)
>  {
> @@ -2116,6 +2199,8 @@ static int ram_save_target_page(RAMState *rs, PageSearchStatus *pss,
>       */
>      if (block == rs->last_sent_block && save_page_use_compression(rs)) {
>          return compress_page_with_multi_thread(rs, block, offset);
> +    } else if (migrate_use_multifd()) {
> +        return ram_save_multifd_page(rs, block, offset);
>      }
>  
>      return ram_save_page(rs, pss, last_stage);
> -- 
> 2.17.0
> 
--
Dr. David Alan Gilbert / dgilbert@redhat.com / Manchester, UK