Re: [Qemu-devel] v6: [PATCH 0/3]: Threadlets: A generic task offloading framework

[Amit Shah]

On (Tue) Oct 19 2010 [23:12:20], Arun R Bharadwaj wrote:
> Hi,
> 
> This is the v6 of the patch-series to have a generic asynchronous task
> offloading framework (called threadlets) within qemu.
> 
> Request to consider pulling this series as discussed during the
> Qemu-devel call.

I tried this out with virtio-serial (patch below).  Have a couple of
things to note:

- Guests get a SIGUSR2 on startup sometimes.  This doesn't happen with
  qemu.git, so looks like it's introduced by this patchset.

- After running some tests, I get an abort.  I still have to look at
  what's causing it, but doesn't look like it's related to virtio-serial
  code.

Program received signal SIGABRT, Aborted.
0x0000003dc76329a5 in raise () from /lib64/libc.so.6
Missing separate debuginfos, use: debuginfo-install
SDL-1.2.14-8.fc13.x86_64 glibc-2.12.1-2.x86_64
libX11-1.3.1-3.fc13.x86_64 libXau-1.0.5-1.fc12.x86_64
libpng-1.2.44-1.fc13.x86_64 libxcb-1.5-1.fc13.x86_64
ncurses-libs-5.7-7.20100130.fc13.x86_64 zlib-1.2.3-23.fc12.x86_64
(gdb) bt
#0  0x0000003dc76329a5 in raise () from /lib64/libc.so.6
#1  0x0000003dc7634185 in abort () from /lib64/libc.so.6
#2  0x00000000004bf829 in qemu_get_ram_ptr (addr=<value optimized out>)
at /home/amit/src/qemu/exec.c:2936
#3  0x00000000004bf9a7 in lduw_phys (addr=<value optimized out>) at
/home/amit/src/qemu/exec.c:3836
#4  0x0000000000557c90 in vring_avail_idx (vq=0x17b9320, idx=1333) at
/home/amit/src/qemu/hw/virtio.c:133
#5  virtqueue_num_heads (vq=0x17b9320, idx=1333) at
/home/amit/src/qemu/hw/virtio.c:252
#6  0x0000000000557e5e in virtqueue_avail_bytes (vq=0x17b9320,
in_bytes=4096, out_bytes=0) at /home/amit/src/qemu/hw/virtio.c:311

- I'm using a threadlet to queue up several work items which are to be
  processed in a fifo order.  There's no cancel function for a threadlet
  that either processes all work and then quits the thread or just
  cancels all pending work and quits.

                Amit


diff --git a/hw/virtio-serial-bus.c b/hw/virtio-serial-bus.c
index 74ba5ec..caaafbe 100644
--- a/hw/virtio-serial-bus.c
+++ b/hw/virtio-serial-bus.c
@@ -51,6 +51,14 @@ struct VirtIOSerial {
     struct virtio_console_config config;
 };
 
+typedef struct VirtIOSerialWork {
+    ThreadletWork work;
+    VirtIOSerialPort *port;
+    VirtQueue *vq;
+    VirtIODevice *vdev;
+    int discard;
+} VirtIOSerialWork;
+
 static VirtIOSerialPort *find_port_by_id(VirtIOSerial *vser, uint32_t id)
 {
     VirtIOSerialPort *port;
@@ -113,10 +121,20 @@ static size_t write_to_port(VirtIOSerialPort *port,
     return offset;
 }
 
-static void do_flush_queued_data(VirtIOSerialPort *port, VirtQueue *vq,
-                                 VirtIODevice *vdev, bool discard)
+static void async_flush_queued_data(ThreadletWork *work)
 {
+    VirtIOSerialPort *port;
+    VirtIOSerialWork *vs_work;
+    VirtQueue *vq;
+    VirtIODevice *vdev;
     VirtQueueElement elem;
+    int discard;
+
+    vs_work = DO_UPCAST(VirtIOSerialWork, work, work);
+    port = vs_work->port;
+    vq = vs_work->vq;
+    vdev = vs_work->vdev;
+    discard = vs_work->discard;
 
     assert(port || discard);
     assert(virtio_queue_ready(vq));
@@ -136,6 +154,24 @@ static void do_flush_queued_data(VirtIOSerialPort *port, 
VirtQueue *vq,
         virtqueue_push(vq, &elem, 0);
     }
     virtio_notify(vdev, vq);
+
+    qemu_free(vs_work);
+}
+
+static void do_flush_queued_data(VirtIOSerialPort *port, VirtQueue *vq,
+                                 VirtIODevice *vdev, bool discard)
+{
+    VirtIOSerialWork *vs_work;
+
+    /* TODO: can just do the needful if discard is true */
+
+    vs_work = qemu_malloc(sizeof(*vs_work));
+    vs_work->work.func = async_flush_queued_data;
+    vs_work->discard = discard;
+    vs_work->vdev = vdev;
+    vs_work->vq = vq;
+    vs_work->port = port;
+    submit_threadletwork_to_queue(&port->tqueue, &vs_work->work);
 }
 
 static void flush_queued_data(VirtIOSerialPort *port, bool discard)
@@ -699,6 +735,12 @@ static int virtser_port_qdev_init(DeviceState *qdev, 
DeviceInfo *base)
     port->ivq = port->vser->ivqs[port->id];
     port->ovq = port->vser->ovqs[port->id];
 
+    /*
+     * Just one thread to process all the work -- we don't want guest
+     * buffers to be processed out-of-order.
+     */
+    threadlet_queue_init(&port->tqueue, 1, 1);
+
     add_port(port->vser, port->id);
 
     /* Send an update to the guest about this new port added */
@@ -717,6 +759,8 @@ static int virtser_port_qdev_exit(DeviceState *qdev)
 
     QTAILQ_REMOVE(&vser->ports, port, next);
 
+    /* TODO: Cancel threadlet */
+
     if (port->info->exit)
         port->info->exit(dev);
 
diff --git a/hw/virtio-serial.h b/hw/virtio-serial.h
index ff08c40..15e0982 100644
--- a/hw/virtio-serial.h
+++ b/hw/virtio-serial.h
@@ -15,6 +15,7 @@
 #ifndef _QEMU_VIRTIO_SERIAL_H
 #define _QEMU_VIRTIO_SERIAL_H
 
+#include "qemu-threadlets.h"
 #include "qdev.h"
 #include "virtio.h"
 
@@ -88,6 +89,13 @@ struct VirtIOSerialPort {
     VirtQueue *ivq, *ovq;
 
     /*
+     * Threadlet queue for processing work for this port -- ensures we
+     * don't block the guest that writes out data and also other
+     * ports.
+     */
+    ThreadletQueue tqueue;
+
+    /*
      * This name is sent to the guest and exported via sysfs.
      * The guest could create symlinks based on this information.
      * The name is in the reverse fqdn format, like org.qemu.console.0