setReadQueue(...), enabled by default in SerialInputOutputManager

for applications doing permanent read() with timeout=0, multiple buffers can be used to copy next data from Linux kernel, while the current data is processed.

Author: kai-morich

usbSerialForAndroid/src/androidTest/java/com/hoho/android/usbserial/DeviceTest.java

@@ -14,6 +14,7 @@
 import android.hardware.usb.UsbDevice;
 import android.hardware.usb.UsbDeviceConnection;
 import android.hardware.usb.UsbManager;
+import android.hardware.usb.UsbRequest;
 import android.os.Process;
 import androidx.test.core.app.ApplicationProvider;
 import androidx.test.platform.app.InstrumentationRegistry;
@@ -43,7 +44,6 @@
 import com.hoho.android.usbserial.driver.UsbSerialPort.FlowControl;
 import com.hoho.android.usbserial.util.XonXoffFilter;
 
-
 import org.junit.After;
 import org.junit.AfterClass;
 import org.junit.Assume;
@@ -60,11 +60,13 @@
 import java.nio.BufferOverflowException;
 import java.util.Arrays;
 import java.util.EnumSet;
+import java.util.LinkedList;
 import java.util.List;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 
 import static org.hamcrest.CoreMatchers.anyOf;
 import static org.hamcrest.MatcherAssert.assertThat;
@@ -1148,7 +1150,7 @@ else if (usb.serialDriver instanceof ProlificSerialDriver)
     public void readBufferOverflow() throws Exception {
         if(usb.serialDriver instanceof CdcAcmSerialDriver)
             telnet.writeDelay = 10; // arduino_leonardo_bridge.ino sends each byte in own USB packet, which is horribly slow
-        usb.open();
+        usb.open(EnumSet.of(UsbWrapper.OpenCloseFlags.NO_IOMANAGER_READQUEUE));
         usb.setParameters(115200, 8, 1, UsbSerialPort.PARITY_NONE);
         telnet.setParameters(115200, 8, 1, UsbSerialPort.PARITY_NONE);
 
@@ -1198,6 +1200,64 @@ public void readBufferOverflow() throws Exception {
         assertTrue(data.length() != expected.length());
     }
 
+    @Test
+    public void readQueue() throws Exception {
+        class CountingUsbRequest extends UsbRequest {
+            int count;
+            @Override public Object getClientData() { count += 1; return super.getClientData(); }
+        }
+
+        usb.open(EnumSet.of(UsbWrapper.OpenCloseFlags.NO_IOMANAGER_THREAD));
+        int len = usb.serialPort.getReadEndpoint().getMaxPacketSize();
+        usb.close();
+        CommonUsbSerialPortWrapper.setReadQueueRequestSupplier(usb.serialPort, CountingUsbRequest::new);
+        CommonUsbSerialPort port = (CommonUsbSerialPort) usb.serialPort;
+
+        port.setReadQueue(2, len);
+        usb.open(EnumSet.of(UsbWrapper.OpenCloseFlags.NO_IOMANAGER_START));
+        usb.setParameters(115200, 8, 1, UsbSerialPort.PARITY_NONE);
+        telnet.setParameters(115200, 8, 1, UsbSerialPort.PARITY_NONE);
+        assertEquals(2, port.getReadQueueBufferCount());
+        assertEquals(4, usb.ioManager.getReadQueueBufferCount()); // not set at port yet
+        assertThrows(IllegalStateException.class, () -> usb.ioManager.setReadQueue(1)); // cannot reduce bufferCount
+        usb.ioManager.setReadQueue(2);
+        usb.ioManager.start();
+        port.setReadQueue(4, len);
+
+        // linux kernel does round-robin
+        LinkedList<UsbRequest> requests = CommonUsbSerialPortWrapper.getReadQueueRequests(usb.serialPort);
+        assertNotNull(requests);
+        for (int i=0; i<16; i++) {
+            telnet.write(new byte[1]);
+            usb.read(1);
+        }
+        List<Integer> requestCounts;
+        if(usb.serialDriver instanceof FtdiSerialDriver) {
+            for (UsbRequest request : requests) {
+                int count = ((CountingUsbRequest)request).count;
+                assertTrue(String.valueOf(count), count >= 4);
+            }
+        } else {
+            requestCounts = requests.stream().map(r -> ((CountingUsbRequest)r).count).collect(Collectors.toList());
+            assertThat(requestCounts, equalTo(Arrays.asList(4, 4, 4, 4)));
+        }
+        usb.ioManager.setReadQueue(6);
+        for (int i=0; i<18; i++) {
+            telnet.write(new byte[1]);
+            usb.read(1);
+        }
+        requestCounts = requests.stream().map(r -> ((CountingUsbRequest)r).count).collect(Collectors.toList());
+        if(!(usb.serialDriver instanceof FtdiSerialDriver)) {
+            assertThat(requestCounts, equalTo(Arrays.asList(7, 7, 7, 7, 3, 3)));
+        }
+        usb.close();
+        usb.open(EnumSet.of(UsbWrapper.OpenCloseFlags.NO_IOMANAGER_START));
+        port.setReadQueue(8, len);
+        assertThrows(IllegalStateException.class, () -> usb.serialPort.read(new byte[len], 1) ); // cannot use timeout != 0
+        assertThrows(IllegalStateException.class, () -> usb.serialPort.read(new byte[4], 0) ); // cannot use different length
+        assertThrows(IllegalStateException.class, () -> usb.ioManager.start()); // cannot reduce bufferCount
+    }
+
     @Test
     public void readSpeed() throws Exception {
         // see logcat for performance results
@@ -1223,7 +1283,7 @@ private int readSpeedInt(int writeSeconds, int readBufferSize, int readTimeout)
         if(usb.serialDriver instanceof CdcAcmSerialDriver)
             writeAhead = 50;
 
-        usb.open(EnumSet.of(UsbWrapper.OpenCloseFlags.NO_IOMANAGER_START));
+        usb.open(EnumSet.of(UsbWrapper.OpenCloseFlags.NO_IOMANAGER_START, UsbWrapper.OpenCloseFlags.NO_IOMANAGER_READQUEUE));
         usb.ioManager.setReadTimeout(readTimeout);
         if(readBufferSize > 0)
             usb.ioManager.setReadBufferSize(readBufferSize);
@@ -1421,7 +1481,7 @@ public void IoManager() throws Exception {
         usb.ioManager.setWriteTimeout(usb.ioManager.getWriteTimeout());
         usb.close();
 
-        usb.open(EnumSet.of(UsbWrapper.OpenCloseFlags.NO_IOMANAGER_START)); // creates new IoManager
+        usb.open(EnumSet.of(UsbWrapper.OpenCloseFlags.NO_IOMANAGER_START, UsbWrapper.OpenCloseFlags.NO_IOMANAGER_READQUEUE)); // creates new IoManager
         usb.setParameters(19200, 8, 1, UsbSerialPort.PARITY_NONE);
         telnet.setParameters(19200, 8, 1, UsbSerialPort.PARITY_NONE);
         usb.ioManager.setThreadPriority(Process.THREAD_PRIORITY_DEFAULT);

usbSerialForAndroid/src/androidTest/java/com/hoho/android/usbserial/driver/CommonUsbSerialPortWrapper.java

@@ -1,8 +1,24 @@
 package com.hoho.android.usbserial.driver;
 
+import android.hardware.usb.UsbRequest;
+
+import com.hoho.android.usbserial.util.UsbUtils;
+
+import java.util.LinkedList;
+
 public class CommonUsbSerialPortWrapper {
     public static byte[] getWriteBuffer(UsbSerialPort serialPort) {
         CommonUsbSerialPort commonSerialPort = (CommonUsbSerialPort) serialPort;
         return commonSerialPort.mWriteBuffer;
     }
+
+    public static LinkedList<UsbRequest> getReadQueueRequests(UsbSerialPort serialPort) {
+        CommonUsbSerialPort commonSerialPort = (CommonUsbSerialPort) serialPort;
+        return commonSerialPort.mReadQueueRequests;
+    }
+
+    public static void setReadQueueRequestSupplier(UsbSerialPort serialPort, UsbUtils.Supplier<UsbRequest> supplier) {
+        CommonUsbSerialPort commonSerialPort = (CommonUsbSerialPort) serialPort;
+        commonSerialPort.mUsbRequestSupplier = supplier;
+    }
 }

usbSerialForAndroid/src/androidTest/java/com/hoho/android/usbserial/util/UsbWrapper.java

@@ -41,7 +41,7 @@ public class UsbWrapper implements SerialInputOutputManager.Listener {
     public final static int     USB_WRITE_WAIT = 500;
     private static final String TAG = UsbWrapper.class.getSimpleName();
 
-    public enum OpenCloseFlags { NO_IOMANAGER_THREAD, NO_IOMANAGER_START, NO_CONTROL_LINE_INIT, NO_DEVICE_CONNECTION };
+    public enum OpenCloseFlags { NO_IOMANAGER_THREAD, NO_IOMANAGER_READQUEUE, NO_IOMANAGER_START, NO_CONTROL_LINE_INIT, NO_DEVICE_CONNECTION };
 
     // constructor
     final Context context;
@@ -198,7 +198,7 @@ public void close(EnumSet<OpenCloseFlags> flags) {
                 serialPort.close();
             } catch (Exception ignored) {
             }
-            //usbSerialPort = null;
+            ((CommonUsbSerialPort)serialPort).setReadQueue(0, 0);
         }
         if(!flags.contains(OpenCloseFlags.NO_DEVICE_CONNECTION)) {
             deviceConnection = null; // closed in usbSerialPort.close()
@@ -233,6 +233,8 @@ public void open(EnumSet<OpenCloseFlags> flags) throws Exception {
         }
         if(!flags.contains(OpenCloseFlags.NO_IOMANAGER_THREAD)) {
             ioManager = new SerialInputOutputManager(serialPort, this);
+            if(flags.contains(OpenCloseFlags.NO_IOMANAGER_READQUEUE))
+                ioManager.setReadQueue(0);
             if(!flags.contains(OpenCloseFlags.NO_IOMANAGER_START))
                 ioManager.start();
         }

usbSerialForAndroid/src/main/java/com/hoho/android/usbserial/driver/CommonUsbSerialPort.java

@@ -14,10 +14,13 @@
 import android.util.Log;
 
 import com.hoho.android.usbserial.util.MonotonicClock;
+import com.hoho.android.usbserial.util.UsbUtils;
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.EnumSet;
+import java.util.LinkedList;
+import java.util.Objects;
 
 /**
  * A base class shared by several driver implementations.
@@ -38,8 +41,12 @@ public abstract class CommonUsbSerialPort implements UsbSerialPort {
     protected UsbDeviceConnection mConnection;
     protected UsbEndpoint mReadEndpoint;
     protected UsbEndpoint mWriteEndpoint;
-    protected UsbRequest mUsbRequest;
+    protected UsbRequest mReadRequest;
+    protected LinkedList<UsbRequest> mReadQueueRequests;
+    private int mReadQueueBufferCount;
+    private int mReadQueueBufferSize;
     protected FlowControl mFlowControl = FlowControl.NONE;
+    protected UsbUtils.Supplier<UsbRequest> mUsbRequestSupplier = UsbRequest::new; // override for testing
 
     /**
      * Internal write buffer.
@@ -110,6 +117,50 @@ public final void setWriteBufferSize(int bufferSize) {
         }
     }
 
+    /**
+     * for applications doing permanent read() with timeout=0, multiple buffers can be
+     * used to copy next data from Linux kernel, while the current data is processed.
+     * @param bufferCount number of buffers to use for readQueue
+     *                    disabled with 0
+     * @param bufferSize size of each buffer
+     */
+    public void setReadQueue(int bufferCount, int bufferSize) {
+        if (bufferCount < 0) {
+            throw new IllegalArgumentException("Invalid bufferCount");
+        }
+        if (bufferCount > 0 && bufferSize <= 0) {
+            throw new IllegalArgumentException("Invalid bufferSize");
+        }
+        if(isOpen()) {
+            if (bufferCount < mReadQueueBufferCount) {
+                throw new IllegalStateException("Cannot reduce bufferCount when port is open");
+            }
+            if (mReadQueueBufferCount != 0 && bufferSize != mReadQueueBufferSize) {
+                throw new IllegalStateException("Cannot change bufferSize when port is open");
+            }
+            if (bufferCount > 0) {
+                if (mReadQueueRequests == null) {
+                    mReadQueueRequests = new LinkedList<>();
+                }
+                for (int i = mReadQueueRequests.size(); i < bufferCount; i++) {
+                    ByteBuffer buffer = ByteBuffer.allocate(bufferSize);
+                    UsbRequest request = mUsbRequestSupplier.get();
+                    request.initialize(mConnection, mReadEndpoint);
+                    request.setClientData(buffer);
+                    request.queue(buffer, bufferSize);
+                    mReadQueueRequests.add(request);
+                }
+            }
+        }
+        mReadQueueBufferCount = bufferCount;
+        mReadQueueBufferSize = bufferSize;
+    }
+
+    public int getReadQueueBufferCount() { return mReadQueueBufferCount; }
+    public int getReadQueueBufferSize() { return mReadQueueBufferSize; }
+
+    private boolean useReadQueue() { return mReadQueueBufferCount != 0; }
+
     @Override
     public void open(UsbDeviceConnection connection) throws IOException {
         if (mConnection != null) {
@@ -125,8 +176,9 @@ public void open(UsbDeviceConnection connection) throws IOException {
             if (mReadEndpoint == null || mWriteEndpoint == null) {
                 throw new IOException("Could not get read & write endpoints");
             }
-            mUsbRequest = new UsbRequest();
-            mUsbRequest.initialize(mConnection, mReadEndpoint);
+            mReadRequest = mUsbRequestSupplier.get();
+            mReadRequest.initialize(mConnection, mReadEndpoint);
+            setReadQueue(mReadQueueBufferCount, mReadQueueBufferSize); // fill mReadQueueRequests
             ok = true;
         } finally {
             if (!ok) {
@@ -144,11 +196,19 @@ public void close() throws IOException {
         if (mConnection == null) {
             throw new IOException("Already closed");
         }
-        UsbRequest usbRequest = mUsbRequest;
-        mUsbRequest = null;
+        UsbRequest readRequest = mReadRequest;
+        mReadRequest = null;
         try {
-            usbRequest.cancel();
+            readRequest.cancel();
         } catch(Exception ignored) {}
+        if(mReadQueueRequests != null) {
+            for(UsbRequest readQueueRequest : mReadQueueRequests) {
+                try {
+                    readQueueRequest.cancel();
+                } catch(Exception ignored) {}
+            }
+            mReadQueueRequests = null;
+        }
         try {
             closeInt();
         } catch(Exception ignored) {}
@@ -168,7 +228,7 @@ protected void testConnection(boolean full) throws IOException {
     }
 
     protected void testConnection(boolean full, String msg) throws IOException {
-        if(mUsbRequest == null) {
+        if(mReadRequest == null) {
             throw new IOException("Connection closed");
         }
         if(!full) {
@@ -199,6 +259,9 @@ protected int read(final byte[] dest, int length, final int timeout, boolean tes
         length = Math.min(length, dest.length);
         final int nread;
         if (timeout != 0) {
+            if(useReadQueue()) {
+                throw new IllegalStateException("Cannot use timeout!=0 if readQueue is enabled");
+            }
             // bulkTransfer will cause data loss with short timeout + high baud rates + continuous transfer
             //   https://stackoverflow.com/questions/9108548/android-usb-host-bulktransfer-is-losing-data
             // but mConnection.requestWait(timeout) available since Android 8.0 es even worse,
@@ -216,15 +279,31 @@ protected int read(final byte[] dest, int length, final int timeout, boolean tes
                 testConnection(MonotonicClock.millis() < endTime);
 
         } else {
-            final ByteBuffer buf = ByteBuffer.wrap(dest, 0, length);
-            if (!mUsbRequest.queue(buf, length)) {
-                throw new IOException("Queueing USB request failed");
+            ByteBuffer buf = null;
+            if(useReadQueue()) {
+                if (length != mReadQueueBufferSize) {
+                    throw new IllegalStateException("Cannot use different length if readQueue is enabled");
+                }
+            } else {
+                buf = ByteBuffer.wrap(dest, 0, length);
+                if (!mReadRequest.queue(buf, length)) {
+                    throw new IOException("Queueing USB request failed");
+                }
             }
             final UsbRequest response = mConnection.requestWait();
             if (response == null) {
                 throw new IOException("Waiting for USB request failed");
             }
-            nread = buf.position();
+            if(useReadQueue()) {
+                buf = (ByteBuffer) response.getClientData();
+                System.arraycopy(buf.array(), 0, dest, 0, buf.position());
+                if(mReadRequest != null) { // re-queue if connection not closed
+                    if (!response.queue(buf, buf.capacity())) {
+                        throw new IOException("Queueing USB request failed");
+                    }
+                }
+            }
+            nread = Objects.requireNonNull(buf).position();
             // Android error propagation is improvable:
             //   response != null & nread == 0 can be: connection lost, buffer to small, ???
             if(nread == 0) {
@@ -297,7 +376,7 @@ public void write(final byte[] src, int length, final int timeout) throws IOExce
 
     @Override
     public boolean isOpen() {
-        return mUsbRequest != null;
+        return mReadRequest != null;
     }
 
     @Override