[Qemu-devel] [PATCH 5/5] virtio-console: Throttle virtio-serial-bus if we can't consume any more guest data

[Amit Shah]

If the char device we're connected to is overwhelmed with data and it
can't accept any more, signal to the virtio-serial-bus to stop sending
us more data till we tell otherwise.

If the current buffer being processed hasn't been completely written out
to the char device, we have to keep it around and re-try sending it
since the virtio-serial-bus code assumes we consume the entire buffer.

Also register with savevm so that we save/restore such a buffer across
migration.

Signed-off-by: Amit Shah <address@hidden>
---
 hw/virtio-console.c |  119 +++++++++++++++++++++++++++++++++++++++++++++++++-
 1 files changed, 116 insertions(+), 3 deletions(-)

diff --git a/hw/virtio-console.c b/hw/virtio-console.c
index dc1f75f..91c5171 100644
--- a/hw/virtio-console.c
+++ b/hw/virtio-console.c
@@ -13,18 +13,84 @@
 #include "qemu-char.h"
 #include "virtio-serial.h"
 
+typedef struct Buffer {
+    uint8_t *buf;
+    size_t rem_len;
+    size_t offset;
+} Buffer;
+
 typedef struct VirtConsole {
     VirtIOSerialPort port;
     CharDriverState *chr;
+    QEMUTimer *host_timer;
+    Buffer *unflushed_buf;
 } VirtConsole;
 
+static void add_unflushed_buf(VirtConsole *vcon, const uint8_t *buf, size_t 
len)
+{
+    vcon->unflushed_buf = qemu_malloc(sizeof(Buffer));
+    vcon->unflushed_buf->buf = qemu_malloc(len);
+
+    memcpy(vcon->unflushed_buf->buf, buf, len);
+    vcon->unflushed_buf->rem_len = len;
+    vcon->unflushed_buf->offset = 0;
+}
+
+static void free_unflushed_buf(VirtConsole *vcon)
+{
+    qemu_free(vcon->unflushed_buf->buf);
+    qemu_free(vcon->unflushed_buf);
+    vcon->unflushed_buf = NULL;
+}
+
+/* Callback function when the timer expires */
+static void unthrottle_port(VirtConsole *vcon)
+{
+    if (vcon->unflushed_buf) {
+        ssize_t ret;
+        size_t len;
+
+        len = vcon->unflushed_buf->rem_len;
+        ret = qemu_chr_write(vcon->chr,
+                             vcon->unflushed_buf->buf
+                             + vcon->unflushed_buf->offset, &len);
+
+        if (vcon->unflushed_buf->rem_len) {
+            vcon->unflushed_buf->offset += len;
+            vcon->unflushed_buf->rem_len -= len;
+
+            qemu_mod_timer(vcon->host_timer,
+                           qemu_get_clock(host_clock) + (int64_t) 100000);
+            return;
+        }
+        free_unflushed_buf(vcon);
+    }
+    virtio_serial_throttle_port(&vcon->port, false);
+}
 
 /* Callback function that's called when the guest sends us data */
 static void flush_buf(VirtIOSerialPort *port, const uint8_t *buf, size_t len)
 {
     VirtConsole *vcon = DO_UPCAST(VirtConsole, port, port);
 
-    qemu_chr_write(vcon->chr, buf, &len);
+    ssize_t ret;
+    size_t written;
+
+    written = len;
+    ret = qemu_chr_write(vcon->chr, buf, &written);
+    if (len > written) {
+        virtio_serial_throttle_port(port, true);
+
+        /*
+         * We shouldn't be called while there are pending buffers to
+         * be flushed.
+         */
+        assert(!vcon->unflushed_buf);
+
+        add_unflushed_buf(vcon, buf + written, len - written);
+        qemu_mod_timer(vcon->host_timer,
+                       qemu_get_clock(host_clock) + (int64_t) 100000);
+    }
 }
 
 /* Readiness of the guest to accept data on a port */
@@ -48,16 +114,59 @@ static void chr_event(void *opaque, int event)
     VirtConsole *vcon = opaque;
 
     switch (event) {
-    case CHR_EVENT_OPENED: {
+    case CHR_EVENT_OPENED:
         virtio_serial_open(&vcon->port);
         break;
-    }
+
     case CHR_EVENT_CLOSED:
+        if (vcon->unflushed_buf) {
+            qemu_del_timer(vcon->host_timer);
+            free_unflushed_buf(vcon);
+        }
         virtio_serial_close(&vcon->port);
         break;
     }
 }
 
+static void virtio_console_port_save(QEMUFile *f, void *opaque)
+{
+    VirtConsole *vcon = opaque;
+    uint32_t have_buffer;
+
+    have_buffer = vcon->unflushed_buf ? true : false;
+
+    qemu_put_be32s(f, &have_buffer);
+    if (have_buffer) {
+        qemu_put_be64s(f, &vcon->unflushed_buf->rem_len);
+        qemu_put_buffer(f, vcon->unflushed_buf->buf
+                           + vcon->unflushed_buf->offset,
+                        vcon->unflushed_buf->rem_len);
+    }
+}
+
+static int virtio_console_port_load(QEMUFile *f, void *opaque, int version_id)
+{
+    VirtConsole *vcon = opaque;
+    uint32_t have_buffer;
+
+    if (version_id > 1) {
+        return -EINVAL;
+    }
+
+    qemu_get_be32s(f, &have_buffer);
+    if (have_buffer) {
+        vcon->unflushed_buf = qemu_mallocz(sizeof(Buffer));
+
+        qemu_get_be64s(f, &vcon->unflushed_buf->rem_len);
+        vcon->unflushed_buf->buf = qemu_malloc(vcon->unflushed_buf->rem_len);
+        vcon->unflushed_buf->offset = 0;
+
+        qemu_get_buffer(f, vcon->unflushed_buf->buf,
+                        vcon->unflushed_buf->rem_len);
+    }
+    return 0;
+}
+
 static int generic_port_init(VirtConsole *vcon, VirtIOSerialDevice *dev)
 {
     vcon->port.info = dev->info;
@@ -67,6 +176,9 @@ static int generic_port_init(VirtConsole *vcon, 
VirtIOSerialDevice *dev)
                               vcon);
         vcon->port.info->have_data = flush_buf;
     }
+    vcon->host_timer = qemu_new_timer(host_clock, (void *)unthrottle_port, 
vcon);
+    register_savevm("virtio-console-ports", -1, 1, virtio_console_port_save,
+                   virtio_console_port_load, vcon);
     return 0;
 }
 
@@ -89,6 +201,7 @@ static int virtconsole_exitfn(VirtIOSerialDevice *dev)
         port->info->have_data = NULL;
         qemu_chr_close(vcon->chr);
     }
+    qemu_free_timer(vcon->host_timer);
 
     return 0;
 }
-- 
1.6.2.5