[Qemu-devel] [PATCH 04/12] usb: Add support for input pipelining

[Hans de Goede]

Currently we effectively only do pipelining for output endpoints, since
controllers will stop filling the queue for a packet with stop the queue
on a short read semantics enabled.

This causes large input transfers to get split into multiple packets, which
comes with a serious performance penalty.

This patch makes it possible to do pipelining for input endpoints, by
re-combining packets which belong to 1 transfer at the guest device-driver
level into 1 large packet before sending them to the device, this is mostly
useful as a significant performance improvement for redirection of real USB
devices.

This patch will combine packets together into 1 until a transfer terminating
packet is encountered. A terminating packet is a packet which meets one of
the following conditions:
1) The packet size is *not* a multiple of the endpoint max packet size
2) The packet does *not* have its short-not-ok flag set
3) The packet has its interrupt-on-complete flag set

The short-not-ok flag of the combined packet is that of the terminating packet.
Multiple combined packets may be submitted to the device, if the combined
packets do not have their short-not-ok flag set, enabling true pipelining.

If a combined packet does have its short-not-ok flag set the queue will
wait with submitting further packets to the device until that packet has
completed.

Once enabled in the ehci driver, this improves the speed (MB/s) of a Linux
guest reading from a USB mass storage device by a factor of 1.2 - 1.5.

And the main reason why I started working on this, when reading from a pl2303
USB<->serial converter, it combines the previous 4 packets submitted per
device-driver level read into 1 big read, reducing the number of packets / sec
by a factor 4, and it allows to have multiple reads outstanding. This allows
for much better latency tolerance without the pl2303's internal buffer
overflowing (which was happening at 115200 bps, without serial flow control).

Signed-off-by: Hans de Goede <address@hidden>
---
 hw/usb.h         |  24 ++++++++
 hw/usb/core.c    | 176 +++++++++++++++++++++++++++++++++++++++++++++++++++++--
 hw/usb/dev-hub.c |   9 +++
 3 files changed, 205 insertions(+), 4 deletions(-)

diff --git a/hw/usb.h b/hw/usb.h
index 2ecd8e9..57d200b 100644
--- a/hw/usb.h
+++ b/hw/usb.h
@@ -312,6 +312,15 @@ typedef struct USBPortOps {
      * the packet originated when a hub is involved.
      */
     void (*complete)(USBPort *port, USBPacket *p);
+    /*
+     * Called by the usb-core when a previous queued up packet
+     * (usb_handle_packet returned USB_RET_ASYNC) gets removed from the
+     * queue, without completing or being cancelled.
+     * This happens with input pipelinging when multiple packets are combined
+     * into one large transfer, and that transfer ends with a short read.
+     * Mandatory for controllers which set supports_queuing on their ports.
+     */
+    void (*remove_from_queue)(USBPort *port, USBPacket *p);
 } USBPortOps;
 
 /* USB port on which a device can be connected */
@@ -322,6 +331,7 @@ struct USBPort {
     USBPortOps *ops;
     void *opaque;
     int index; /* internal port index, may be used with the opaque */
+    bool supports_queuing;
     QTAILQ_ENTRY(USBPort) next;
 };
 
@@ -347,7 +357,12 @@ struct USBPacket {
     int result; /* transfer length or USB_RET_* status code */
     /* Internal use by the USB layer.  */
     USBPacketState state;
+    USBPacket *combined_in;
+    QTAILQ_HEAD(, USBPacket) uncombined_packets;
     QTAILQ_ENTRY(USBPacket) queue;
+    /* Flags set by the hcd when queuing is supported */
+    bool short_not_ok;
+    bool int_req;
 };
 
 void usb_packet_init(USBPacket *p);
@@ -387,6 +402,15 @@ int usb_ep_get_max_packet_size(USBDevice *dev, int pid, 
int ep);
 void usb_ep_set_pipeline(USBDevice *dev, int pid, int ep, bool enabled);
 USBPacket *usb_ep_find_packet_by_id(USBDevice *dev, int pid, int ep,
                                     uint64_t id);
+static inline bool usb_ep_input_pipeline(USBEndpoint *ep)
+{
+    return ep->pid == USB_TOKEN_IN && ep->pipeline;
+}
+static inline bool usb_ep_output_pipeline(USBEndpoint *ep)
+{
+    return ep->pid == USB_TOKEN_OUT && ep->pipeline;
+}
+void usb_ep_process_queue(USBEndpoint *ep);
 
 void usb_attach(USBPort *port);
 void usb_detach(USBPort *port);
diff --git a/hw/usb/core.c b/hw/usb/core.c
index b9f1f7a..54db17b 100644
--- a/hw/usb/core.c
+++ b/hw/usb/core.c
@@ -388,7 +388,8 @@ int usb_handle_packet(USBDevice *dev, USBPacket *p)
         p->ep->halted = false;
     }
 
-    if (QTAILQ_EMPTY(&p->ep->queue) || p->ep->pipeline) {
+    if (!usb_ep_input_pipeline(p->ep) &&
+            (QTAILQ_EMPTY(&p->ep->queue) || usb_ep_output_pipeline(p->ep))) {
         ret = usb_process_one(p);
         if (ret == USB_RET_ASYNC) {
             usb_packet_set_state(p, USB_PACKET_ASYNC);
@@ -398,7 +399,8 @@ int usb_handle_packet(USBDevice *dev, USBPacket *p)
              * When pipelining is enabled usb-devices must always return async,
              * otherwise packets can complete out of order!
              */
-            assert(!p->ep->pipeline || QTAILQ_EMPTY(&p->ep->queue));
+            assert(!usb_ep_output_pipeline(p->ep) ||
+                   QTAILQ_EMPTY(&p->ep->queue));
             if (ret != USB_RET_NAK) {
                 p->result = ret;
                 usb_packet_set_state(p, USB_PACKET_COMPLETE);
@@ -412,6 +414,21 @@ int usb_handle_packet(USBDevice *dev, USBPacket *p)
     return ret;
 }
 
+static void usb_add_packet_to_uncombined(USBPacket *dst, USBPacket *src)
+{
+    qemu_iovec_concat(&dst->iov, &src->iov, 0, src->iov.size);
+    QTAILQ_REMOVE(&dst->ep->queue, src, queue);
+    QTAILQ_INSERT_TAIL(&dst->uncombined_packets, src, queue);
+    src->combined_in = dst;
+}
+
+static void usb_remove_packet_from_uncombined(USBPacket *dst, USBPacket *src)
+{
+    QTAILQ_REMOVE(&dst->uncombined_packets, src, queue);
+    usb_packet_set_state(src, USB_PACKET_COMPLETE);
+    qemu_iovec_destroy(&dst->iov);
+}
+
 static void __usb_packet_complete(USBDevice *dev, USBPacket *p)
 {
     USBEndpoint *ep = p->ep;
@@ -421,11 +438,133 @@ static void __usb_packet_complete(USBDevice *dev, 
USBPacket *p)
     if (p->result < 0) {
         ep->halted = true;
     }
+
     usb_packet_set_state(p, USB_PACKET_COMPLETE);
+
+    /* If this is a combined packet, split it */
+    if (!QTAILQ_EMPTY(&p->uncombined_packets)) {
+        USBPacket *uncombined, *next;
+        bool need_zero_packet = true;
+        bool free_p = true;
+
+        QTAILQ_FOREACH_SAFE(uncombined, &p->uncombined_packets, queue, next) {
+            if (p->result > 0) {
+                /* Distribute data over uncombined packets */
+                if (p->result >= uncombined->iov.size) {
+                    uncombined->result = uncombined->iov.size;
+                } else {
+                    if (p->short_not_ok) {
+                        ep->halted = true;
+                    }
+                    uncombined->result = p->result;
+                    need_zero_packet = false;
+                }
+                p->result -= uncombined->result;
+                usb_remove_packet_from_uncombined(p, uncombined);
+                dev->port->ops->complete(dev->port, uncombined);
+            } else if (need_zero_packet) {
+                /* Send terminating 0 or error status packet */
+                if (p->short_not_ok) {
+                    ep->halted = true;
+                }
+                uncombined->result = p->result;
+                usb_remove_packet_from_uncombined(p, uncombined);
+                dev->port->ops->complete(dev->port, uncombined);
+                need_zero_packet = false;
+            } else {
+                /* Remove any leftover packets from the queue */
+                dev->port->ops->remove_from_queue(dev->port, uncombined);
+                free_p = false; /* remove_from_queue calls usb_cancel_packet */
+            }
+        }
+        if (free_p) {
+            QTAILQ_REMOVE(&ep->queue, p, queue);
+            g_free(p);
+        }
+        return;
+    }
+
+    if (p->short_not_ok && (p->result < p->iov.size)) {
+        ep->halted = true;
+    }
     QTAILQ_REMOVE(&ep->queue, p, queue);
     dev->port->ops->complete(dev->port, p);
 }
 
+/*
+ * Large input transfers can get split into multiple input packets, this
+ * function recombines them, removing the short_not_ok checks which all but
+ * the last packet of such splits transfers have, thereby allowing input
+ * transfer pipelining (which we cannot do on short_not_ok transfers)
+ */
+void usb_ep_process_queue(USBEndpoint *ep)
+{
+    USBPacket *u, *n, *p, *next, *prev = NULL, *first = NULL;
+    USBPort *port = ep->dev->port;
+
+    if (!usb_ep_input_pipeline(ep)) {
+        return;
+    }
+
+    QTAILQ_FOREACH_SAFE(p, &ep->queue, queue, next) {
+        /* Empty the queue on a halt */
+        if (ep->halted) {
+            if (!QTAILQ_EMPTY(&p->uncombined_packets)) {
+                QTAILQ_FOREACH_SAFE(u, &p->uncombined_packets, queue, n) {
+                    port->ops->remove_from_queue(port, u);
+                }
+            } else {
+                port->ops->remove_from_queue(port, p);
+            }
+            continue;
+        }
+
+        /* Skip packets already submitted to the device */
+        if (p->state == USB_PACKET_ASYNC) {
+            prev = p;
+            continue;
+        }
+        usb_packet_check_state(p, USB_PACKET_QUEUED);
+
+        /*
+         * If the previous (combined) packet has the short_not_ok flag set
+         * stop, as we must not submit packets to the device after a transfer
+         * ending with short_not_ok packet.
+         */
+        if (prev && prev->short_not_ok) {
+            break;
+        }
+
+        if (first) {
+            if (QTAILQ_EMPTY(&first->uncombined_packets)) {
+                USBPacket *new_first = g_new0(USBPacket, 1);
+
+                usb_packet_init(new_first);
+                usb_packet_setup(new_first, first->pid, first->ep, first->id);
+                new_first->state = USB_PACKET_QUEUED;
+
+                usb_add_packet_to_uncombined(new_first, first);
+                QTAILQ_INSERT_BEFORE(p, new_first, queue);
+                first = new_first;
+            }
+            usb_add_packet_to_uncombined(first, p);
+            first->short_not_ok = p->short_not_ok;
+        } else {
+            first = p;
+        }
+
+        /* Is this packet the last one of a (combined) transfer? */
+        if ((p->iov.size % ep->max_packet_size) != 0 || !p->short_not_ok ||
+                p->int_req || next == NULL) {
+            int ret = usb_process_one(first);
+            assert(ret == USB_RET_ASYNC);
+            usb_packet_set_state(first, USB_PACKET_ASYNC);
+            prev = first;
+            first = NULL;
+        }
+    }
+}
+
 /* Notify the controller that an async packet is complete.  This should only
    be called for packets previously deferred by returning USB_RET_ASYNC from
    handle_packet. */
@@ -438,6 +577,11 @@ void usb_packet_complete(USBDevice *dev, USBPacket *p)
     assert(QTAILQ_FIRST(&ep->queue) == p);
     __usb_packet_complete(dev, p);
 
+    if (usb_ep_input_pipeline(ep)) {
+        usb_ep_process_queue(ep);
+        return;
+    }
+
     while (!ep->halted && !QTAILQ_EMPTY(&ep->queue)) {
         p = QTAILQ_FIRST(&ep->queue);
         if (p->state == USB_PACKET_ASYNC) {
@@ -459,19 +603,34 @@ void usb_packet_complete(USBDevice *dev, USBPacket *p)
    completed.  */
 void usb_cancel_packet(USBPacket * p)
 {
-    bool callback = (p->state == USB_PACKET_ASYNC);
+    USBPacket *free_me = NULL;
+    bool callback;
+
     assert(usb_packet_is_inflight(p));
+
+    /* Is this part of a combined packet? */
+    if (p->combined_in) {
+        usb_remove_packet_from_uncombined(p->combined_in, p);
+        if (!QTAILQ_EMPTY(&p->combined_in->uncombined_packets)) {
+            return;
+        }
+        free_me = p = p->combined_in;
+    }
+
+    callback = (p->state == USB_PACKET_ASYNC);
     usb_packet_set_state(p, USB_PACKET_CANCELED);
     QTAILQ_REMOVE(&p->ep->queue, p, queue);
     if (callback) {
         usb_device_cancel_packet(p->ep->dev, p);
     }
+    g_free(free_me);
 }
 
 
 void usb_packet_init(USBPacket *p)
 {
     qemu_iovec_init(&p->iov, 1);
+    QTAILQ_INIT(&p->uncombined_packets);
 }
 
 static const char *usb_packet_state_name(USBPacketState state)
@@ -531,6 +690,9 @@ void usb_packet_setup(USBPacket *p, int pid, USBEndpoint 
*ep, uint64_t id)
     p->ep = ep;
     p->result = 0;
     p->parameter = 0;
+    p->combined_in = NULL;
+    p->short_not_ok = false;
+    p->int_req = false;
     qemu_iovec_reset(&p->iov);
     usb_packet_set_state(p, USB_PACKET_SETUP);
 }
@@ -724,7 +886,13 @@ int usb_ep_get_max_packet_size(USBDevice *dev, int pid, 
int ep)
 void usb_ep_set_pipeline(USBDevice *dev, int pid, int ep, bool enabled)
 {
     struct USBEndpoint *uep = usb_ep_get(dev, pid, ep);
-    uep->pipeline = enabled;
+
+    if (pid == USB_TOKEN_IN) {
+        uep->pipeline =
+            enabled && uep->max_packet_size != 0 && 
dev->port->supports_queuing;
+    } else {
+        uep->pipeline = enabled;
+    }
 }
 
 USBPacket *usb_ep_find_packet_by_id(USBDevice *dev, int pid, int ep,
diff --git a/hw/usb/dev-hub.c b/hw/usb/dev-hub.c
index 8fd30df..d326281 100644
--- a/hw/usb/dev-hub.c
+++ b/hw/usb/dev-hub.c
@@ -222,6 +222,13 @@ static void usb_hub_complete(USBPort *port, USBPacket 
*packet)
     s->dev.port->ops->complete(s->dev.port, packet);
 }
 
+static void usb_hub_remove_from_queue(USBPort *port, USBPacket *packet)
+{
+    USBHubState *s = port->opaque;
+
+    s->dev.port->ops->remove_from_queue(s->dev.port, packet);
+}
+
 static USBDevice *usb_hub_find_device(USBDevice *dev, uint8_t addr)
 {
     USBHubState *s = DO_UPCAST(USBHubState, dev, dev);
@@ -512,6 +519,7 @@ static USBPortOps usb_hub_port_ops = {
     .child_detach = usb_hub_child_detach,
     .wakeup = usb_hub_wakeup,
     .complete = usb_hub_complete,
+    .remove_from_queue = usb_hub_remove_from_queue,
 };
 
 static int usb_hub_initfn(USBDevice *dev)
@@ -529,6 +537,7 @@ static int usb_hub_initfn(USBDevice *dev)
                           &port->port, s, i, &usb_hub_port_ops,
                           USB_SPEED_MASK_LOW | USB_SPEED_MASK_FULL);
         usb_port_location(&port->port, dev->port, i+1);
+        port->port.supports_queuing = dev->port->supports_queuing;
     }
     usb_hub_handle_reset(dev);
     return 0;
-- 
1.7.12.1