01. Refactored SerialInputOutputManager (#615)

Used separate threads for reading and writing, enhancing concurrency and performance.

Note: before was possible to start `SerialInputOutputManager` with `Executors.newSingleThreadExecutor().submit(ioManager)`. Now you have to use `ioManager.start()`

usbSerialForAndroid/build.gradle

@@ -34,9 +34,10 @@ dependencies {
     implementation "androidx.annotation:annotation:1.8.0"
     testImplementation 'junit:junit:4.13.2'
     testImplementation 'org.mockito:mockito-core:5.12.0'
+    androidTestImplementation 'androidx.appcompat:appcompat:1.6.1'
     androidTestImplementation 'androidx.test:core:1.5.0'
     androidTestImplementation 'androidx.test:runner:1.5.2'
-    androidTestImplementation 'commons-net:commons-net:3.10.0'
+    androidTestImplementation 'commons-net:commons-net:3.9.0' // later version fails on old Android devices with missing java.time.Duration class
     androidTestImplementation 'org.apache.commons:commons-lang3:3.14.0'
 }
 

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

@@ -1425,19 +1425,16 @@ public class DeviceTest {
         usb.setParameters(19200, 8, 1, UsbSerialPort.PARITY_NONE);
         telnet.setParameters(19200, 8, 1, UsbSerialPort.PARITY_NONE);
         usb.ioManager.setThreadPriority(Process.THREAD_PRIORITY_DEFAULT);
+        assertEquals(SerialInputOutputManager.State.STOPPED, usb.ioManager.getState());
         usb.ioManager.start();
         usb.waitForIoManagerStarted();
-        assertTrue("iomanager thread", usb.hasIoManagerThread());
+        assertEquals(SerialInputOutputManager.State.RUNNING, usb.ioManager.getState());
+        assertTrue("iomanager thread", usb.hasIoManagerThreads());
         try {
             usb.ioManager.start();
             fail("already running error expected");
         } catch (IllegalStateException ignored) {
         }
-        try {
-            usb.ioManager.run();
-            fail("already running error expected");
-        } catch (IllegalStateException ignored) {
-        }
         try {
             usb.ioManager.setThreadPriority(Process.THREAD_PRIORITY_LOWEST);
             fail("setThreadPriority IllegalStateException expected");
@@ -1462,23 +1459,12 @@ public class DeviceTest {
         telnet.write(new byte[1]); // now uses 8 byte buffer
         usb.read(3);
 
-        // writebuffer resize
+        // small writebuffer
         try {
             usb.ioManager.writeAsync(new byte[8192]);
             fail("expected BufferOverflowException");
         } catch (BufferOverflowException ignored) {}
 
-        usb.ioManager.setWriteBufferSize(16);
-        usb.ioManager.writeAsync("1234567890AB".getBytes());
-        try {
-            usb.ioManager.setWriteBufferSize(8);
-            fail("expected BufferOverflowException");
-        } catch (BufferOverflowException ignored) {}
-        usb.ioManager.setWriteBufferSize(24); // pending date copied to new buffer
-        telnet.write("a".getBytes());
-        assertThat(usb.read(1), equalTo("a".getBytes()));
-        assertThat(telnet.read(12), equalTo("1234567890AB".getBytes()));
-
         // small readbuffer
         usb.ioManager.setReadBufferSize(8);
         Log.d(TAG, "setReadBufferSize(8)");
@@ -1490,74 +1476,31 @@ public class DeviceTest {
         telnet.write("d".getBytes());
         assertThat(usb.read(1), equalTo("d".getBytes()));
 
+        SerialInputOutputManager ioManager = usb.ioManager;
+        assertEquals(SerialInputOutputManager.State.RUNNING, usb.ioManager.getState());
         usb.close();
-        for (int i = 0; i < 100 && usb.hasIoManagerThread(); i++) {
+        for (int i = 0; i < 100 && usb.hasIoManagerThreads(); i++) {
             Thread.sleep(1);
         }
-        assertFalse("iomanager thread", usb.hasIoManagerThread());
+        assertFalse("iomanager threads", usb.hasIoManagerThreads());
+        assertNull(usb.ioManager);
+        assertEquals(SerialInputOutputManager.State.STOPPED, ioManager.getState());
         SerialInputOutputManager.DEBUG = false;
-
-        // legacy start
-        usb.open(EnumSet.of(UsbWrapper.OpenCloseFlags.NO_IOMANAGER_START)); // 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);
-        Executors.newSingleThreadExecutor().submit(usb.ioManager);
-        usb.waitForIoManagerStarted();
-        try {
-            usb.ioManager.start();
-            fail("already running error expected");
-        } catch (IllegalStateException ignored) {
-        }
     }
 
     @Test
     public void writeAsync() throws Exception {
         byte[] data, buf = new byte[]{1};
 
-        // w/o timeout: write delayed until something is read
+        // write immediately, without waiting for read
         usb.open();
         usb.setParameters(19200, 8, 1, UsbSerialPort.PARITY_NONE);
         telnet.setParameters(19200, 8, 1, UsbSerialPort.PARITY_NONE);
         usb.ioManager.writeAsync(buf);
         usb.ioManager.writeAsync(buf);
-        data = telnet.read(1);
-        assertEquals(0, data.length);
-        telnet.write(buf);
-        data = usb.read(1);
-        assertEquals(1, data.length);
         data = telnet.read(2);
         assertEquals(2, data.length);
         usb.close();
-
-        // with timeout: write after timeout
-        usb.open(EnumSet.of(UsbWrapper.OpenCloseFlags.NO_IOMANAGER_START));
-        usb.ioManager.setReadTimeout(100);
-        usb.ioManager.start();
-        usb.setParameters(19200, 8, 1, UsbSerialPort.PARITY_NONE);
-        telnet.setParameters(19200, 8, 1, UsbSerialPort.PARITY_NONE);
-        usb.ioManager.writeAsync(buf);
-        usb.ioManager.writeAsync(buf);
-        data = telnet.read(2);
-        assertEquals(2, data.length);
-        usb.ioManager.setReadTimeout(200);
-
-        // with internal SerialTimeoutException
-        TestBuffer tbuf = new TestBuffer(usb.writeBufferSize + 2*usb.writePacketSize);
-        byte[] pbuf1 = new byte[tbuf.buf.length - 4];
-        byte[] pbuf2 = new byte[1];
-        System.arraycopy(tbuf.buf, 0,pbuf1, 0, pbuf1.length);
-        usb.ioManager.setWriteTimeout(20); // tbuf len >= 128, needs 133msec @ 9600 baud
-        usb.setParameters(9600, 8, 1, UsbSerialPort.PARITY_NONE);
-        telnet.setParameters(9600, 8, 1, UsbSerialPort.PARITY_NONE);
-        usb.ioManager.writeAsync(pbuf1);
-        for(int i = pbuf1.length; i < tbuf.buf.length; i++) {
-            Thread.sleep(20);
-            pbuf2[0] = tbuf.buf[i];
-            usb.ioManager.writeAsync(pbuf2);
-        }
-        while(!tbuf.testRead(telnet.read(-1)))
-            ;
     }
 
     @Test

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

@@ -33,6 +33,8 @@ import java.util.concurrent.Callable;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 
+import androidx.core.content.ContextCompat;
+
 public class UsbWrapper implements SerialInputOutputManager.Listener {
 
     public final static int     USB_READ_WAIT = 500;
@@ -92,7 +94,7 @@ public class UsbWrapper implements SerialInputOutputManager.Listener {
             intent.setPackage(context.getPackageName());
             PendingIntent permissionIntent = PendingIntent.getBroadcast(context, 0, intent, flags);
             IntentFilter filter = new IntentFilter("com.android.example.USB_PERMISSION");
-            context.registerReceiver(usbReceiver, filter, Context.RECEIVER_NOT_EXPORTED);
+            ContextCompat.registerReceiver(context, usbReceiver, filter, Context.RECEIVER_NOT_EXPORTED);
             usbManager.requestPermission(serialDriver.getDevice(), permissionIntent);
             for(int i=0; i<5000; i++) {
                 if(granted[0] != null) break;
@@ -256,12 +258,15 @@ public class UsbWrapper implements SerialInputOutputManager.Listener {
         throw new IOException("IoManager not started");
     }
 
-    public boolean hasIoManagerThread() {
+    public boolean hasIoManagerThreads() {
+        int c = 0;
         for (Thread thread : Thread.getAllStackTraces().keySet()) {
-            if (thread.getName().equals(SerialInputOutputManager.class.getSimpleName()))
+            if (thread.getName().equals(SerialInputOutputManager.class.getSimpleName() + "_read"))
-                return true;
+                c += 1;
+            if (thread.getName().equals(SerialInputOutputManager.class.getSimpleName() + "_write"))
+                c += 1;
         }
-        return false;
+        return c == 2;
     }
 
     // wait full time

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

@@ -9,21 +9,23 @@ package com.hoho.android.usbserial.util;
 import android.os.Process;
 import android.util.Log;
 
-import com.hoho.android.usbserial.driver.SerialTimeoutException;
 import com.hoho.android.usbserial.driver.UsbSerialPort;
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicReference;
 
 /**
- * Utility class which services a {@link UsbSerialPort} in its {@link #run()} method.
+ * Utility class which services a {@link UsbSerialPort} in its {@link #runWrite()} ()} and {@link #runRead()} ()} ()} methods.
  *
  * @author mike wakerly (opensource@hoho.com)
  */
-public class SerialInputOutputManager implements Runnable {
+public class SerialInputOutputManager {
 
     public enum State {
         STOPPED,
+        STARTING,
         RUNNING,
         STOPPING
     }
@@ -33,9 +35,6 @@ public class SerialInputOutputManager implements Runnable {
     private static final String TAG = SerialInputOutputManager.class.getSimpleName();
     private static final int BUFSIZ = 4096;
 
-    /**
-     * default read timeout is infinite, to avoid data loss with bulkTransfer API
-     */
     private int mReadTimeout = 0;
     private int mWriteTimeout = 0;
 
@@ -46,7 +45,8 @@ public class SerialInputOutputManager implements Runnable {
     private ByteBuffer mWriteBuffer = ByteBuffer.allocate(BUFSIZ);
 
     private int mThreadPriority = Process.THREAD_PRIORITY_URGENT_AUDIO;
-    private State mState = State.STOPPED; // Synchronized by 'this'
+    private final AtomicReference<State> mState = new AtomicReference<>(State.STOPPED);
+    private CountDownLatch mStartuplatch = new CountDownLatch(2);
     private Listener mListener; // Synchronized by 'this'
     private final UsbSerialPort mSerialPort;
 
@@ -57,7 +57,7 @@ public class SerialInputOutputManager implements Runnable {
         void onNewData(byte[] data);
 
         /**
-         * Called when {@link SerialInputOutputManager#run()} aborts due to an error.
+         * Called when {@link SerialInputOutputManager#runRead()} ()} or {@link SerialInputOutputManager#runWrite()} ()} ()} aborts due to an error.
          */
         void onRunError(Exception e);
     }
@@ -87,8 +87,9 @@ public class SerialInputOutputManager implements Runnable {
      * @param threadPriority  see {@link Process#setThreadPriority(int)}
      * */
     public void setThreadPriority(int threadPriority) {
-        if (mState != State.STOPPED)
+        if (!mState.compareAndSet(State.STOPPED, State.STOPPED)) {
             throw new IllegalStateException("threadPriority only configurable before SerialInputOutputManager is started");
+        }
         mThreadPriority = threadPriority;
     }
 
@@ -97,7 +98,7 @@ public class SerialInputOutputManager implements Runnable {
      */
     public void setReadTimeout(int timeout) {
         // when set if already running, read already blocks and the new value will not become effective now
-        if(mReadTimeout == 0 && timeout != 0 && mState != State.STOPPED)
+        if(mReadTimeout == 0 && timeout != 0 && mState.get() != State.STOPPED)
             throw new IllegalStateException("readTimeout only configurable before SerialInputOutputManager is started");
         mReadTimeout = timeout;
     }
@@ -145,91 +146,150 @@ public class SerialInputOutputManager implements Runnable {
     }
 
     /**
-     * when using writeAsync, it is recommended to use readTimeout != 0,
+     * write data asynchronously
-     * else the write will be delayed until read data is available
      */
     public void writeAsync(byte[] data) {
         synchronized (mWriteBufferLock) {
             mWriteBuffer.put(data);
+            mWriteBufferLock.notifyAll(); // Notify waiting threads
         }
     }
 
     /**
-     * start SerialInputOutputManager in separate thread
+     * start SerialInputOutputManager in separate threads
      */
     public void start() {
-        if(mState != State.STOPPED)
+        if(mState.compareAndSet(State.STOPPED, State.STARTING)) {
+            mStartuplatch = new CountDownLatch(2);
+            new Thread(this::runRead, this.getClass().getSimpleName() + "_read").start();
+            new Thread(this::runWrite, this.getClass().getSimpleName() + "_write").start();
+            try {
+                mStartuplatch.await();
+                mState.set(State.RUNNING);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            }
+        } else {
             throw new IllegalStateException("already started");
-        new Thread(this, this.getClass().getSimpleName()).start();
+        }
     }
 
     /**
-     * stop SerialInputOutputManager thread
+     * stop SerialInputOutputManager threads
      *
      * when using readTimeout == 0 (default), additionally use usbSerialPort.close() to
      * interrupt blocking read
      */
-    public synchronized void stop() {
+    public void stop() {
-        if (getState() == State.RUNNING) {
+        if(mState.compareAndSet(State.RUNNING, State.STOPPING)) {
+            synchronized (mWriteBufferLock) {
+                mWriteBufferLock.notifyAll(); // Wake up any waiting thread to check the stop condition
+            }
             Log.i(TAG, "Stop requested");
-            mState = State.STOPPING;
         }
     }
 
-    public synchronized State getState() {
+    public State getState() {
-        return mState;
+        return mState.get();
     }
 
     /**
-     * Continuously services the read and write buffers until {@link #stop()} is
+     * @return true if the thread is still running
-     * called, or until a driver exception is raised.
      */
-    @Override
+    private boolean isStillRunning() {
-    public void run() {
+        State state = mState.get();
-        synchronized (this) {
+        return ((state == State.RUNNING) || (state == State.STARTING))
-            if (getState() != State.STOPPED) {
+            && !Thread.currentThread().isInterrupted();
-                throw new IllegalStateException("Already running");
     }
-            mState = State.RUNNING;
+
-        }
+    /**
-        Log.i(TAG, "Running ...");
+     * Notify listener of an error
-        try {
+     *
-            if(mThreadPriority != Process.THREAD_PRIORITY_DEFAULT)
+     * @param e the exception
-                Process.setThreadPriority(mThreadPriority);
+     */
-            while (true) {
+    private void notifyErrorListener(Throwable e) {
-                if (getState() != State.RUNNING) {
+        Listener listener = getListener();
-                    Log.i(TAG, "Stopping mState=" + getState());
-                    break;
-                }
-                step();
-            }
-        } catch (Throwable e) {
-            if(mSerialPort.isOpen()) {
-                Log.w(TAG, "Run ending due to exception: " + e.getMessage(), e);
-            } else {
-                Log.i(TAG, "Socket closed");
-            }
-            final Listener listener = getListener();
         if (listener != null) {
             try {
-                    if (e instanceof Exception) {
+                listener.onRunError(e instanceof Exception ? (Exception) e : new Exception(e));
-                        listener.onRunError((Exception) e);
-                    } else {
-                        listener.onRunError(new Exception(e));
-                    }
             } catch (Throwable t) {
                 Log.w(TAG, "Exception in onRunError: " + t.getMessage(), t);
             }
         }
+    }
+
+    /**
+     * Set the thread priority
+     */
+    private void setThreadPriority() {
+        if (mThreadPriority != Process.THREAD_PRIORITY_DEFAULT) {
+            Process.setThreadPriority(mThreadPriority);
+        }
+    }
+
+    /**
+     * Continuously services the read buffers until {@link #stop()} is called, or until a driver exception is
+     * raised.
+     */
+    void runRead() {
+        Log.i(TAG, "runRead running ...");
+        try {
+            setThreadPriority();
+            mStartuplatch.countDown();
+            do {
+                stepRead();
+            } while (isStillRunning());
+            Log.i(TAG, "runRead: Stopping mState=" + getState());
+        } catch (Throwable e) {
+            if (Thread.currentThread().isInterrupted()) {
+                Log.w(TAG, "runRead: interrupted");
+            } else if(mSerialPort.isOpen()) {
+                Log.w(TAG, "runRead ending due to exception: " + e.getMessage(), e);
+            } else {
+                Log.i(TAG, "runRead: Socket closed");
+            }
+            notifyErrorListener(e);
         } finally {
-            synchronized (this) {
+            if (!mState.compareAndSet(State.RUNNING, State.STOPPING)) {
-                mState = State.STOPPED;
+                if (mState.compareAndSet(State.STOPPING, State.STOPPED)) {
-                Log.i(TAG, "Stopped");
+                    Log.i(TAG, "runRead: Stopped mState=" + getState());
+                }
             }
         }
     }
 
-    private void step() throws IOException {
+    /**
+     * Continuously services the write buffers until {@link #stop()} is called, or until a driver exception is
+     * raised.
+     */
+    void runWrite() {
+        Log.i(TAG, "runWrite running ...");
+        try {
+            setThreadPriority();
+            mStartuplatch.countDown();
+            do {
+                stepWrite();
+            } while (isStillRunning());
+            Log.i(TAG, "runWrite: Stopping mState=" + getState());
+        } catch (Throwable e) {
+            if (Thread.currentThread().isInterrupted()) {
+                Log.w(TAG, "runWrite: interrupted");
+            } else if(mSerialPort.isOpen()) {
+                Log.w(TAG, "runWrite ending due to exception: " + e.getMessage(), e);
+            } else {
+                Log.i(TAG, "runWrite: Socket closed");
+            }
+            notifyErrorListener(e);
+        } finally {
+            if (!mState.compareAndSet(State.RUNNING, State.STOPPING)) {
+                if (mState.compareAndSet(State.STOPPING, State.STOPPED)) {
+                    Log.i(TAG, "runWrite: Stopped mState=" + getState());
+                }
+            }
+        }
+    }
+
+    private void stepRead() throws IOException {
         // Handle incoming data.
         byte[] buffer;
         synchronized (mReadBufferLock) {
@@ -247,39 +307,28 @@ public class SerialInputOutputManager implements Runnable {
                 listener.onNewData(data);
             }
         }
+    }
 
+    private void stepWrite() throws IOException, InterruptedException {
         // Handle outgoing data.
-        buffer = null;
+        byte[] buffer = null;
         synchronized (mWriteBufferLock) {
-            len = mWriteBuffer.position();
+            int len = mWriteBuffer.position();
             if (len > 0) {
                 buffer = new byte[len];
                 mWriteBuffer.rewind();
                 mWriteBuffer.get(buffer, 0, len);
                 mWriteBuffer.clear();
+                mWriteBufferLock.notifyAll(); // Notify writeAsync that there is space in the buffer
+            } else {
+                mWriteBufferLock.wait();
             }
         }
         if (buffer != null) {
             if (DEBUG) {
-                Log.d(TAG, "Writing data len=" + len);
+                Log.d(TAG, "Writing data len=" + buffer.length);
             }
-            try {
             mSerialPort.write(buffer, mWriteTimeout);
-            } catch (SerialTimeoutException ex) {
-                synchronized (mWriteBufferLock) {
-                    byte[] buffer2 = null;
-                    int len2 = mWriteBuffer.position();
-                    if (len2 > 0) {
-                        buffer2 = new byte[len2];
-                        mWriteBuffer.rewind();
-                        mWriteBuffer.get(buffer2, 0, len2);
-                        mWriteBuffer.clear();
-                    }
-                    mWriteBuffer.put(buffer, ex.bytesTransferred, buffer.length - ex.bytesTransferred);
-                    if (buffer2 != null)
-                        mWriteBuffer.put(buffer2);
-                }
-            }
         }
     }
 

usbSerialForAndroid/src/test/java/com/hoho/android/usbserial/util/SerialInputOutputManagerTest.java

@@ -34,18 +34,19 @@ public class SerialInputOutputManagerTest {
         CommonUsbSerialPort port = mock(CommonUsbSerialPort.class);
         when(port.getReadEndpoint()).thenReturn(readEndpoint);
         when(port.read(new byte[16], 0)).thenReturn(1);
+        when(port.isOpen()).thenReturn(true);
         SerialInputOutputManager manager = new SerialInputOutputManager(port);
         manager.setThreadPriority(Process.THREAD_PRIORITY_DEFAULT);
 
         ExceptionListener exceptionListener = new ExceptionListener();
         manager.setListener(exceptionListener);
-        manager.run();
+        manager.runRead();
         assertEquals(RuntimeException.class, exceptionListener.e.getClass());
         assertEquals("exception1", exceptionListener.e.getMessage());
 
         ErrorListener errorListener = new ErrorListener();
         manager.setListener(errorListener);
-        manager.run();
+        manager.runRead();
         assertEquals(Exception.class, errorListener.e.getClass());
         assertEquals("java.lang.UnknownError: error1", errorListener.e.getMessage());
         assertEquals(UnknownError.class, errorListener.e.getCause().getClass());