[Qemu-devel] [PATCH v3 0/4] char: write callback, virtio-console: flow control

[Qemu-devel] [PATCH v3 0/4] char: write callback, virtio-console: flow control

[Amit Shah]

Hello,

This series lets interested callers ask for an -EAGAIN return from the
chardev backends (only unix and tcp sockets as of now) to implement
their own flow control.

Support for other backend types is easy to add, I'll do that in a
separate series if this is acceptable.

Amit Shah (4):
  char: Let writers know how much data was written in case of errors
  char: Add ability to provide a callback when write won't return
    -EAGAIN
  virtio-console: Factor out common init between console and generic
    ports
  virtio-console: Throttle virtio-serial-bus if we can't consume any
    more guest data

 gdbstub.c            |    2 +-
 hw/debugcon.c        |    2 +-
 hw/escc.c            |    3 +-
 hw/etraxfs_ser.c     |    4 +-
 hw/mcf_uart.c        |    2 +-
 hw/pl011.c           |    2 +-
 hw/pxa2xx.c          |    2 +-
 hw/serial.c          |    2 +-
 hw/sh_serial.c       |    2 +-
 hw/syborg_serial.c   |    3 +-
 hw/usb-serial.c      |    2 +-
 hw/virtio-console.c  |  156 +++++++++++++++++++++++++++++++++++++++++++-------
 hw/xen_console.c     |    7 +-
 hw/xilinx_uartlite.c |    5 +-
 monitor.c            |    4 +-
 net/slirp.c          |    2 +-
 net/socket.c         |    4 +-
 qemu-char.c          |   68 ++++++++++++++++++----
 qemu-char.h          |    6 ++
 qemu_socket.h        |    3 +-
 20 files changed, 227 insertions(+), 54 deletions(-)

[Qemu-devel] [PATCH v3 1/4] char: Let writers know how much data was written in case of errors

[Amit Shah]

On writing errors, we just returned -1 even if some bytes were already
written out. Ensure we return the number of bytes written before we
return the error (on a subsequent call to qemu_chr_write()).

Signed-off-by: Amit Shah <amit.shah@redhat.com>
---
 qemu-char.c |    9 +++++++--
 1 files changed, 7 insertions(+), 2 deletions(-)

diff --git a/qemu-char.c b/qemu-char.c
index 05df971..6aaa362 100644
--- a/qemu-char.c
+++ b/qemu-char.c
@@ -522,8 +522,13 @@ static int unix_write(int fd, const uint8_t *buf, int len1)
     while (len > 0) {
         ret = write(fd, buf, len);
         if (ret < 0) {
-            if (errno != EINTR && errno != EAGAIN)
-                return -1;
+            if (errno != EINTR && errno != EAGAIN) {
+                if (len1 - len) {
+                    return len1 - len;
+                } else {
+                    return -1;
+                }
+            }
         } else if (ret == 0) {
             break;
         } else {
-- 
1.6.2.5

[Qemu-devel] [PATCH v3 2/4] char: Add ability to provide a callback when write won't return -EAGAIN

[Amit Shah]

For nonblocking chardevs, we can return -EAGAIN to callers of
qemu_chr_write() and let the caller know that the backend is ready to
accept more data via a callback.

Currently only unix and tcp sockets have the ability to return -EAGAIN
and to provide such a callback.

Update all callers of qemu chardevs to not register the callback
handler so that the current behaviour of qemu_chr_write() not returning
till the entire buffer is written out is maintained.

Individual callers can update their code to add a callback handler and
register the handler at the time of calling qemu_chr_add_handlers() if
they wish to receive -EAGAIN notifications.

Signed-off-by: Amit Shah <amit.shah@redhat.com>
---
 gdbstub.c            |    2 +-
 hw/debugcon.c        |    2 +-
 hw/escc.c            |    3 +-
 hw/etraxfs_ser.c     |    4 +-
 hw/mcf_uart.c        |    2 +-
 hw/pl011.c           |    2 +-
 hw/pxa2xx.c          |    2 +-
 hw/serial.c          |    2 +-
 hw/sh_serial.c       |    2 +-
 hw/syborg_serial.c   |    3 +-
 hw/usb-serial.c      |    2 +-
 hw/virtio-console.c  |    8 +++---
 hw/xen_console.c     |    7 +++--
 hw/xilinx_uartlite.c |    5 ++-
 monitor.c            |    4 +-
 net/slirp.c          |    2 +-
 net/socket.c         |    4 +-
 qemu-char.c          |   59 ++++++++++++++++++++++++++++++++++++++++++-------
 qemu-char.h          |    6 +++++
 qemu_socket.h        |    3 +-
 20 files changed, 88 insertions(+), 36 deletions(-)

diff --git a/gdbstub.c b/gdbstub.c
index 93c4850..d6e5b23 100644
--- a/gdbstub.c
+++ b/gdbstub.c
@@ -2650,7 +2650,7 @@ int gdbserver_start(const char *device)
         if (!chr)
             return -1;
 
-        qemu_chr_add_handlers(chr, gdb_chr_can_receive, gdb_chr_receive,
+        qemu_chr_add_handlers(chr, gdb_chr_can_receive, gdb_chr_receive, NULL,
                               gdb_chr_event, NULL);
     }
 
diff --git a/hw/debugcon.c b/hw/debugcon.c
index 5ee6821..4fa9189 100644
--- a/hw/debugcon.c
+++ b/hw/debugcon.c
@@ -73,7 +73,7 @@ static void debugcon_init_core(DebugconState *s)
         exit(1);
     }
 
-    qemu_chr_add_handlers(s->chr, NULL, NULL, NULL, s);
+    qemu_chr_add_handlers(s->chr, NULL, NULL, NULL, NULL, s);
 }
 
 static int debugcon_isa_initfn(ISADevice *dev)
diff --git a/hw/escc.c b/hw/escc.c
index 6d2fd36..94d1ba7 100644
--- a/hw/escc.c
+++ b/hw/escc.c
@@ -904,7 +904,8 @@ static int escc_init1(SysBusDevice *dev)
         s->chn[i].clock = s->frequency / 2;
         if (s->chn[i].chr) {
             qemu_chr_add_handlers(s->chn[i].chr, serial_can_receive,
-                                  serial_receive1, serial_event, &s->chn[i]);
+                                  serial_receive1, NULL, serial_event,
+                                  &s->chn[i]);
         }
     }
     s->chn[0].otherchn = &s->chn[1];
diff --git a/hw/etraxfs_ser.c b/hw/etraxfs_ser.c
index e1f9615..5e592c9 100644
--- a/hw/etraxfs_ser.c
+++ b/hw/etraxfs_ser.c
@@ -176,8 +176,8 @@ static int etraxfs_ser_init(SysBusDevice *dev)
     s->chr = qdev_init_chardev(&dev->qdev);
     if (s->chr)
         qemu_chr_add_handlers(s->chr,
-                      serial_can_receive, serial_receive,
-                      serial_event, s);
+                              serial_can_receive, serial_receive, NULL,
+                              serial_event, s);
     return 0;
 }
 
diff --git a/hw/mcf_uart.c b/hw/mcf_uart.c
index d16bac7..5cea623 100644
--- a/hw/mcf_uart.c
+++ b/hw/mcf_uart.c
@@ -277,7 +277,7 @@ void *mcf_uart_init(qemu_irq irq, CharDriverState *chr)
     s->irq = irq;
     if (chr) {
         qemu_chr_add_handlers(chr, mcf_uart_can_receive, mcf_uart_receive,
-                              mcf_uart_event, s);
+                              NULL, mcf_uart_event, s);
     }
     mcf_uart_reset(s);
     return s;
diff --git a/hw/pl011.c b/hw/pl011.c
index 81de91e..73ae086 100644
--- a/hw/pl011.c
+++ b/hw/pl011.c
@@ -304,7 +304,7 @@ static int pl011_init(SysBusDevice *dev, const unsigned char *id)
     s->flags = 0x90;
     if (s->chr) {
         qemu_chr_add_handlers(s->chr, pl011_can_receive, pl011_receive,
-                              pl011_event, s);
+                              NULL, pl011_event, s);
     }
     register_savevm("pl011_uart", -1, 1, pl011_save, pl011_load, s);
     return 0;
diff --git a/hw/pxa2xx.c b/hw/pxa2xx.c
index 9095386..04dbda4 100644
--- a/hw/pxa2xx.c
+++ b/hw/pxa2xx.c
@@ -2012,7 +2012,7 @@ static PXA2xxFIrState *pxa2xx_fir_init(target_phys_addr_t base,
 
     if (chr)
         qemu_chr_add_handlers(chr, pxa2xx_fir_is_empty,
-                        pxa2xx_fir_rx, pxa2xx_fir_event, s);
+                              pxa2xx_fir_rx, NULL, pxa2xx_fir_event, s);
 
     register_savevm("pxa2xx_fir", 0, 0, pxa2xx_fir_save, pxa2xx_fir_load, s);
 
diff --git a/hw/serial.c b/hw/serial.c
index 90213c4..0320494 100644
--- a/hw/serial.c
+++ b/hw/serial.c
@@ -738,7 +738,7 @@ static void serial_init_core(SerialState *s)
 
     qemu_register_reset(serial_reset, s);
 
-    qemu_chr_add_handlers(s->chr, serial_can_receive1, serial_receive1,
+    qemu_chr_add_handlers(s->chr, serial_can_receive1, serial_receive1, NULL,
                           serial_event, s);
 }
 
diff --git a/hw/sh_serial.c b/hw/sh_serial.c
index 93dc144..2c81798 100644
--- a/hw/sh_serial.c
+++ b/hw/sh_serial.c
@@ -403,7 +403,7 @@ void sh_serial_init (target_phys_addr_t base, int feat,
 
     if (chr)
         qemu_chr_add_handlers(chr, sh_serial_can_receive1, sh_serial_receive1,
-			      sh_serial_event, s);
+                              NULL, sh_serial_event, s);
 
     s->eri = eri_source;
     s->rxi = rxi_source;
diff --git a/hw/syborg_serial.c b/hw/syborg_serial.c
index cac00ea..9cf3591 100644
--- a/hw/syborg_serial.c
+++ b/hw/syborg_serial.c
@@ -327,7 +327,8 @@ static int syborg_serial_init(SysBusDevice *dev)
     s->chr = qdev_init_chardev(&dev->qdev);
     if (s->chr) {
         qemu_chr_add_handlers(s->chr, syborg_serial_can_receive,
-                              syborg_serial_receive, syborg_serial_event, s);
+                              syborg_serial_receive, NULL,
+                              syborg_serial_event, s);
     }
     if (s->fifo_size <= 0) {
         fprintf(stderr, "syborg_serial: fifo too small\n");
diff --git a/hw/usb-serial.c b/hw/usb-serial.c
index 69f0e44..c44d4b0 100644
--- a/hw/usb-serial.c
+++ b/hw/usb-serial.c
@@ -545,7 +545,7 @@ static int usb_serial_initfn(USBDevice *dev)
     USBSerialState *s = DO_UPCAST(USBSerialState, dev, dev);
     s->dev.speed = USB_SPEED_FULL;
 
-    qemu_chr_add_handlers(s->cs, usb_serial_can_read, usb_serial_read,
+    qemu_chr_add_handlers(s->cs, usb_serial_can_read, usb_serial_read, NULL,
                           usb_serial_event, s);
     usb_serial_handle_reset(dev);
     return 0;
diff --git a/hw/virtio-console.c b/hw/virtio-console.c
index caea11f..2bccdd0 100644
--- a/hw/virtio-console.c
+++ b/hw/virtio-console.c
@@ -69,8 +69,8 @@ static int virtconsole_initfn(VirtIOSerialDevice *dev)
     port->is_console = true;
 
     if (vcon->chr) {
-        qemu_chr_add_handlers(vcon->chr, chr_can_read, chr_read, chr_event,
-                              vcon);
+        qemu_chr_add_handlers(vcon->chr, chr_can_read, chr_read,
+                              NULL, chr_event, vcon);
         port->info->have_data = flush_buf;
     }
     return 0;
@@ -118,8 +118,8 @@ static int virtserialport_initfn(VirtIOSerialDevice *dev)
     port->info = dev->info;
 
     if (vcon->chr) {
-        qemu_chr_add_handlers(vcon->chr, chr_can_read, chr_read, chr_event,
-                              vcon);
+        qemu_chr_add_handlers(vcon->chr, chr_can_read, chr_read,
+                              NULL, chr_event, vcon);
         port->info->have_data = flush_buf;
     }
     return 0;
diff --git a/hw/xen_console.c b/hw/xen_console.c
index d2261f4..387c7ed 100644
--- a/hw/xen_console.c
+++ b/hw/xen_console.c
@@ -224,7 +224,7 @@ static int con_connect(struct XenDevice *xendev)
     xen_be_bind_evtchn(&con->xendev);
     if (con->chr)
         qemu_chr_add_handlers(con->chr, xencons_can_receive, xencons_receive,
-                              NULL, con);
+                              NULL, NULL, con);
 
     xen_be_printf(xendev, 1, "ring mfn %d, remote port %d, local port %d, limit %zd\n",
 		  con->ring_ref,
@@ -238,8 +238,9 @@ static void con_disconnect(struct XenDevice *xendev)
 {
     struct XenConsole *con = container_of(xendev, struct XenConsole, xendev);
 
-    if (con->chr)
-        qemu_chr_add_handlers(con->chr, NULL, NULL, NULL, NULL);
+    if (con->chr) {
+        qemu_chr_add_handlers(con->chr, NULL, NULL, NULL, NULL, NULL);
+    }
     xen_be_unbind_evtchn(&con->xendev);
 
     if (con->sring) {
diff --git a/hw/xilinx_uartlite.c b/hw/xilinx_uartlite.c
index adab759..b811167 100644
--- a/hw/xilinx_uartlite.c
+++ b/hw/xilinx_uartlite.c
@@ -205,8 +205,9 @@ static int xilinx_uartlite_init(SysBusDevice *dev)
     sysbus_init_mmio(dev, R_MAX * 4, uart_regs);
 
     s->chr = qdev_init_chardev(&dev->qdev);
-    if (s->chr)
-        qemu_chr_add_handlers(s->chr, uart_can_rx, uart_rx, uart_event, s);
+    if (s->chr) {
+        qemu_chr_add_handlers(s->chr, uart_can_rx, uart_rx, NULL, uart_event, s);
+    }
     return 0;
 }
 
diff --git a/monitor.c b/monitor.c
index 9c05981..e43a130 100644
--- a/monitor.c
+++ b/monitor.c
@@ -4590,10 +4590,10 @@ void monitor_init(CharDriverState *chr, int flags)
     if (monitor_ctrl_mode(mon)) {
         mon->mc = qemu_mallocz(sizeof(MonitorControl));
         /* Control mode requires special handlers */
-        qemu_chr_add_handlers(chr, monitor_can_read, monitor_control_read,
+        qemu_chr_add_handlers(chr, monitor_can_read, monitor_control_read, NULL,
                               monitor_control_event, mon);
     } else {
-        qemu_chr_add_handlers(chr, monitor_can_read, monitor_read,
+        qemu_chr_add_handlers(chr, monitor_can_read, monitor_read, NULL,
                               monitor_event, mon);
     }
 
diff --git a/net/slirp.c b/net/slirp.c
index b41c60a..2a915ab 100644
--- a/net/slirp.c
+++ b/net/slirp.c
@@ -634,7 +634,7 @@ static int slirp_guestfwd(SlirpState *s, const char *config_str,
     fwd->slirp = s->slirp;
 
     qemu_chr_add_handlers(fwd->hd, guestfwd_can_read, guestfwd_read,
-                          NULL, fwd);
+                          NULL, NULL, fwd);
     return 0;
 
  fail_syntax:
diff --git a/net/socket.c b/net/socket.c
index 1c4e153..8a401e6 100644
--- a/net/socket.c
+++ b/net/socket.c
@@ -56,8 +56,8 @@ static ssize_t net_socket_receive(VLANClientState *nc, const uint8_t *buf, size_
     uint32_t len;
     len = htonl(size);
 
-    send_all(s->fd, (const uint8_t *)&len, sizeof(len));
-    return send_all(s->fd, buf, size);
+    send_all(s->fd, (const uint8_t *)&len, sizeof(len), false);
+    return send_all(s->fd, buf, size, false);
 }
 
 static ssize_t net_socket_receive_dgram(VLANClientState *nc, const uint8_t *buf, size_t size)
diff --git a/qemu-char.c b/qemu-char.c
index 6aaa362..4964d35 100644
--- a/qemu-char.c
+++ b/qemu-char.c
@@ -194,16 +194,21 @@ void qemu_chr_send_event(CharDriverState *s, int event)
 void qemu_chr_add_handlers(CharDriverState *s,
                            IOCanReadHandler *fd_can_read,
                            IOReadHandler *fd_read,
+                           IOHandler *fd_write_unblocked,
                            IOEventHandler *fd_event,
                            void *opaque)
 {
     s->chr_can_read = fd_can_read;
     s->chr_read = fd_read;
+    s->chr_write_unblocked = fd_write_unblocked;
     s->chr_event = fd_event;
     s->handler_opaque = opaque;
     if (s->chr_update_read_handler)
         s->chr_update_read_handler(s);
 
+    /* We'll set this at connect-time */
+    s->nonblock = false;
+
     /* We're connecting to an already opened device, so let's make sure we
        also get the open event */
     if (s->opened) {
@@ -456,7 +461,7 @@ static void mux_chr_update_read_handler(CharDriverState *chr)
     d->chr_event[d->mux_cnt] = chr->chr_event;
     /* Fix up the real driver with mux routines */
     if (d->mux_cnt == 0) {
-        qemu_chr_add_handlers(d->drv, mux_chr_can_read, mux_chr_read,
+        qemu_chr_add_handlers(d->drv, mux_chr_can_read, mux_chr_read, NULL,
                               mux_chr_event, chr);
     }
     if (d->focus != -1) {
@@ -490,7 +495,7 @@ static CharDriverState *qemu_chr_open_mux(CharDriverState *drv)
 
 
 #ifdef _WIN32
-int send_all(int fd, const void *buf, int len1)
+int send_all(int fd, const void *buf, int len1, bool nonblock)
 {
     int ret, len;
 
@@ -514,7 +519,7 @@ int send_all(int fd, const void *buf, int len1)
 
 #else
 
-static int unix_write(int fd, const uint8_t *buf, int len1)
+static int unix_write(int fd, const uint8_t *buf, int len1, bool nonblock)
 {
     int ret, len;
 
@@ -522,6 +527,9 @@ static int unix_write(int fd, const uint8_t *buf, int len1)
     while (len > 0) {
         ret = write(fd, buf, len);
         if (ret < 0) {
+            if (errno == EAGAIN && nonblock) {
+                return -EAGAIN;
+            }
             if (errno != EINTR && errno != EAGAIN) {
                 if (len1 - len) {
                     return len1 - len;
@@ -539,9 +547,9 @@ static int unix_write(int fd, const uint8_t *buf, int len1)
     return len1 - len;
 }
 
-int send_all(int fd, const void *buf, int len1)
+int send_all(int fd, const void *buf, int len1, bool nonblock)
 {
-    return unix_write(fd, buf, len1);
+    return unix_write(fd, buf, len1, nonblock);
 }
 #endif /* !_WIN32 */
 
@@ -558,7 +566,7 @@ static int stdio_nb_clients = 0;
 static int fd_chr_write(CharDriverState *chr, const uint8_t *buf, int len)
 {
     FDCharDriver *s = chr->opaque;
-    return send_all(s->fd_out, buf, len);
+    return send_all(s->fd_out, buf, len, chr->nonblock);
 }
 
 static int fd_chr_read_poll(void *opaque)
@@ -870,7 +878,7 @@ static int pty_chr_write(CharDriverState *chr, const uint8_t *buf, int len)
         pty_chr_update_read_handler(chr);
         return 0;
     }
-    return send_all(s->fd, buf, len);
+    return send_all(s->fd, buf, len, chr->nonblock);
 }
 
 static int pty_chr_read_poll(void *opaque)
@@ -1934,8 +1942,14 @@ static void tcp_chr_accept(void *opaque);
 static int tcp_chr_write(CharDriverState *chr, const uint8_t *buf, int len)
 {
     TCPCharDriver *s = chr->opaque;
+    int ret;
+
     if (s->connected) {
-        return send_all(s->fd, buf, len);
+        ret = send_all(s->fd, buf, len, chr->nonblock);
+        if (ret == -EAGAIN) {
+            chr->write_blocked = true;
+        }
+        return ret;
     } else {
         /* XXX: indicate an error ? */
         return len;
@@ -2101,14 +2115,41 @@ static void tcp_chr_read(void *opaque)
     }
 }
 
+static void tcp_chr_write_unblocked(void *opaque)
+{
+    CharDriverState *chr = opaque;
+
+    if (chr->write_blocked && chr->chr_write_unblocked) {
+        chr->write_blocked = false;
+        chr->chr_write_unblocked(chr->handler_opaque);
+    }
+}
+
 static void tcp_chr_connect(void *opaque)
 {
     CharDriverState *chr = opaque;
     TCPCharDriver *s = chr->opaque;
+    IOHandler *write_cb;
+    int flags;
+    bool nonblock;
+
+    flags = fcntl(s->fd, F_GETFL);
+    if (flags == -1) {
+        flags = 0;
+    }
+    nonblock = flags & O_NONBLOCK;
+
+    write_cb = NULL;
+    chr->nonblock = false;
+    if (nonblock && chr->chr_write_unblocked) {
+        write_cb = chr->chr_write_unblocked;
+        chr->nonblock = true;
+    }
 
+    chr->write_blocked = false;
     s->connected = 1;
     qemu_set_fd_handler2(s->fd, tcp_chr_read_poll,
-                         tcp_chr_read, NULL, chr);
+                         tcp_chr_read, tcp_chr_write_unblocked, chr);
     qemu_chr_generic_open(chr);
 }
 
diff --git a/qemu-char.h b/qemu-char.h
index e3a0783..d65c31a 100644
--- a/qemu-char.h
+++ b/qemu-char.h
@@ -1,6 +1,8 @@
 #ifndef QEMU_CHAR_H
 #define QEMU_CHAR_H
 
+#include <stdbool.h>
+
 #include "qemu-common.h"
 #include "qemu-queue.h"
 #include "qemu-option.h"
@@ -59,6 +61,7 @@ struct CharDriverState {
     IOEventHandler *chr_event;
     IOCanReadHandler *chr_can_read;
     IOReadHandler *chr_read;
+    IOHandler *chr_write_unblocked;
     void *handler_opaque;
     void (*chr_send_event)(struct CharDriverState *chr, int event);
     void (*chr_close)(struct CharDriverState *chr);
@@ -68,6 +71,8 @@ struct CharDriverState {
     char *label;
     char *filename;
     int opened;
+    bool nonblock;
+    bool write_blocked;
     QTAILQ_ENTRY(CharDriverState) next;
 };
 
@@ -82,6 +87,7 @@ void qemu_chr_send_event(CharDriverState *s, int event);
 void qemu_chr_add_handlers(CharDriverState *s,
                            IOCanReadHandler *fd_can_read,
                            IOReadHandler *fd_read,
+                           IOHandler *fd_write_unblocked,
                            IOEventHandler *fd_event,
                            void *opaque);
 int qemu_chr_ioctl(CharDriverState *s, int cmd, void *arg);
diff --git a/qemu_socket.h b/qemu_socket.h
index 164ae3e..bdf878b 100644
--- a/qemu_socket.h
+++ b/qemu_socket.h
@@ -23,6 +23,7 @@ int inet_aton(const char *cp, struct in_addr *ia);
 #include <arpa/inet.h>
 #include <netdb.h>
 #include <sys/un.h>
+#include <stdbool.h>
 
 #define socket_error() errno
 #define closesocket(s) close(s)
@@ -35,7 +36,7 @@ int inet_aton(const char *cp, struct in_addr *ia);
 int qemu_socket(int domain, int type, int protocol);
 int qemu_accept(int s, struct sockaddr *addr, socklen_t *addrlen);
 void socket_set_nonblock(int fd);
-int send_all(int fd, const void *buf, int len1);
+int send_all(int fd, const void *buf, int len1, bool nonblock);
 
 /* New, ipv6-ready socket helper functions, see qemu-sockets.c */
 int inet_listen_opts(QemuOpts *opts, int port_offset);
-- 
1.6.2.5

[Qemu-devel] [PATCH v3 3/4] virtio-console: Factor out common init between console and generic ports

[Amit Shah]

The initialisation for generic ports and console ports is similar.
Factor out the parts that are the same in a different function that can
be called from each of the initfns.

Signed-off-by: Amit Shah <amit.shah@redhat.com>
---
 hw/virtio-console.c |   31 ++++++++++++++-----------------
 1 files changed, 14 insertions(+), 17 deletions(-)

diff --git a/hw/virtio-console.c b/hw/virtio-console.c
index 2bccdd0..1552f47 100644
--- a/hw/virtio-console.c
+++ b/hw/virtio-console.c
@@ -58,24 +58,28 @@ static void chr_event(void *opaque, int event)
     }
 }
 
-/* Virtio Console Ports */
-static int virtconsole_initfn(VirtIOSerialDevice *dev)
+static int generic_port_init(VirtConsole *vcon, VirtIOSerialDevice *dev)
 {
-    VirtIOSerialPort *port = DO_UPCAST(VirtIOSerialPort, dev, &dev->qdev);
-    VirtConsole *vcon = DO_UPCAST(VirtConsole, port, port);
-
-    port->info = dev->info;
-
-    port->is_console = true;
+    vcon->port.info = dev->info;
 
     if (vcon->chr) {
         qemu_chr_add_handlers(vcon->chr, chr_can_read, chr_read,
                               NULL, chr_event, vcon);
-        port->info->have_data = flush_buf;
+        vcon->port.info->have_data = flush_buf;
     }
     return 0;
 }
 
+/* Virtio Console Ports */
+static int virtconsole_initfn(VirtIOSerialDevice *dev)
+{
+    VirtIOSerialPort *port = DO_UPCAST(VirtIOSerialPort, dev, &dev->qdev);
+    VirtConsole *vcon = DO_UPCAST(VirtConsole, port, port);
+
+    port->is_console = true;
+    return generic_port_init(vcon, dev);
+}
+
 static int virtconsole_exitfn(VirtIOSerialDevice *dev)
 {
     VirtIOSerialPort *port = DO_UPCAST(VirtIOSerialPort, dev, &dev->qdev);
@@ -115,14 +119,7 @@ static int virtserialport_initfn(VirtIOSerialDevice *dev)
     VirtIOSerialPort *port = DO_UPCAST(VirtIOSerialPort, dev, &dev->qdev);
     VirtConsole *vcon = DO_UPCAST(VirtConsole, port, port);
 
-    port->info = dev->info;
-
-    if (vcon->chr) {
-        qemu_chr_add_handlers(vcon->chr, chr_can_read, chr_read,
-                              NULL, chr_event, vcon);
-        port->info->have_data = flush_buf;
-    }
-    return 0;
+    return generic_port_init(vcon, dev);
 }
 
 static VirtIOSerialPortInfo virtserialport_info = {
-- 
1.6.2.5

[Qemu-devel] [PATCH v3 4/4] 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.

Allow the chardev backends to return -EAGAIN; we're ready with a
callback handler that will flush the remainder of the buffer.

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

Signed-off-by: Amit Shah <amit.shah@redhat.com>
---
 hw/virtio-console.c |  127 +++++++++++++++++++++++++++++++++++++++++++++++++--
 1 files changed, 123 insertions(+), 4 deletions(-)

diff --git a/hw/virtio-console.c b/hw/virtio-console.c
index 1552f47..1f8e695 100644
--- a/hw/virtio-console.c
+++ b/hw/virtio-console.c
@@ -13,18 +13,92 @@
 #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;
+    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)
+{
+    if (vcon->unflushed_buf) {
+        qemu_free(vcon->unflushed_buf->buf);
+        qemu_free(vcon->unflushed_buf);
+        vcon->unflushed_buf = NULL;
+    }
+}
+
+static int buffered_write_to_chardev(VirtConsole *vcon, const uint8_t *buf,
+                                     size_t len)
+{
+    size_t written;
+    ssize_t ret;
+
+    written = 0;
+    do {
+        ret = qemu_chr_write(vcon->chr, buf + written, len - written);
+        if (ret < 0) {
+            if (vcon->unflushed_buf) {
+                vcon->unflushed_buf->offset += written;
+                vcon->unflushed_buf->rem_len -= written;
+            } else {
+                virtio_serial_throttle_port(&vcon->port, true);
+                add_unflushed_buf(vcon, buf + written, len - written);
+            }
+
+            return -EAGAIN;
+        }
+
+        written += ret;
+    } while (written != len);
+
+    return 0;
+}
+
+/* Callback function called when the chardev can accept more data */
+static void chr_write_unblocked(void *opaque)
+{
+    VirtConsole *vcon = opaque;
+
+    if (vcon->unflushed_buf) {
+        int ret;
+
+        ret = buffered_write_to_chardev(vcon, vcon->unflushed_buf->buf
+                                              + vcon->unflushed_buf->offset,
+                                        vcon->unflushed_buf->rem_len);
+        if (ret < 0) {
+            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);
+    /* If a previous write was incomplete, we should've been throttled. */
+    assert(!vcon->unflushed_buf);
+
+    buffered_write_to_chardev(vcon, buf, len);
 }
 
 /* Readiness of the guest to accept data on a port */
@@ -48,25 +122,69 @@ 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) {
+            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;
 
     if (vcon->chr) {
         qemu_chr_add_handlers(vcon->chr, chr_can_read, chr_read,
-                              NULL, chr_event, vcon);
+                              chr_write_unblocked, chr_event, vcon);
         vcon->port.info->have_data = flush_buf;
     }
+    register_savevm("virtio-console-ports", -1, 1, virtio_console_port_save,
+		    virtio_console_port_load, vcon);
     return 0;
 }
 
@@ -88,6 +206,7 @@ static int virtconsole_exitfn(VirtIOSerialDevice *dev)
     if (vcon->chr) {
         port->info->have_data = NULL;
         qemu_chr_close(vcon->chr);
+        free_unflushed_buf(vcon);
     }
 
     return 0;
-- 
1.6.2.5

Re: [Qemu-devel] [PATCH v3 0/4] char: write callback, virtio-console: flow control

[Paul Brook]

> This series lets interested callers ask for an -EAGAIN return from the
> chardev backends (only unix and tcp sockets as of now) to implement
> their own flow control.

As mentioned previously, I think this is a bad idea.  The device has no useful 
way of determining when to transmit the rest of the data.

What you really want is an asynchronous transmit API.

Paul

Re: [Qemu-devel] [PATCH v3 0/4] char: write callback, virtio-console: flow control

[Amit Shah]

On (Thu) Apr 15 2010 [13:04:40], Paul Brook wrote:
> > This series lets interested callers ask for an -EAGAIN return from the
> > chardev backends (only unix and tcp sockets as of now) to implement
> > their own flow control.
> 
> As mentioned previously, I think this is a bad idea.  The device has no useful 
> way of determining when to transmit the rest of the data.

The device gets a callback which indicates the chardev is ready to
accept more data.

		Amit

Re: [Qemu-devel] [PATCH v3 2/4] char: Add ability to provide a callback when write won't return -EAGAIN

[Gerd Hoffmann]

>   static void tcp_chr_connect(void *opaque)
>   {

> +    chr->write_blocked = false;
>       s->connected = 1;
>       qemu_set_fd_handler2(s->fd, tcp_chr_read_poll,
> -                         tcp_chr_read, NULL, chr);
> +                         tcp_chr_read, tcp_chr_write_unblocked, chr);

This is wrong.  You want register tcp_chr_write_unblocked only for the 
chr->write_blocked == true (i.e. output buffers are full) case. 
Otherwise qemu will burn cpu calling tcp_chr_write_unblocked.

Yes, you'll have to call qemu_set_fd_handler2 each time write_blocked 
changes state.

Also implementing the whole logic at the individual chardev drivers 
level feels somewhat wrong as it will identical for most (all?) unix 
chardev drivers.

cheers,
   Gerd

Re: [Qemu-devel] [PATCH v3 2/4] char: Add ability to provide a callback when write won't return -EAGAIN

[Amit Shah]

On (Tue) Apr 20 2010 [13:32:18], Gerd Hoffmann wrote:
>>   static void tcp_chr_connect(void *opaque)
>>   {
>
>> +    chr->write_blocked = false;
>>       s->connected = 1;
>>       qemu_set_fd_handler2(s->fd, tcp_chr_read_poll,
>> -                         tcp_chr_read, NULL, chr);
>> +                         tcp_chr_read, tcp_chr_write_unblocked, chr);
>
> This is wrong.  You want register tcp_chr_write_unblocked only for the  
> chr->write_blocked == true (i.e. output buffers are full) case.  
> Otherwise qemu will burn cpu calling tcp_chr_write_unblocked.
>
> Yes, you'll have to call qemu_set_fd_handler2 each time write_blocked  
> changes state.

Agreed. That's coming in the next round when I make it generic enough
for all the backends to call the common code.

> Also implementing the whole logic at the individual chardev drivers  
> level feels somewhat wrong as it will identical for most (all?) unix  
> chardev drivers.

Yes, both these come together, I guess.

I sent out this series as a "feeler" to see if the approach was
acceptable.

Paul didn't reply to my reply addressing his concern, so I take that as
he's OK with the approach as well :-)

		Amit

Re: [Qemu-devel] [PATCH v3 2/4] char: Add ability to provide a callback when write won't return -EAGAIN

[Paul Brook]

> I sent out this series as a "feeler" to see if the approach was
> acceptable.
> 
> Paul didn't reply to my reply addressing his concern, so I take that as
> he's OK with the approach as well :-)

I'd probably exposed this as an asyncronous write rather than nonblocking 
operation. However both have their issues and I guess for character devices 
your approach makes sense (c.f. block devices where we want concurrent 
transfers).

It would be useful to have a debugging mode where the chardev layer 
deliberately returns spurious EAGAIN and short writes. Otherwise you've got a 
lot of very poorly tested device fallback code. I have low confidence in 
getting this right first time :-)

Paul

Re: [Qemu-devel] [PATCH v3 2/4] char: Add ability to provide a callback when write won't return -EAGAIN

[Amit Shah]

On (Tue) Apr 20 2010 [13:28:48], Paul Brook wrote:
> 
> It would be useful to have a debugging mode where the chardev layer 
> deliberately returns spurious EAGAIN and short writes. Otherwise you've got a 
> lot of very poorly tested device fallback code. I have low confidence in 
> getting this right first time :-)

Ah, hm, I'll see how I can do that.

BTW I test this a couple of ways:

1. make the guest write till the host blocks because the chardev can't
accept any more data. I ascertain this during development using some
fprintfs. Any more data that gets written fills the guest's vq and then
the write() in the guest starts blocking. Once this happens, my testcase
succeeds.

2. Send a huge file from the guest to the host and compare the sha1sums
at the end of the file transfer. This always results in the chardev
returning -eagain a few times (confirm during testing via fprintfs
again). The testcase passes if the sha1sums match.

The test suite is at

http://fedorapeople.org/gitweb?p=amitshah/public_git/test-virtserial.git

		Amit

Re: [Qemu-devel] [PATCH v3 2/4] char: Add ability to provide a callback when write won't return -EAGAIN

[Gerd Hoffmann]

On 04/20/10 14:28, Paul Brook wrote:
>> I sent out this series as a "feeler" to see if the approach was
>> acceptable.
>>
>> Paul didn't reply to my reply addressing his concern, so I take that as
>> he's OK with the approach as well :-)
>
> I'd probably exposed this as an asyncronous write rather than nonblocking
> operation. However both have their issues and I guess for character devices
> your approach makes sense (c.f. block devices where we want concurrent
> transfers).

For chardevs async operation introduces ordering issues, I think 
supporting non-blocking writes is more useful here.

> It would be useful to have a debugging mode where the chardev layer
> deliberately returns spurious EAGAIN and short writes. Otherwise you've got a
> lot of very poorly tested device fallback code. I have low confidence in
> getting this right first time :-)

It might be a good idea to have a different interface for it, i.e. 
qemu_chr_write() keeps current behavior and we'll add a new 
qemu_chr_write_nonblocking() for users which want (and can handle) the 
non-blocking behavior with short writes and -EAGAIN return values.

We should also clearly define what qemu_chr_write_nonblocking() should 
do in case the underlying chardev backend doesn't support nonblocking 
operation.  Option one is to fail and expect the caller handle the 
situation.  Option two is fallback to blocking mode.  I'd tend to pick 
option two, fallback to blocking mode is what the caller most likely 
will do anyway.

cheers,
   Gerd