Refactored SerialInputOutputManager

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

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

@@ -1434,7 +1434,7 @@ public class DeviceTest {
         } catch (IllegalStateException ignored) {
         }
         try {
-            usb.ioManager.run();
+            usb.ioManager.runRead();
             fail("already running error expected");
         } catch (IllegalStateException ignored) {
         }
@@ -1502,7 +1502,7 @@ 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);
-        Executors.newSingleThreadExecutor().submit(usb.ioManager);
+        usb.ioManager.start();
         usb.waitForIoManagerStarted();
         try {
             usb.ioManager.start();

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,9 @@ 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 CountDownLatch mShutdownlatch = new CountDownLatch(2);
     private Listener mListener; // Synchronized by 'this'
     private final UsbSerialPort mSerialPort;
 
@@ -57,7 +58,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,33 +88,12 @@ 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;
     }
 
-    /**
-     * read/write timeout
-     */
-    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)
-            throw new IllegalStateException("readTimeout only configurable before SerialInputOutputManager is started");
-        mReadTimeout = timeout;
-    }
-
-    public int getReadTimeout() {
-        return mReadTimeout;
-    }
-
-    public void setWriteTimeout(int timeout) {
-        mWriteTimeout = timeout;
-    }
-
-    public int getWriteTimeout() {
-        return mWriteTimeout;
-    }
-
     /**
      * read/write buffer size
      */
@@ -129,6 +109,28 @@ public class SerialInputOutputManager implements Runnable {
         return mReadBuffer.capacity();
     }
 
+    /**
+     * read/write timeout
+     */
+    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.get() != State.STOPPED)
+            throw new IllegalStateException("readTimeout only configurable before SerialInputOutputManager is started");
+        mReadTimeout = timeout;
+    }
+
+    public int getReadTimeout() {
+        return mReadTimeout;
+    }
+
+    public void setWriteTimeout(int timeout) {
+        mWriteTimeout = timeout;
+    }
+
+    public int getWriteTimeout() {
+        return mWriteTimeout;
+    }
+
     public void setWriteBufferSize(int bufferSize) {
         if(getWriteBufferSize() == bufferSize)
             return;
@@ -155,81 +157,137 @@ public class SerialInputOutputManager implements Runnable {
     }
 
     /**
-     * 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);
+            mShutdownlatch = 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)) {
             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) {
+            && (mShutdownlatch.getCount() == 2)
-                throw new IllegalStateException("Already running");
+            && !Thread.currentThread().isInterrupted();
     }
-            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);
             }
         }
-        } finally {
-            synchronized (this) {
-                mState = State.STOPPED;
-                Log.i(TAG, "Stopped");
     }
+
+    /**
+     * Set the thread priority
+     */
+    private void setThreadPriority() {
+        if (mThreadPriority != Process.THREAD_PRIORITY_DEFAULT) {
+            Process.setThreadPriority(mThreadPriority);
         }
     }
 
-    private void step() throws IOException {
+    /**
+     * Continuously services the read buffers until {@link #stop()} is called, or until a driver exception is
+     * raised.
+     */
+    public 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, "Thread interrupted, stopping runRead.");
+            } else {
+                Log.w(TAG, "runRead ending due to exception: " + e.getMessage(), e);
+                notifyErrorListener(e);
+            }
+        } finally {
+            if (!mState.compareAndSet(State.RUNNING, State.STOPPING)) {
+                if (mState.compareAndSet(State.STOPPING, State.STOPPED)) {
+                    Log.i(TAG, "runRead: Stopped mState=" + getState());
+                }
+            }
+            mShutdownlatch.countDown();
+        }
+    }
+
+    /**
+     * Continuously services the write buffers until {@link #stop()} is called, or until a driver exception is
+     * raised.
+     */
+    public 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, "Thread interrupted, stopping runWrite.");
+            } else {
+                Log.w(TAG, "runWrite ending due to exception: " + e.getMessage(), e);
+                notifyErrorListener(e);
+            }
+        } finally {
+            if (!mState.compareAndSet(State.RUNNING, State.STOPPING)) {
+                if (mState.compareAndSet(State.STOPPING, State.STOPPED)) {
+                    Log.i(TAG, "runWrite: Stopped mState=" + getState());
+                }
+            }
+            mShutdownlatch.countDown();
+        }
+    }
+
+    private void stepRead() throws IOException {
         // Handle incoming data.
         byte[] buffer;
         synchronized (mReadBufferLock) {
@@ -247,11 +305,13 @@ public class SerialInputOutputManager implements Runnable {
                 listener.onNewData(data);
             }
         }
+    }
 
+    private void stepWrite() throws IOException {
         // 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();
@@ -261,25 +321,9 @@ public class SerialInputOutputManager implements Runnable {
         }
         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

@@ -39,13 +39,13 @@ public class SerialInputOutputManagerTest {
 
         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());