[PATCH 3 of 4 RFC] xl/remus: Control network buffering in remus callbacks

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

This patch constitutes the core network buffering logic.
Libxl would receive a list of IFB devices that collectively act as
network buffering devices, for a given guest. Also, libxl expects that
every ifb device in the list supplied by a toolstack (netbuf_iflist)
should have a plug qdisc installed.

This patch does the following:
 a) establish a dedicated remus context containing libnl related
    state (netlink sockets, qdisc caches, etc.,)
 b) Obtain handles to plug qdiscs installed on the supplied list of
    IFB devices
 c) add a new network buffer (i.e., create a new one) when the domain is
    suspended (remus_domain_suspend_callback)
 d) release the network buffer pertaining to the acknowledged checkpoint
    in remus_domain_checkpoint_dm_saved, which is invoked for both PV & HVM.

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

diff -r 3cd67f6ff63a -r bef729fc4336 tools/libxl/libxl_dom.c
--- a/tools/libxl/libxl_dom.c	Thu Jul 25 00:02:19 2013 -0700
+++ b/tools/libxl/libxl_dom.c	Thu Jul 25 00:02:22 2013 -0700
@@ -20,6 +20,7 @@
 #include "libxl_internal.h"
 #include "libxl_arch.h"
 
+#include <netlink/route/qdisc/plug.h>
 #include <xc_dom.h>
 #include <xen/hvm/hvm_info_table.h>
 #include <xen/hvm/hvm_xs_strings.h>
@@ -1212,12 +1213,36 @@ int libxl__toolstack_save(uint32_t domid
 
 /*----- remus callbacks -----*/
 
+/* REMUS TODO: Issue disk checkpoint reqs. */
 static int libxl__remus_domain_suspend_callback(void *data)
 {
-    /* REMUS TODO: Issue disk and network checkpoint reqs. */
-    return libxl__domain_suspend_common_callback(data);
+    libxl__save_helper_state *shs = data;
+    libxl__domain_suspend_state *dss = CONTAINER_OF(shs, *dss, shs);
+    STATE_AO_GC(dss->ao);
+
+    int is_suspended = 0, i, ret;
+    libxl__remus_ctx *remus_ctx = dss->remus_ctx;
+    is_suspended = libxl__domain_suspend_common_callback(data);
+    if (!remus_ctx->num_netbufs) return is_suspended;
+
+    if (is_suspended) {
+        for (i = 0; i < remus_ctx->num_netbufs; ++i) {
+            ret = rtnl_qdisc_plug_buffer(remus_ctx->netbuf_qdisc_list[i]);
+            if (!ret)
+                ret = rtnl_qdisc_add(remus_ctx->nlsock, remus_ctx->netbuf_qdisc_list[i],
+                                     NLM_F_REQUEST);
+            if (ret) {
+                LOG(ERROR, "Cannot create new buffer on %s:%s",
+                    dss->remus->netbuf_iflist[i], nl_geterror(ret));
+                return 0;
+            }
+        }
+    }
+
+    return is_suspended;
 }
 
+/* REMUS TODO: Deal with disk. */
 static int libxl__remus_domain_resume_callback(void *data)
 {
     libxl__save_helper_state *shs = data;
@@ -1228,7 +1253,6 @@ 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 */
     return 1;
 }
 
@@ -1255,13 +1279,104 @@ 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->interval * 1000);
+    /* REMUS TODO: Wait for disk and explicit memory ack (through restore
+       callback from remote) before release network buffer. */
+    int i, ret;
+    libxl__remus_ctx *remus_ctx = dss->remus_ctx;
+    STATE_AO_GC(dss->ao);
+
+    assert(!rc);
+
+    if (remus_ctx->num_netbufs  > 0) {
+        for (i = 0; i < remus_ctx->num_netbufs; ++i) {
+            ret = rtnl_qdisc_plug_release_one(remus_ctx->netbuf_qdisc_list[i]);
+            if (!ret)
+                ret = rtnl_qdisc_add(remus_ctx->nlsock, remus_ctx->netbuf_qdisc_list[i],
+                                     NLM_F_REQUEST);
+            if (ret) {
+                LOG(ERROR, "Cannot release buffer from %s:%s",
+                    dss->remus->netbuf_iflist[i], nl_geterror(ret));
+                ret= 0;
+                break;
+            }
+        }
+    }
+
+    usleep(dss->remus_ctx->interval * 1000);
     libxl__xc_domain_saverestore_async_callback_done(egc, &dss->shs, 1);
 }
 
+static libxl__remus_ctx *setup_remus_ctx(libxl__gc *gc,
+                                         const libxl_domain_remus_info *r_info)
+{
+    libxl__remus_ctx *remus_ctx = NULL;
+    int num_netbufs = 0, i;
+    struct nl_cache *qdisc_cache = NULL;
+    struct rtnl_link *ifb = NULL;
+    struct nl_sock *nlsock = NULL;
+    struct rtnl_qdisc *qdisc = NULL;
+    libxl_string_list l;
+    int ifindex;
+
+    remus_ctx = libxl__zalloc(gc, sizeof(libxl__remus_ctx));
+    remus_ctx->interval = r_info->interval;
+
+    l = r_info->netbuf_iflist;
+    num_netbufs = libxl_string_list_length(&l);
+    if (!num_netbufs) return remus_ctx;
+
+    nlsock = nl_socket_alloc();
+    if (!nlsock) {
+        LOG(ERROR, "setup_remus_ctx: cannot allocate nl socket");
+        return NULL;
+    }
+
+    nl_connect(nlsock, NETLINK_ROUTE);
+
+    if (rtnl_qdisc_alloc_cache(nlsock, &qdisc_cache) < 0) {
+        LOG(ERROR, "setup_remus_ctx: failed to allocate qdisc cache");
+        goto end;
+    }
+
+    remus_ctx->netbuf_qdisc_list = libxl__calloc(gc, num_netbufs + 1, 
+                                                 sizeof(struct rtnl_qdisc *));
+    remus_ctx->num_netbufs = num_netbufs;
+    remus_ctx->nlsock = nlsock;
+    remus_ctx->qdisc_cache = qdisc_cache;
+    ifb = rtnl_link_alloc();
+
+    for (i = 0; i < num_netbufs; ++i) {
+
+        if (rtnl_link_get_kernel(nlsock, 0, l[i], &ifb) < 0) {
+            LOG(ERROR, "setup_remus_ctx: cannot obtain handle for %s", l[i]);
+            goto end;
+        }
+
+        ifindex = rtnl_link_get_ifindex(ifb);
+        if (!ifindex) {
+            LOG(ERROR, "setup_remus_ctx: invalid interface %s", l[i]);
+            goto end;
+        }
+
+        qdisc = rtnl_qdisc_get_by_parent(qdisc_cache, ifindex, TC_H_ROOT);
+        if (!qdisc || strcmp(rtnl_tc_get_kind(TC_CAST(qdisc)), "plug")) {
+            LOG(ERROR, "setup_remus_ctx: plug qdisc is not installed on %s", l[i]);
+            goto end;
+        }
+
+        remus_ctx->netbuf_qdisc_list[i] = qdisc;
+    }
+
+    rtnl_link_put(ifb);
+    return remus_ctx;
+
+ end:
+    if (ifb) rtnl_link_put(ifb);
+    if (qdisc_cache) nl_cache_free(qdisc_cache);
+    if (nlsock) nl_close(nlsock);
+    return NULL;
+}
+
 /*----- main code for suspending, in order of execution -----*/
 
 void libxl__domain_suspend(libxl__egc *egc, libxl__domain_suspend_state *dss)
@@ -1312,7 +1427,12 @@ void libxl__domain_suspend(libxl__egc *e
     dss->dm_savefile = libxl__device_model_savefile(gc, domid);
 
     if (r_info != NULL) {
-        dss->interval = r_info->interval;
+        /* This suspend is for Remus. We need to get a handle on
+         * the network output buffers and setup the remus_ctx;
+         */
+        dss->remus_ctx = setup_remus_ctx(gc, r_info);
+        if (!dss->remus_ctx)
+            goto out;
         if (r_info->compression)
             dss->xcflags |= XCFLAGS_CHECKPOINT_COMPRESS;
     }
diff -r 3cd67f6ff63a -r bef729fc4336 tools/libxl/libxl_internal.h
--- a/tools/libxl/libxl_internal.h	Thu Jul 25 00:02:19 2013 -0700
+++ b/tools/libxl/libxl_internal.h	Thu Jul 25 00:02:22 2013 -0700
@@ -44,6 +44,13 @@
 #include <sys/wait.h>
 #include <sys/socket.h>
 
+#include <netlink/cache.h>
+#include <netlink/socket.h>
+#include <netlink/attr.h>
+#include <netlink/route/link.h>
+#include <netlink/route/route.h>
+#include <netlink/route/qdisc.h>
+
 #include <xenstore.h>
 #include <xenctrl.h>
 #include <xenguest.h>
@@ -2242,6 +2249,18 @@ typedef struct libxl__logdirty_switch {
     libxl__ev_time timeout;
 } libxl__logdirty_switch;
 
+typedef struct libxl__remus_ctx {
+    /* checkpoint interval */
+    int interval;
+    /* array of plug qdisc pointers, that hold
+     * network output from the guest's vifs.
+     */
+    int num_netbufs;
+    struct rtnl_qdisc **netbuf_qdisc_list;
+    struct nl_sock *nlsock;
+    struct nl_cache *qdisc_cache;
+} libxl__remus_ctx;
+
 struct libxl__domain_suspend_state {
     /* set by caller of libxl__domain_suspend */
     libxl__ao *ao;
@@ -2260,7 +2279,7 @@ struct libxl__domain_suspend_state {
     int xcflags;
     int guest_responded;
     const char *dm_savefile;
-    int interval; /* checkpoint interval (for Remus) */
+    libxl__remus_ctx *remus_ctx;
     libxl__save_helper_state shs;
     libxl__logdirty_switch logdirty;
     /* private for libxl__domain_save_device_model */