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.

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

@@ -14,6 +14,7 @@ import android.content.Context;
 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.ControlLine;
 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.io.IOException;
 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 @@ public class DeviceTest {
     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 class DeviceTest {
         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 @@ public class DeviceTest {
         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 class DeviceTest {
         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 class UsbWrapper implements SerialInputOutputManager.Listener {
                 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 class UsbWrapper implements SerialInputOutputManager.Listener {
         }
         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.os.Build;
 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 abstract class CommonUsbSerialPort implements UsbSerialPort {
         }
     }
 
+    /**
+     * 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 abstract class CommonUsbSerialPort implements UsbSerialPort {
             if (mReadEndpoint == null || mWriteEndpoint == null) {
                 throw new IOException("Could not get read & write endpoints");
             }
-            mUsbRequest = new UsbRequest();
+            mReadRequest = mUsbRequestSupplier.get();
-            mUsbRequest.initialize(mConnection, mReadEndpoint);
+            mReadRequest.initialize(mConnection, mReadEndpoint);
+            setReadQueue(mReadQueueBufferCount, mReadQueueBufferSize); // fill mReadQueueRequests
             ok = true;
         } finally {
             if (!ok) {
@@ -144,11 +196,19 @@ public abstract class CommonUsbSerialPort implements UsbSerialPort {
         if (mConnection == null) {
             throw new IOException("Already closed");
         }
-        UsbRequest usbRequest = mUsbRequest;
+        UsbRequest readRequest = mReadRequest;
-        mUsbRequest = null;
+        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 @@ public abstract class CommonUsbSerialPort implements UsbSerialPort {
     }
 
     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 @@ public abstract class CommonUsbSerialPort implements UsbSerialPort {
         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 @@ public abstract class CommonUsbSerialPort implements UsbSerialPort {
                 testConnection(MonotonicClock.millis() < endTime);
 
         } else {
-            final ByteBuffer buf = ByteBuffer.wrap(dest, 0, length);
+            ByteBuffer buf = null;
-            if (!mUsbRequest.queue(buf, length)) {
+            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 abstract class CommonUsbSerialPort implements UsbSerialPort {
 
     @Override
     public boolean isOpen() {
-        return mUsbRequest != null;
+        return mReadRequest != null;
     }
 
     @Override

usbSerialForAndroid/src/main/java/com/hoho/android/usbserial/util/SerialInputOutputManager.java

@@ -9,6 +9,7 @@ package com.hoho.android.usbserial.util;
 import android.os.Process;
 import android.util.Log;
 
+import com.hoho.android.usbserial.driver.CommonUsbSerialPort;
 import com.hoho.android.usbserial.driver.UsbSerialPort;
 
 import java.io.IOException;
@@ -33,16 +34,18 @@ public class SerialInputOutputManager {
     public static boolean DEBUG = false;
 
     private static final String TAG = SerialInputOutputManager.class.getSimpleName();
-    private static final int BUFSIZ = 4096;
+    private static final int WRITE_BUFFER_SIZE = 4096;
+    private static final int READ_QUEUE_BUFFER_COUNT = 4;
 
     private int mReadTimeout = 0;
     private int mWriteTimeout = 0;
+    private int mReadQueueBufferCount; // = READ_QUEUE_BUFFER_COUNT
 
     private final Object mReadBufferLock = new Object();
     private final Object mWriteBufferLock = new Object();
 
     private ByteBuffer mReadBuffer; // default size = getReadEndpoint().getMaxPacketSize()
-    private ByteBuffer mWriteBuffer = ByteBuffer.allocate(BUFSIZ);
+    private ByteBuffer mWriteBuffer = ByteBuffer.allocate(WRITE_BUFFER_SIZE);
 
     private int mThreadPriority = Process.THREAD_PRIORITY_URGENT_AUDIO;
     private final AtomicReference<State> mState = new AtomicReference<>(State.STOPPED);
@@ -65,12 +68,13 @@ public class SerialInputOutputManager {
     public SerialInputOutputManager(UsbSerialPort serialPort) {
         mSerialPort = serialPort;
         mReadBuffer = ByteBuffer.allocate(serialPort.getReadEndpoint().getMaxPacketSize());
+        mReadQueueBufferCount = serialPort instanceof CommonUsbSerialPort ? READ_QUEUE_BUFFER_COUNT : 0;
+        //readQueueBufferSize fixed to getMaxPacketSize()
     }
 
     public SerialInputOutputManager(UsbSerialPort serialPort, Listener listener) {
-        mSerialPort = serialPort;
+        this(serialPort);
         mListener = listener;
-        mReadBuffer = ByteBuffer.allocate(serialPort.getReadEndpoint().getMaxPacketSize());
     }
 
     public synchronized void setListener(Listener listener) {
@@ -145,6 +149,22 @@ public class SerialInputOutputManager {
         return mWriteBuffer.capacity();
     }
 
+    /**
+     * Set read queue. Set buffer size before.
+     * @param bufferCount number of buffers to use for readQueue
+     *                    disable with value 0,
+     *                    default enabled as value 4 (READ_QUEUE_BUFFER_COUNT)
+     */
+    public void setReadQueue(int bufferCount) {
+        if (!(mSerialPort instanceof CommonUsbSerialPort)) {
+            throw new IllegalArgumentException("only for CommonUsbSerialPort based drivers");
+        }
+        mReadQueueBufferCount = bufferCount;
+        ((CommonUsbSerialPort) mSerialPort).setReadQueue(getReadQueueBufferCount(), getReadBufferSize());
+    }
+
+    public int getReadQueueBufferCount() { return mReadQueueBufferCount; }
+
     /**
      * write data asynchronously
      */
@@ -159,6 +179,9 @@ public class SerialInputOutputManager {
      * start SerialInputOutputManager in separate threads
      */
     public void start() {
+        if(mSerialPort instanceof CommonUsbSerialPort) {
+            ((CommonUsbSerialPort) mSerialPort).setReadQueue(mReadQueueBufferCount, getReadBufferSize());
+        }
         if(mState.compareAndSet(State.STOPPED, State.STARTING)) {
             mStartuplatch = new CountDownLatch(2);
             new Thread(this::runRead, this.getClass().getSimpleName() + "_read").start();

usbSerialForAndroid/src/main/java/com/hoho/android/usbserial/util/UsbUtils.java

@@ -6,6 +6,11 @@ import java.util.ArrayList;
 
 public class UsbUtils {
 
+    // copied from java.util.function.Supplier which is not available < API 24
+    public interface Supplier<T> {
+        T get();
+    }
+
     private UsbUtils() {
     }
 
@@ -29,5 +34,4 @@ public class UsbUtils {
         return descriptors;
     }
 
-
 }

usbSerialForAndroid/src/test/java/com/hoho/android/usbserial/driver/CommonUsbSerialPortTest.java

@@ -0,0 +1,105 @@
+package com.hoho.android.usbserial.driver;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThrows;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import android.hardware.usb.UsbDevice;
+import android.hardware.usb.UsbDeviceConnection;
+import android.hardware.usb.UsbEndpoint;
+import android.hardware.usb.UsbRequest;
+
+import org.junit.Test;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+
+public class CommonUsbSerialPortTest {
+
+    static class DummySerialDriver implements UsbSerialDriver {
+        ArrayList<UsbSerialPort> ports = new ArrayList<>();
+
+        DummySerialDriver() { ports.add(new DummySerialPort(null, 0)); }
+
+        @Override
+        public UsbDevice getDevice() { return null; }
+
+        @Override
+        public List<UsbSerialPort> getPorts() { return ports; }
+    }
+
+    static class DummySerialPort extends CommonUsbSerialPort {
+        public DummySerialPort(UsbDevice device, int portNumber) {
+            super(device, portNumber);
+            mUsbRequestSupplier = DummyUsbRequest::new;
+        }
+
+        @Override
+        protected void openInt() throws IOException { mReadEndpoint = mWriteEndpoint = mock(UsbEndpoint.class); }
+
+        @Override
+        protected void closeInt() { }
+
+        @Override
+        public UsbSerialDriver getDriver() { return null; }
+
+        @Override
+        public void setParameters(int baudRate, int dataBits, int stopBits, int parity) throws IOException { }
+    }
+
+    static class DummyUsbRequest extends UsbRequest {
+        @Override
+        public boolean initialize(UsbDeviceConnection connection, UsbEndpoint endpoint) { return true; }
+
+        @Override
+        public void setClientData(Object data) { }
+
+        @Override
+        public boolean queue(ByteBuffer buffer, int length) { return true; }
+    }
+
+    @Test
+    public void readQueue() throws Exception {
+        UsbDeviceConnection usbDeviceConnection = mock(UsbDeviceConnection.class);
+        DummySerialDriver driver = new DummySerialDriver();
+        CommonUsbSerialPort port = (CommonUsbSerialPort) driver.getPorts().get(0);
+
+        // set before open
+        port.setReadQueue(0, 0);
+        port.setReadQueue(0, -1);
+        assertThrows(IllegalArgumentException.class, () -> port.setReadQueue(-1, 1));
+        assertThrows(IllegalArgumentException.class, () -> port.setReadQueue(1, 0));
+        port.setReadQueue(2, 256);
+        assertNull(port.mReadQueueRequests);
+
+        // change after open
+        port.open(usbDeviceConnection);
+        assertNotNull(port.mReadQueueRequests);
+        assertEquals(2, port.mReadQueueRequests.size());
+        assertThrows(IllegalStateException.class, () -> port.setReadQueue(1, 256));
+        port.setReadQueue(3, 256);
+        assertEquals(3, port.mReadQueueRequests.size());
+        assertThrows(IllegalStateException.class, () -> port.setReadQueue(3, 128));
+        assertThrows(IllegalStateException.class, () -> port.setReadQueue(3, 512));
+
+        // set after open
+        port.close();
+        port.setReadQueue(0, 0);
+        port.open(usbDeviceConnection);
+        assertNull(port.mReadQueueRequests);
+        port.setReadQueue(3, 256);
+
+        // retain over close
+        port.close();
+        assertNull(port.mReadQueueRequests);
+        port.open(usbDeviceConnection);
+        assertEquals(3, port.mReadQueueRequests.size());
+    }
+
+}
+