[Qemu-devel] [PATCH 2/3] virtio-serial: Add support for flow control

[Amit Shah <amit.shah@redhat.com>]

This commit lets apps signal an incomplete write.  When that happens,
stop stop sending out any more data to the app and wait for it to
unthrottle the port.

This involves storing some extra state in the port struct.  This state
will have to be migrated along in a new subsection.  That's an item for
the TODO list.

Signed-off-by: Amit Shah <amit.shah@redhat.com>
---
 hw/virtio-serial-bus.c |   50 +++++++++++++++++++++++++++++++++++++----------
 hw/virtio-serial.h     |   17 ++++++++++++++++
 2 files changed, 56 insertions(+), 11 deletions(-)

diff --git a/hw/virtio-serial-bus.c b/hw/virtio-serial-bus.c
index 6726f72..c5ce60b 100644
--- a/hw/virtio-serial-bus.c
+++ b/hw/virtio-serial-bus.c
@@ -126,24 +126,49 @@ static void discard_vq_data(VirtQueue *vq, VirtIODevice *vdev)
 static void do_flush_queued_data(VirtIOSerialPort *port, VirtQueue *vq,
                                  VirtIODevice *vdev)
 {
-    VirtQueueElement elem;
-
     assert(port);
     assert(virtio_queue_ready(vq));
 
-    while (!port->throttled && virtqueue_pop(vq, &elem)) {
+    while (!port->throttled) {
         unsigned int i;
 
-        for (i = 0; i < elem.out_num; i++) {
-            size_t buf_size;
-
-            buf_size = elem.out_sg[i].iov_len;
+        /* Pop an elem only if we haven't left off a previous one mid-way */
+        if (!port->elem.out_num) {
+            if (!virtqueue_pop(vq, &port->elem)) {
+                break;
+            }
+            port->iov_idx = 0;
+            port->iov_offset = 0;
+        }
 
-            port->info->have_data(port,
-                                  elem.out_sg[i].iov_base,
-                                  buf_size);
+        for (i = port->iov_idx; i < port->elem.out_num; i++) {
+            size_t buf_size;
+            ssize_t ret;
+
+            buf_size = port->elem.out_sg[i].iov_len - port->iov_offset;
+            ret = port->info->have_data(port,
+                                        port->elem.out_sg[i].iov_base
+                                          + port->iov_offset,
+                                        buf_size);
+            if (ret < 0 && ret != -EAGAIN) {
+                /* We don't handle any other type of errors here */
+                abort();
+            }
+            if (ret == -EAGAIN || (ret >= 0 && ret < buf_size)) {
+                virtio_serial_throttle_port(port, true);
+                port->iov_idx = i;
+                if (ret > 0) {
+                    port->iov_offset += ret;
+                }
+                break;
+            }
+            port->iov_offset = 0;
         }
-        virtqueue_push(vq, &elem, 0);
+        if (port->throttled) {
+            break;
+        }
+        virtqueue_push(vq, &port->elem, 0);
+        port->elem.out_num = 0;
     }
     virtio_notify(vdev, vq);
 }
@@ -505,6 +530,7 @@ static void virtio_serial_save(QEMUFile *f, void *opaque)
         qemu_put_be32s(f, &port->id);
         qemu_put_byte(f, port->guest_connected);
         qemu_put_byte(f, port->host_connected);
+        /* TODO: Save & load port->elem and port->iov* */
     }
 }
 
@@ -709,6 +735,8 @@ static int virtser_port_qdev_init(DeviceState *qdev, DeviceInfo *base)
         port->guest_connected = true;
     }
 
+    port->elem.out_num = 0;
+
     QTAILQ_INSERT_TAIL(&port->vser->ports, port, next);
     port->ivq = port->vser->ivqs[port->id];
     port->ovq = port->vser->ovqs[port->id];
diff --git a/hw/virtio-serial.h b/hw/virtio-serial.h
index 9cc0fb3..5f41706 100644
--- a/hw/virtio-serial.h
+++ b/hw/virtio-serial.h
@@ -102,6 +102,23 @@ struct VirtIOSerialPort {
      */
     uint32_t id;
 
+    /*
+     * This is the elem that we pop from the virtqueue.  A slow
+     * backend that consumes guest data (e.g. the file backend for
+     * qemu chardevs) can cause the guest to block till all the output
+     * is flushed.  This isn't desired, so we keep a note of the last
+     * element popped and continue consuming it once the backend
+     * becomes writable again.
+     */
+    VirtQueueElement elem;
+
+    /*
+     * The index and the offset into the iov buffer that was popped in
+     * elem above.
+     */
+    unsigned int iov_idx;
+    size_t iov_offset;
+
     /* Identify if this is a port that binds with hvc in the guest */
     uint8_t is_console;
 
-- 
1.7.3.4