[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 <address@hidden>
---
 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