usb-linux: massive read perf improvement with 3 parallel transfers

By maintaining 3 parallel usb trasfers when reading we get 2-3x more
throughput when reading. Without this the usb port is mostly just idling.
I get 23mb/s on my system compared to a clean Apple stack that gives me
17mb/s.

3 was chosen because it is simple to hard code, gives very good performance,
and have very little impact on out resource consumption.

1 files changed, 73 insertions, 34 deletions

diff --git a/src/usb-linux.c b/src/usb-linux.c
index 94db8f2..40bf502 100644
--- a/src/usb-linux.c
+++ b/src/usb-linux.c
@@ -4,6 +4,7 @@
 Copyright (C) 2009	Hector Martin "marcan" <hector@marcansoft.com>
 Copyright (C) 2009	Nikias Bassen <nikias@gmx.li>
 Copyright (C) 2009	Martin Szulecki <opensuse@sukimashita.com>
+Copyright (C) 2014  Mikkel Kamstrup Erlandsen <mikkel.kamstrup@xamarin.com>
 
 This program is free software; you can redistribute it and/or modify
 it under the terms of the GNU General Public License as published by
@@ -39,6 +40,12 @@ Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
 // we need this because there is currently no asynchronous device discovery mechanism in libusb
 #define DEVICE_POLL_TIME 1000
 
+// Number of parallel bulk transfers we have running for reading data from the device.
+// Older versions of usbmuxd kept only 1, which leads to a mostly dormant USB port.
+// 3 seems to be an all round sensible number - giving better read perf than
+// Apples usbmuxd, at least.
+#define NUM_RX_LOOPS 3
+
 struct usb_device {
 	libusb_device_handle *dev;
 	uint8_t bus, address;
@@ -46,7 +53,7 @@ struct usb_device {
 	char serial[256];
 	int alive;
 	uint8_t interface, ep_in, ep_out;
-	struct libusb_transfer *rx_xfer;
+	struct collection rx_xfers;
 	struct collection tx_xfers;
 	int wMaxPacketSize;
 };
@@ -64,17 +71,20 @@ static void usb_disconnect(struct usb_device *dev)
 		return;
 	}
 
-	// kill the rx xfer and tx xfers and try to make sure the callbacks get called before we free the device
-	if(dev->rx_xfer) {
-		usbmuxd_log(LL_DEBUG, "usb_disconnect: cancelling RX xfer");
-		libusb_cancel_transfer(dev->rx_xfer);
-	}
+	// kill the rx xfer and tx xfers and try to make sure the callbacks
+	// get called before we free the device
+	FOREACH(struct libusb_transfer *xfer, &dev->rx_xfers) {
+		usbmuxd_log(LL_DEBUG, "usb_disconnect: cancelling RX xfer %p", xfer);
+		libusb_cancel_transfer(xfer);
+	} ENDFOREACH
+
 	FOREACH(struct libusb_transfer *xfer, &dev->tx_xfers) {
 		usbmuxd_log(LL_DEBUG, "usb_disconnect: cancelling TX xfer %p", xfer);
 		libusb_cancel_transfer(xfer);
 	} ENDFOREACH
 
-	while(dev->rx_xfer || collection_count(&dev->tx_xfers)) {
+	// Busy-wait until all xfers are closed
+	while(collection_count(&dev->rx_xfers) || collection_count(&dev->tx_xfers)) {
 		struct timeval tv;
 		int res;
 
@@ -85,7 +95,9 @@ static void usb_disconnect(struct usb_device *dev)
 			break;
 		}
 	}
+
 	collection_free(&dev->tx_xfers);
+	collection_free(&dev->rx_xfers);
 	libusb_release_interface(dev->dev, dev->interface);
 	libusb_close(dev->dev);
 	dev->dev = NULL;
@@ -93,6 +105,15 @@ static void usb_disconnect(struct usb_device *dev)
 	free(dev);
 }
 
+static void reap_dead_devices(void) {
+	FOREACH(struct usb_device *usbdev, &device_list) {
+		if(!usbdev->alive) {
+			device_remove(usbdev);
+			usb_disconnect(usbdev);
+		}
+	} ENDFOREACH
+}
+
 // Callback from write operation
 static void tx_callback(struct libusb_transfer *xfer)
 {
@@ -201,9 +222,11 @@ static void rx_callback(struct libusb_transfer *xfer)
 				// this should never be reached.
 				break;
 		}
+
 		free(xfer->buffer);
-		dev->rx_xfer = NULL;
+		collection_remove(&dev->rx_xfers, xfer);
 		libusb_free_transfer(xfer);
+
 		// we can't usb_disconnect here due to a deadlock, so instead mark it as dead and reap it after processing events
 		// we'll do device_remove there too
 		dev->alive = 0;
@@ -211,19 +234,21 @@ static void rx_callback(struct libusb_transfer *xfer)
 }
 
 // Start a read-callback loop for this device
-static int start_rx(struct usb_device *dev)
+static int start_rx_loop(struct usb_device *dev)
 {
 	int res;
 	void *buf;
-	dev->rx_xfer = libusb_alloc_transfer(0);
+	struct libusb_transfer *xfer = libusb_alloc_transfer(0);
 	buf = malloc(USB_MRU);
-	libusb_fill_bulk_transfer(dev->rx_xfer, dev->dev, dev->ep_in, buf, USB_MRU, rx_callback, dev, 0);
-	if((res = libusb_submit_transfer(dev->rx_xfer)) != 0) {
+	libusb_fill_bulk_transfer(xfer, dev->dev, dev->ep_in, buf, USB_MRU, rx_callback, dev, 0);
+	if((res = libusb_submit_transfer(xfer)) != 0) {
 		usbmuxd_log(LL_ERROR, "Failed to submit RX transfer to device %d-%d: %d", dev->bus, dev->address, res);
-		libusb_free_transfer(dev->rx_xfer);
-		dev->rx_xfer = NULL;
+		libusb_free_transfer(xfer);
 		return res;
 	}
+
+	collection_add(&dev->rx_xfers, xfer);
+
 	return 0;
 }
 
@@ -253,10 +278,14 @@ int usb_discover(void)
 
 	usbmuxd_log(LL_SPEW, "usb_discover: scanning %d devices", cnt);
 
+	// Mark all devices as dead, and do a mark-sweep like
+	// collection of dead devices
 	FOREACH(struct usb_device *usbdev, &device_list) {
 		usbdev->alive = 0;
 	} ENDFOREACH
 
+	// Enumerate all USB devices and mark the ones we already know
+	// about as live, again
 	for(i=0; i<cnt; i++) {
 		// the following are non-blocking operations on the device list
 		libusb_device *dev = devs[i];
@@ -407,6 +436,7 @@ int usb_discover(void)
 		}
 
 		collection_init(&usbdev->tx_xfers);
+		collection_init(&usbdev->rx_xfers);
 
 		collection_add(&device_list, usbdev);
 
@@ -414,19 +444,37 @@ int usb_discover(void)
 			usb_disconnect(usbdev);
 			continue;
 		}
-		if(start_rx(usbdev) < 0) {
+
+		// Spin up NUM_RX_LOOPS parallel usb data retrieval loops
+		// Old usbmuxds used only 1 rx loop, but that leaves the
+		// USB port sleeping most of the time
+		int rx_loops = NUM_RX_LOOPS;
+		for (rx_loops = NUM_RX_LOOPS; rx_loops > 0; rx_loops--) {
+			if(start_rx_loop(usbdev) < 0) {
+				usbmuxd_log(LL_WARNING, "Failed to start RX loop number %d", NUM_RX_LOOPS - rx_loops);
+			}
+		}
+
+		// Ensure we have at least 1 RX loop going
+		if (rx_loops == NUM_RX_LOOPS) {
+			usbmuxd_log(LL_FATAL, "Failed to start any RX loop for device %d-%d",
+						usbdev->bus, usbdev->address);
 			device_remove(usbdev);
 			usb_disconnect(usbdev);
 			continue;
+		} else if (rx_loops > 0) {
+			usbmuxd_log(LL_WARNING, "Failed to start all %d RX loops. Going on with %d loops. "
+						"This may have negative impact on device read speed.",
+						NUM_RX_LOOPS, NUM_RX_LOOPS - rx_loops);
+		} else {
+			usbmuxd_log(LL_DEBUG, "All %d RX loops started successfully", NUM_RX_LOOPS);
 		}
+
 		valid_count++;
 	}
-	FOREACH(struct usb_device *usbdev, &device_list) {
-		if(!usbdev->alive) {
-			device_remove(usbdev);
-			usb_disconnect(usbdev);
-		}
-	} ENDFOREACH
+
+	// Clean out any device we didn't mark back as live
+	reap_dead_devices();
 
 	libusb_free_device_list(devs, 1);
 
@@ -530,13 +578,9 @@ int usb_process(void)
 		usbmuxd_log(LL_ERROR, "libusb_handle_events_timeout failed: %d", res);
 		return res;
 	}
+
 	// reap devices marked dead due to an RX error
-	FOREACH(struct usb_device *usbdev, &device_list) {
-		if(!usbdev->alive) {
-			device_remove(usbdev);
-			usb_disconnect(usbdev);
-		}
-	} ENDFOREACH
+	reap_dead_devices();
 
 	if(dev_poll_remain_ms() <= 0) {
 		res = usb_discover();
@@ -570,13 +614,8 @@ int usb_process_timeout(int msec)
 			return res;
 		}
 		// reap devices marked dead due to an RX error
-		FOREACH(struct usb_device *usbdev, &device_list) {
-			if(!usbdev->alive) {
-				device_remove(usbdev);
-				usb_disconnect(usbdev);
-			}
-		} ENDFOREACH
-	gettimeofday(&tcur, NULL);
+		reap_dead_devices();
+		gettimeofday(&tcur, NULL);
 	}
 	return 0;
 }