[PATCH 5 of 7 V4] tools/libxl: Control network buffering in remus callbacks

[Shriram Rajagopalan <rshriram@cs.ubc.ca>]

# HG changeset patch
# User Shriram Rajagopalan <rshriram@cs.ubc.ca>
# Date 1384492677 28800
# Node ID 8e5366a6cd958c00a9b7a726c149a552a02a7af6
# Parent  d3d7da1c2289749b2a1c5b8baaf9d9c18a914e9b
tools/libxl: Control network buffering in remus callbacks

This patch constitutes the core network buffering logic.
and does the following:
 a) create a new network buffer when the domain is suspended
    (remus_domain_suspend_callback)
 b) release the previous network buffer pertaining to the
    committed checkpoint (remus_domain_checkpoint_dm_saved)

Signed-off-by: Shriram Rajagopalan <rshriram@cs.ubc.ca>

diff -r d3d7da1c2289 -r 8e5366a6cd95 tools/libxl/libxl_dom.c
--- a/tools/libxl/libxl_dom.c	Thu Nov 14 21:17:56 2013 -0800
+++ b/tools/libxl/libxl_dom.c	Thu Nov 14 21:17:57 2013 -0800
@@ -1285,9 +1285,23 @@ static void libxl__remus_domain_suspend_
     libxl__save_helper_state *shs = data;
     libxl__egc *egc = shs->egc;
     libxl__domain_suspend_state *dss = CONTAINER_OF(shs, *dss, shs);
+    libxl__remus_ctx *remus_ctx = dss->remus_ctx;
+    STATE_AO_GC(dss->ao);
 
-    /* REMUS TODO: Issue disk and network checkpoint reqs. */
+    /* REMUS TODO: Issue disk checkpoint reqs. */
     int ok = libxl__domain_suspend_callback_common(dss);
+
+    if (!remus_ctx->netbuf_ctx || !ok) goto out;
+
+    /* The domain was suspended successfully. Start a new network
+     * buffer for the next epoch. If this operation fails, then act
+     * as though domain suspend failed -- libxc exits its infinite
+     * loop and ultimately, the replication stops.
+     */
+    if (libxl__remus_netbuf_start_new_epoch(gc, dss->domid,
+                                            remus_ctx))
+        ok = 0;
+ out:
     libxl__xc_domain_saverestore_async_callback_done(egc, shs, ok);
 }
 
@@ -1301,7 +1315,7 @@ static int libxl__remus_domain_resume_ca
     if (libxl__domain_resume(gc, dss->domid, /* Fast Suspend */1))
         return 0;
 
-    /* REMUS TODO: Deal with disk. Start a new network output buffer */
+    /* REMUS TODO: Deal with disk. */
     return 1;
 }
 
@@ -1310,6 +1324,9 @@ static int libxl__remus_domain_resume_ca
 static void remus_checkpoint_dm_saved(libxl__egc *egc,
                                       libxl__domain_suspend_state *dss, int rc);
 
+static void remus_next_checkpoint(libxl__egc *egc, libxl__ev_time *ev,
+                                  const struct timeval *requested_abs);
+
 static void libxl__remus_domain_checkpoint_callback(void *data)
 {
     libxl__save_helper_state *shs = data;
@@ -1328,10 +1345,51 @@ static void libxl__remus_domain_checkpoi
 static void remus_checkpoint_dm_saved(libxl__egc *egc,
                                       libxl__domain_suspend_state *dss, int rc)
 {
-    /* REMUS TODO: Wait for disk and memory ack, release network buffer */
-    /* REMUS TODO: make this asynchronous */
-    assert(!rc); /* REMUS TODO handle this error properly */
-    usleep(dss->remus_ctx->interval * 1000);
+    /*
+     * REMUS TODO: Wait for disk and explicit memory ack (through restore
+     * callback from remote) before releasing network buffer.
+     */
+    libxl__remus_ctx *remus_ctx = dss->remus_ctx;
+    STATE_AO_GC(dss->ao);
+
+    if (rc) {
+        LOG(ERROR, "Failed to save device model. Terminating Remus..");
+        goto out;
+    }
+
+    if (remus_ctx->netbuf_ctx) {
+        rc = libxl__remus_netbuf_release_prev_epoch(gc, dss->domid,
+                                                     remus_ctx);
+        if (rc) {
+            LOG(ERROR, "Failed to release network buffer."
+                " Terminating Remus..");
+            goto out;
+        }
+    }
+
+    /* Set checkpoint interval timeout */
+    rc = libxl__ev_time_register_rel(gc, &remus_ctx->timeout,
+                                     remus_next_checkpoint,
+                                     dss->remus_ctx->interval);
+    if (rc) {
+        LOG(ERROR, "unable to register timeout for next epoch."
+            " Terminating Remus..");
+        goto out;
+    }
+    return;
+
+ out:
+    libxl__xc_domain_saverestore_async_callback_done(egc, &dss->shs, 0);
+}
+
+static void remus_next_checkpoint(libxl__egc *egc, libxl__ev_time *ev,
+                                  const struct timeval *requested_abs)
+{
+    libxl__remus_ctx *remus_ctx = CONTAINER_OF(ev, *remus_ctx, timeout);
+    libxl__domain_suspend_state *dss = remus_ctx->dss;
+    STATE_AO_GC(dss->ao);
+
+    libxl__ev_time_deregister(gc, &remus_ctx->timeout);
     libxl__xc_domain_saverestore_async_callback_done(egc, &dss->shs, 1);
 }