vmspawn: let SetMemory() report progress with 'more'

With the 'more' flag, SetMemory() notifies the caller whenever the
guest's balloon moves and replies once the guest reached the target,
or once the size did not change for five seconds, since partial
convergence is normal. A newer request ends a pending call with the
new Superseded error. The reply now also carries the effective target
and the current size.

Signed-off-by: Paul Meyer <katexochen0@gmail.com>

man/systemd-vmspawn.xml

@@ -222,10 +222,13 @@
           to the host. The guest's memory may be shrunk and grown again at runtime, up to
           <replaceable>BYTES</replaceable>, via the <function>SetMemory()</function> method of the
           <literal>io.systemd.MachineInstance</literal> Varlink interface on the machine's control
-          socket. <function>Describe()</function> reports the requested and the current size. Resizing
-          is asynchronous, the guest may follow only partially, and it is permitted to deflate the
-          balloon on its own when it runs out of memory. The requested size persists across guest
-          reboots: a freshly booted guest inflates its balloon to it again.
+          socket. <function>Describe()</function> reports the requested and the current size. When
+          <function>SetMemory()</function> is called with the <literal>more</literal> flag, it reports the
+          current size whenever it changes, and returns once the guest reached the requested size, moved
+          away from it, or did not move for five seconds. Resizing is asynchronous, the guest may follow
+          only partially, and it is permitted to deflate the balloon on its own when it runs out of
+          memory. The requested size persists across guest reboots: a freshly booted guest inflates its
+          balloon to it again.
           <xi:include href="version-info.xml" xpointer="v263"/></para>
 
           <xi:include href="version-info.xml" xpointer="v255"/>

src/shared/varlink-io.systemd.MachineInstance.c

@@ -60,9 +60,13 @@ static SD_VARLINK_DEFINE_METHOD(
 
 static SD_VARLINK_DEFINE_METHOD_FULL(
                 SetMemory,
-                SD_VARLINK_EXPERIMENTAL,
-                SD_VARLINK_FIELD_COMMENT("Requested memory size in bytes. The machine converges toward it asynchronously. Describe() reports progress"),
-                SD_VARLINK_DEFINE_INPUT(bytes, SD_VARLINK_INT, 0));
+                SD_VARLINK_SUPPORTS_MORE|SD_VARLINK_EXPERIMENTAL,
+                SD_VARLINK_FIELD_COMMENT("Requested memory size in bytes"),
+                SD_VARLINK_DEFINE_INPUT(bytes, SD_VARLINK_INT, 0),
+                SD_VARLINK_FIELD_COMMENT("Memory size the machine converges toward: the requested size, rounded up to the backend's granularity"),
+                SD_VARLINK_DEFINE_OUTPUT(targetBytes, SD_VARLINK_INT, 0),
+                SD_VARLINK_FIELD_COMMENT("Memory size the machine currently sees, as last reported by the backend. null if not yet known"),
+                SD_VARLINK_DEFINE_OUTPUT(actualBytes, SD_VARLINK_INT, SD_VARLINK_NULLABLE));
 
 static SD_VARLINK_DEFINE_ERROR(NotConnected);
 static SD_VARLINK_DEFINE_ERROR(NotSupported);
@@ -71,6 +75,7 @@ static SD_VARLINK_DEFINE_ERROR(StorageExists);
 static SD_VARLINK_DEFINE_ERROR(StorageImmutable);
 static SD_VARLINK_DEFINE_ERROR(BadConfig);
 static SD_VARLINK_DEFINE_ERROR(ConfigNotSupported);
+static SD_VARLINK_DEFINE_ERROR_FULL(Superseded, SD_VARLINK_EXPERIMENTAL);
 
 SD_VARLINK_DEFINE_INTERFACE(
                 io_systemd_MachineInstance,
@@ -95,7 +100,9 @@ SD_VARLINK_DEFINE_INTERFACE(
                 &vl_method_RemoveStorage,
                 SD_VARLINK_SYMBOL_COMMENT("Replace the backing of a previously-attached storage volume in place"),
                 &vl_method_ReplaceStorage,
-                SD_VARLINK_SYMBOL_COMMENT("Resize the machine's memory. Backends without a resizing mechanism return NotSupported"),
+                SD_VARLINK_SYMBOL_COMMENT("Resize the machine's memory. Without 'more', replies once the backend accepted the request, and the machine converges asynchronously. "
+                                          "With 'more', sends a notification whenever the current size changes, and replies once it reached the target, moved away from it, or stopped moving short of it. "
+                                          "Fails with Superseded if another call replaces the target first. Backends without a resizing mechanism return NotSupported"),
                 &vl_method_SetMemory,
                 SD_VARLINK_SYMBOL_COMMENT("Memory state of a machine, all sizes in bytes"),
                 &vl_type_Memory,
@@ -112,4 +119,6 @@ SD_VARLINK_DEFINE_INTERFACE(
                 SD_VARLINK_SYMBOL_COMMENT("The supplied 'config' value is not valid for this backend"),
                 &vl_error_BadConfig,
                 SD_VARLINK_SYMBOL_COMMENT("The supplied 'config' value is recognized but not supported by this backend"),
-                &vl_error_ConfigNotSupported);
+                &vl_error_ConfigNotSupported,
+                SD_VARLINK_SYMBOL_COMMENT("The operation was replaced by a newer request before it completed"),
+                &vl_error_Superseded);

src/vmspawn/vmspawn-memory.c

@@ -1,30 +1,118 @@
 /* SPDX-License-Identifier: LGPL-2.1-or-later */
 
+#include "sd-event.h"
 #include "sd-json.h"
 #include "sd-varlink.h"
 
 #include "alloc-util.h"
+#include "event-util.h"
 #include "json-util.h"
 #include "log.h"
 #include "memory-util.h"
 #include "qmp-client.h"
 #include "string-util.h"
+#include "time-util.h"
 #include "vmspawn-memory.h"
 #include "vmspawn-qmp.h"
 
+/* Debug logs show sizes in exact KiB rather than through format_bytes(): its 0.1 precision cannot
+ * show the 64 KiB rounding of SetMemory() or a guest that stops a few MiB short of its target. */
+static const char* format_kib(char *buf, size_t l, uint64_t bytes) {
+        assert(buf);
+
+        (void) snprintf(buf, l, "%" PRIu64 " KiB", bytes / U64_KB);
+        return buf;
+}
+
+#define FORMAT_KIB_MAX (DECIMAL_STR_MAX(uint64_t) + STRLEN(" KiB"))
+#define FORMAT_KIB(bytes) format_kib((char[FORMAT_KIB_MAX]){}, FORMAT_KIB_MAX, bytes)
+
+/* Partial convergence is normal (unmovable pages, deflate-on-oom). QEMU throttles
+ * BALLOON_CHANGE to one per second, so this is several missed updates. */
+#define MEMORY_STALL_TIMEOUT_USEC (5 * USEC_PER_SEC)
+
 VmspawnMemory* vmspawn_memory_free(VmspawnMemory *m) {
         if (!m)
                 return NULL;
 
+        sd_event_source_disable_unref(m->stall_timer);
+        sd_varlink_unref(m->progress_link);
         return mfree(m);
 }
 
+static int memory_send_progress(VmspawnMemory *m, sd_varlink *link, bool final) {
+        _cleanup_(sd_json_variant_unrefp) sd_json_variant *v = NULL;
+        int r;
+
+        assert(m);
+        assert(link);
+
+        r = sd_json_buildo(
+                        &v,
+                        SD_JSON_BUILD_PAIR_UNSIGNED("targetBytes", m->target),
+                        JSON_BUILD_PAIR_UNSIGNED_NON_ZERO("actualBytes", m->actual));
+        if (r < 0)
+                return r;
+
+        return final ? sd_varlink_reply(link, v) : sd_varlink_notify(link, v);
+}
+
+/* Ends a pending SetMemory() call with 'more': with 'error' if set, else with 'error_errno' if negative,
+ * else with the final reply. */
+static void memory_end_progress(VmspawnMemory *m, const char *error, int error_errno) {
+        _cleanup_(sd_varlink_unrefp) sd_varlink *link = NULL;
+        int r;
+
+        assert(m);
+
+        m->stall_timer = sd_event_source_disable_unref(m->stall_timer);
+
+        link = TAKE_PTR(m->progress_link);
+        if (!link)
+                return;
+
+        if (error)
+                r = sd_varlink_error(link, error, /* parameters= */ NULL);
+        else if (error_errno < 0)
+                r = sd_varlink_error_errno(link, error_errno);
+        else
+                r = memory_send_progress(m, link, /* final= */ true);
+        if (r < 0)
+                log_debug_errno(r, "Failed to end SetMemory() call, ignoring: %m");
+}
+
+static int on_stall_timer(sd_event_source *s, uint64_t usec, void *userdata) {
+        VmspawnMemory *m = ASSERT_PTR(userdata);
+
+        log_debug("Guest memory size stopped at %s, short of target %s.",
+                  FORMAT_KIB(m->actual), FORMAT_KIB(m->target));
+
+        memory_end_progress(m, /* error= */ NULL, /* error_errno= */ 0);
+        return 0;
+}
+
+static int memory_progress_rearm(VmspawnMemory *m) {
+        assert(m);
+
+        return event_reset_time_relative(
+                        qmp_client_get_event(m->bridge->qmp),
+                        &m->stall_timer,
+                        CLOCK_MONOTONIC,
+                        MEMORY_STALL_TIMEOUT_USEC,
+                        /* accuracy= */ 0,
+                        on_stall_timer,
+                        m,
+                        SD_EVENT_PRIORITY_NORMAL,
+                        "vmspawn-memory-stall",
+                        /* force_reset= */ true);
+}
+
 static int memory_update_actual(VmspawnMemory *m, sd_json_variant *v) {
         static const sd_json_dispatch_field dispatch_table[] = {
                 { "actual", _SD_JSON_VARIANT_TYPE_INVALID, sd_json_dispatch_uint64, 0, SD_JSON_MANDATORY },
                 {}
         };
-        uint64_t actual;
+        uint64_t actual, previous;
         int r;
 
         assert(m);
@@ -33,11 +121,32 @@ static int memory_update_actual(VmspawnMemory *m, sd_json_variant *v) {
         if (r < 0)
                 return log_debug_errno(r, "Failed to parse balloon size reported by QEMU, ignoring: %m");
 
-        if (actual != m->actual)
-                log_debug("Guest memory size is now %" PRIu64 " KiB (target %" PRIu64 " KiB).",
-                          actual / U64_KB, m->target / U64_KB);
+        if (actual == m->actual)
+                return 0;
 
+        log_debug("Guest memory size is now %s (target %s).", FORMAT_KIB(actual), FORMAT_KIB(m->target));
+        previous = m->actual;
         m->actual = actual;
+
+        if (!m->progress_link)
+                return 0;
+
+        /* Also end the call once the guest moves away from the target, e.g. on deflate-on-oom, so that
+         * a guest moving its balloon back and forth cannot hold it open forever. */
+        if (actual == m->target ||
+            (previous > 0 && ABS_DIFF(actual, m->target) >= ABS_DIFF(previous, m->target))) {
+                memory_end_progress(m, /* error= */ NULL, /* error_errno= */ 0);
+                return 0;
+        }
+
+        r = memory_send_progress(m, m->progress_link, /* final= */ false);
+        if (r >= 0)
+                r = memory_progress_rearm(m);
+        if (r < 0) {
+                log_debug_errno(r, "Failed to continue SetMemory() call, ending it: %m");
+                memory_end_progress(m, /* error= */ NULL, r);
+        }
+
         return 0;
 }
 
@@ -65,6 +174,7 @@ typedef struct SetTargetRequest {
         VmspawnMemory *m;
         sd_varlink *link;       /* ref'd */
         uint64_t bytes;
+        bool more;
 } SetTargetRequest;
 
 static SetTargetRequest* set_target_request_free(SetTargetRequest *req) {
@@ -85,16 +195,29 @@ static int on_balloon_complete(
                 void *userdata) {
 
         _cleanup_(set_target_request_freep) SetTargetRequest *req = ASSERT_PTR(userdata);
+        VmspawnMemory *m = req->m;
+        int r;
 
         assert(client);
 
         if (error < 0)
                 return vmspawn_qmp_reply_error(req->link, error_desc, error);
 
-        req->m->target = req->bytes;
-        log_debug("Requested guest memory size %" PRIu64 " KiB.", req->bytes / U64_KB);
+        m->target = req->bytes;
+        log_debug("Requested guest memory size %s.", FORMAT_KIB(req->bytes));
 
-        return sd_varlink_reply(req->link, /* parameters= */ NULL);
+        /* There is one target at a time, so a call still waiting for the previous one ends here. */
+        memory_end_progress(m, "io.systemd.MachineInstance.Superseded", /* error_errno= */ 0);
+
+        if (!req->more || m->actual == m->target)
+                return memory_send_progress(m, req->link, /* final= */ true);
+
+        r = memory_progress_rearm(m);
+        if (r < 0)
+                return sd_varlink_error_errno(req->link, r);
+
+        m->progress_link = TAKE_PTR(req->link);
+        return 0;
 }
 
 static int memory_refresh(VmspawnMemory *m) {
@@ -112,7 +235,7 @@ static int memory_refresh(VmspawnMemory *m) {
                                  m);
 }
 
-int vmspawn_memory_set_target(VmspawnMemory *m, sd_varlink *link, uint64_t bytes) {
+int vmspawn_memory_set_target(VmspawnMemory *m, sd_varlink *link, bool more, uint64_t bytes) {
         _cleanup_(set_target_request_freep) SetTargetRequest *req = NULL;
         _cleanup_(sd_json_variant_unrefp) sd_json_variant *args = NULL;
         int r;
@@ -143,6 +266,7 @@ int vmspawn_memory_set_target(VmspawnMemory *m, sd_varlink *link, uint64_t bytes
                 .m = m,
                 .link = sd_varlink_ref(link),
                 .bytes = bytes,
+                .more = more,
         };
 
         r = qmp_client_invoke(m->bridge->qmp,
@@ -186,6 +310,13 @@ int vmspawn_memory_setup(VmspawnQmpBridge *bridge, uint64_t configured, bool bal
         return 0;
 }
 
+void vmspawn_memory_disconnect(VmspawnMemory *m) {
+        if (!m)
+                return;
+
+        memory_end_progress(m, "io.systemd.MachineInstance.NotConnected", /* error_errno= */ 0);
+}
+
 int vmspawn_memory_dispatch_balloon_change(VmspawnMemory *m, sd_json_variant *data) {
         assert(m);
 

src/vmspawn/vmspawn-memory.h

@@ -10,6 +10,8 @@ struct VmspawnMemory {
         uint64_t configured;            /* boot RAM as handed to QEMU */
         uint64_t target;                /* last requested size */
         uint64_t actual;                /* last reported size, 0 = not yet known */
+        sd_varlink *progress_link;      /* SetMemory() call with 'more' waiting for convergence */
+        sd_event_source *stall_timer;   /* ends progress_link once actual stopped moving */
         bool balloon;                   /* virtio-balloon device present */
 };
 
@@ -19,9 +21,15 @@ DEFINE_TRIVIAL_CLEANUP_FUNC(VmspawnMemory*, vmspawn_memory_free);
 /* Allocate the per-VM memory state on the bridge and query the balloon size. */
 int vmspawn_memory_setup(VmspawnQmpBridge *bridge, uint64_t configured, bool balloon);
 
-/* Resize the guest memory to 'bytes', rounded up to 64 KiB, the largest guest page size. 'link' is answered
- * once the VMM accepted or rejected the request. The guest converges afterwards, observable via Describe(). */
-int vmspawn_memory_set_target(VmspawnMemory *m, sd_varlink *link, uint64_t bytes);
+/* Resize the guest memory to 'bytes', rounded up to 64 KiB, the largest guest page size. Without 'more', 'link'
+ * gets the effective target and the current size once the VMM accepted the request. With 'more', it gets a
+ * notification per change of the current size, and that final reply once the guest reached the target, moved
+ * away from it, or did not move for 5s. A newer accepted request ends a still pending call with Superseded.
+ * 'link' gets an error if the VMM rejects the request. */
+int vmspawn_memory_set_target(VmspawnMemory *m, sd_varlink *link, bool more, uint64_t bytes);
+
+/* End a pending SetMemory() call with NotConnected, as its size can no longer change. */
+void vmspawn_memory_disconnect(VmspawnMemory *m);
 
 /* Update ->actual from the payload of a BALLOON_CHANGE event. Returns negative if malformed. */
 int vmspawn_memory_dispatch_balloon_change(VmspawnMemory *m, sd_json_variant *data);

src/vmspawn/vmspawn-varlink.c

@@ -314,7 +314,7 @@ static int vl_method_set_memory(sd_varlink *link, sd_json_variant *parameters, s
         if (r != 0)
                 return r;
 
-        r = vmspawn_memory_set_target(ctx->bridge->memory, link, bytes);
+        r = vmspawn_memory_set_target(ctx->bridge->memory, link, FLAGS_SET(flags, SD_VARLINK_METHOD_MORE), bytes);
         if (r == -EOPNOTSUPP)
                 return sd_varlink_error(link, "io.systemd.MachineInstance.NotSupported", /* parameters= */ NULL);
         if (r == -ERANGE)
@@ -448,6 +448,8 @@ static void on_qmp_disconnect(QmpClient *client, void *userdata) {
 
         /* Propagate connection loss by closing all subscriber connections */
         drain_event_subscribers(&ctx->subscribed);
+
+        vmspawn_memory_disconnect(ctx->bridge->memory);
 }
 
 int vmspawn_varlink_setup(