[Qemu-devel] [PATCH v4 0/5] char: non-blocking writes, virtio-console flow control

[Qemu-devel] [PATCH v4 0/5] char: non-blocking writes, 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.

A new call, qemu_chr_write_nb() is added, that will fallback to
qemu_chr_write() if the backend file isn't non-blocking or if no
callback was supplied.

Support for other backend types is easy to add and will be done in
later patches.

Amit Shah (5):
  char: Let writers know how much data was written in case of errors
  char: Add qemu_chr_write_nb() for nonblocking writes
  char: unix/tcp: Add a non-blocking write handler
  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          |   94 +++++++++++++++++++++++++++---
 qemu-char.h          |    9 +++
 qemu_socket.h        |    3 +-
 20 files changed, 257 insertions(+), 53 deletions(-)

[Qemu-devel] [PATCH v4 1/5] 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 ac65a1c..bc76000 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 v4 2/5] char: Add qemu_chr_write_nb() for nonblocking writes

[Amit Shah]

For char devices whose backing files are open in non-blocking mode,
non-blocking writes can now be made using qemu_chr_write_nb().

For non-blocking chardevs, we can return -EAGAIN to callers of
qemu_chr_write_nb(). When the backend is ready to accept more data,
we can let the caller know via a callback.

-EAGAIN is returned only when the backend's file is non-blocking
and a callback is registered by the caller when invoking
qemu_chr_add_handlers().

In case a backend doesn't support non-blocking writes,
qemu_chr_write_nb() fallsback to qemu_chr_write().

Update all callers of qemu chardevs to not register any callback
handler for non-blocked writes.

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

No backend currently supports non-blocking writes.

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          |   35 +++++++++++++++++++++++++++--------
 qemu-char.h          |    9 +++++++++
 qemu_socket.h        |    3 ++-
 20 files changed, 68 insertions(+), 35 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 5b2483a..d522a4d 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 754bcc5..d5c5fb0 100644
--- a/monitor.c
+++ b/monitor.c
@@ -4617,10 +4617,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 bc76000..1cae1d2 100644
--- a/qemu-char.c
+++ b/qemu-char.c
@@ -145,6 +145,16 @@ int qemu_chr_write(CharDriverState *s, const uint8_t *buf, int len)
     return s->chr_write(s, buf, len);
 }
 
+ssize_t qemu_chr_write_nb(CharDriverState *s, const uint8_t *buf, size_t len)
+{
+    if (!s->nonblock) {
+        /* Fallback to blocking write if no callback registered */
+        return qemu_chr_write(s, buf, len);
+    }
+
+    return s->chr_write_nb(s, buf, len);
+}
+
 int qemu_chr_ioctl(CharDriverState *s, int cmd, void *arg)
 {
     if (!s->chr_ioctl)
@@ -194,16 +204,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 +471,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 +505,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 +529,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 +537,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 +557,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 +576,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, false);
 }
 
 static int fd_chr_read_poll(void *opaque)
@@ -870,7 +888,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, false);
 }
 
 static int pty_chr_read_poll(void *opaque)
@@ -1934,8 +1952,9 @@ static void tcp_chr_accept(void *opaque);
 static int tcp_chr_write(CharDriverState *chr, const uint8_t *buf, int len)
 {
     TCPCharDriver *s = chr->opaque;
+
     if (s->connected) {
-        return send_all(s->fd, buf, len);
+        return send_all(s->fd, buf, len, false);
     } else {
         /* XXX: indicate an error ? */
         return len;
diff --git a/qemu-char.h b/qemu-char.h
index e3a0783..f0168f6 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"
@@ -53,12 +55,15 @@ typedef void IOEventHandler(void *opaque, int event);
 struct CharDriverState {
     void (*init)(struct CharDriverState *s);
     int (*chr_write)(struct CharDriverState *s, const uint8_t *buf, int len);
+    ssize_t (*chr_write_nb)(struct CharDriverState *s, const uint8_t *buf,
+                            size_t len);
     void (*chr_update_read_handler)(struct CharDriverState *s);
     int (*chr_ioctl)(struct CharDriverState *s, int cmd, void *arg);
     int (*get_msgfd)(struct CharDriverState *s);
     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 +73,8 @@ struct CharDriverState {
     char *label;
     char *filename;
     int opened;
+    bool nonblock;
+    bool write_blocked;
     QTAILQ_ENTRY(CharDriverState) next;
 };
 
@@ -78,10 +85,12 @@ CharDriverState *qemu_chr_open(const char *label, const char *filename, void (*i
 void qemu_chr_close(CharDriverState *chr);
 void qemu_chr_printf(CharDriverState *s, const char *fmt, ...);
 int qemu_chr_write(CharDriverState *s, const uint8_t *buf, int len);
+ssize_t qemu_chr_write_nb(CharDriverState *s, const uint8_t *buf, size_t len);
 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 v4 3/5] char: unix/tcp: Add a non-blocking write handler

[Amit Shah]

Add a non-blocking write handler that can return with -EAGAIN to the
caller and also callback when the socket becomes writable.

Non-blocking writes are only enabled for sockets that are opened in
non-blocking mode and only for callers that have registered a callback
handler for resuming writes.

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

diff --git a/qemu-char.c b/qemu-char.c
index 1cae1d2..476bb4e 100644
--- a/qemu-char.c
+++ b/qemu-char.c
@@ -2117,11 +2117,60 @@ static void tcp_chr_read(void *opaque)
     }
 }
 
+static void tcp_chr_write_unblocked(void *opaque)
+{
+    CharDriverState *chr = opaque;
+    TCPCharDriver *s = chr->opaque;
+
+    assert(chr->write_blocked && chr->chr_write_unblocked);
+
+    chr->write_blocked = false;
+    qemu_set_fd_handler2(s->fd, tcp_chr_read_poll, tcp_chr_read, NULL, chr);
+    chr->chr_write_unblocked(chr->handler_opaque);
+}
+
+static ssize_t tcp_chr_write_nb(CharDriverState *chr, const uint8_t *buf,
+                                size_t len)
+{
+    TCPCharDriver *s = chr->opaque;
+    ssize_t ret;
+
+    if (!s->connected) {
+        /* XXX: indicate an error? */
+        return len;
+    }
+
+    ret = send_all(s->fd, buf, len, true);
+    if (ret == -EAGAIN) {
+        chr->write_blocked = true;
+        qemu_set_fd_handler2(s->fd, tcp_chr_read_poll,
+                             tcp_chr_read, tcp_chr_write_unblocked, chr);
+    }
+    return ret;
+}
+
 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);
@@ -2254,6 +2303,7 @@ static CharDriverState *qemu_chr_open_socket(QemuOpts *opts)
 
     chr->opaque = s;
     chr->chr_write = tcp_chr_write;
+    chr->chr_write_nb = tcp_chr_write_nb;
     chr->chr_close = tcp_chr_close;
     chr->get_msgfd = tcp_get_msgfd;
 
-- 
1.6.2.5

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

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..035ef94 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_nb(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

[Qemu-devel] Re: [PATCH v4 2/5] char: Add qemu_chr_write_nb() for nonblocking writes

[Gerd Hoffmann]

   Hi,

> -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 +537,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;
> +            }

You've just re-introduced the bug you've fixed in patch 1/5 ...

cheers,
   Gerd

[Qemu-devel] Re: [PATCH v4 3/5] char: unix/tcp: Add a non-blocking write handler

[Gerd Hoffmann]

On 05/04/10 09:17, Amit Shah wrote:
> Add a non-blocking write handler that can return with -EAGAIN to the
> caller and also callback when the socket becomes writable.
>
> Non-blocking writes are only enabled for sockets that are opened in
> non-blocking mode and only for callers that have registered a callback
> handler for resuming writes.

The logic will be pretty much the same for all filehandle-based backends 
on unix.  So maybe create some helper functions, so implementing 
$backend_chr_write_unblocked can be done with just three lines of code?

cheers,
   Gerd

[Qemu-devel] Re: [PATCH v4 3/5] char: unix/tcp: Add a non-blocking write handler

[Amit Shah]

On (Tue) May 04 2010 [09:43:04], Gerd Hoffmann wrote:
> On 05/04/10 09:17, Amit Shah wrote:
>> Add a non-blocking write handler that can return with -EAGAIN to the
>> caller and also callback when the socket becomes writable.
>>
>> Non-blocking writes are only enabled for sockets that are opened in
>> non-blocking mode and only for callers that have registered a callback
>> handler for resuming writes.
>
> The logic will be pretty much the same for all filehandle-based backends  
> on unix.  So maybe create some helper functions, so implementing  
> $backend_chr_write_unblocked can be done with just three lines of code?

Yes, that's the next step. Let's push this for now and do that in a
later patch series?

		Amit

[Qemu-devel] Re: [PATCH v4 2/5] char: Add qemu_chr_write_nb() for nonblocking writes

[Amit Shah]

On (Tue) May 04 2010 [09:38:11], Gerd Hoffmann wrote:
>   Hi,
>
>> -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 +537,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;
>> +            }
>
> You've just re-introduced the bug you've fixed in patch 1/5 ...

Uh!

		Amit

[Qemu-devel] Re: [PATCH v4 3/5] char: unix/tcp: Add a non-blocking write handler

[Gerd Hoffmann]

On 05/04/10 09:56, Amit Shah wrote:
>> The logic will be pretty much the same for all filehandle-based backends
>> on unix.  So maybe create some helper functions, so implementing
>> $backend_chr_write_unblocked can be done with just three lines of code?
>
> Yes, that's the next step. Let's push this for now and do that in a
> later patch series?

Fine with me.

cheers,
   Gerd