Fix #11 (race condition in Linux)

manuelbl · manuelbl · commit 547950d652b5 · 2023-05-06T17:29:35.000+02:00
diff --git a/java-does-usb/jextract/README.md b/java-does-usb/jextract/README.md
@@ -68,7 +68,7 @@ SymbolLookup loaderLookup = SymbolLookup.libraryLookup("libudev.so.1", MemorySes
 
 ## MacOS
 
-Most of the required native functions on macOS are part of a framework. Frameworks internally have a more complex file organization of header and binary files than appears from the outside. Thus they require a special logic to locate framework header files. *clang* supports it with the `-F`. *jextract* allows to specify the options via `compiler_flags.txt` file. Since the file must be in the local directory and since it does not apply to Linux and Windows, separate directories must be used for the operating systems.
+Most of the required native functions on macOS are part of a framework. Frameworks internally have a more complex file organization of header and binary files than appears from the outside. Thus, they require a special logic to locate framework header files. *clang* supports it with the `-F`. *jextract* allows to specify the options via `compiler_flags.txt` file. Since the file must be in the local directory and since it does not apply to Linux and Windows, separate directories must be used for the operating systems.
 
 The generated code has the same problem as the Linux code for *udev*. It must be manually changed to use `SymbolLookup.libraryLookup()` for the libraries `CoreFoundation.framework/CoreFoundation` and `IOKit.framework/IOKit` respectively.
 
@@ -102,12 +102,12 @@ The worst example is [`IOUSBInterfaceStruct100`](https://github.com/manuelbl/Jav
 
 The table below shows statictics for version 0.2.0 of the library:
 
-| Operating Systems | Manually Created | % | Generated | % | Total | % |
-| - | -:| -:| -:| -:| -:| -:|
-| Linux       |  24,140 |  1.96% |   182,774 | 14.84% |   206,914 |  16.80% |
-| macOS       |  57,788 |  4.69% |   666,037 | 54.08% |   723,825 |  58.77% |
-| Windows     |  46,085 |  3.74% |   201,000 | 16.32% |   247,085 |  20.06% |
-| Common      |  53,774 |  4.37% |    53,774 |  0.00% |           |   4.37% |
-| Grand Total | 181,787 | 14.76% | 1,049,811 | 85.24% | 1,231,598 | 100.00% |
+| Operating Systems | Manually Created |      % | Generated |      % |     Total |       % |
+|-------------------|-----------------:|-------:|----------:|-------:|----------:|--------:|
+| Linux             |           24,140 |  1.96% |   182,774 | 14.84% |   206,914 |  16.80% |
+| macOS             |           57,788 |  4.69% |   666,037 | 54.08% |   723,825 |  58.77% |
+| Windows           |           46,085 |  3.74% |   201,000 | 16.32% |   247,085 |  20.06% |
+| Common            |           53,774 |  4.37% |    53,774 |  0.00% |           |   4.37% |
+| Grand Total       |          181,787 | 14.76% | 1,049,811 | 85.24% | 1,231,598 | 100.00% |
 
 *Code Size (compiled), in bytes and percentage of total size*
diff --git a/java-does-usb/jextract/linux/gen_linux.sh b/java-does-usb/jextract/linux/gen_linux.sh
@@ -6,10 +6,10 @@ JEXTRACT=../../../../jextract/build/jextract/bin/jextract
 $JEXTRACT --source --output ../../src/main/java \
   --header-class-name errno \
   --target-package net.codecrete.usb.linux.gen.errno \
-  --include-constant ETIMEDOUT \
   --include-constant EPIPE \
   --include-constant EBADF \
   --include-constant EAGAIN \
+  --include-constant EINVAL \
   /usr/include/errno.h
 
 # string.h
diff --git a/java-does-usb/src/main/java/net/codecrete/usb/linux/LinuxAsyncTask.java b/java-does-usb/src/main/java/net/codecrete/usb/linux/LinuxAsyncTask.java
@@ -15,10 +15,8 @@
 
 import java.lang.foreign.Arena;
 import java.lang.foreign.MemorySegment;
-import java.lang.foreign.SegmentAllocator;
-import java.lang.foreign.SegmentScope;
 import java.util.ArrayList;
-import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 
@@ -58,15 +56,15 @@ static synchronized LinuxAsyncTask instance() {
         return singletonInstance;
     }
 
-    private final SegmentAllocator GLOBAL_ALLOCATOR = SegmentAllocator.nativeAllocator(SegmentScope.global());
+    private final Arena urbArena = Arena.openShared();
     /// available URBs
     private final List<MemorySegment> availableURBs = new ArrayList<>();
     /// map of URB addresses to transfer (for outstanding transfers)
-    private final Map<Long, LinuxTransfer> transfersByURB = new HashMap<>();
+    private final Map<MemorySegment, LinuxTransfer> transfersByURB = new LinkedHashMap<>();
     /// array of file descriptors using asynchronous completion
     private int[] asyncFds;
     /// file descriptor to notify async IO background thread about an update
-    private int asyncIOUpdateEventFd;
+    private int asyncIOWakeUpEventFd;
 
     /**
      * Background task for handling asynchronous IO completions.
@@ -103,7 +101,8 @@ private void asyncCompletionTask() {
                     pollfd.revents$set(asyncPolls, i, (short) 0);
                 }
 
-                pollfd.fd$set(asyncPolls, n, asyncIOUpdateEventFd);
+                // entry n is the wake-up event file descriptor
+                pollfd.fd$set(asyncPolls, n, asyncIOWakeUpEventFd);
                 pollfd.events$set(asyncPolls, n, (short) poll.POLLIN());
                 pollfd.revents$set(asyncPolls, n, (short) 0);
 
@@ -112,74 +111,79 @@ private void asyncCompletionTask() {
                 if (res < 0)
                     throwException("internal error (poll)");
 
-                // check for events
-                for (int i = 0; i < n + 1; i++) {
-                    var revent = pollfd.revents$get(asyncPolls, i);
-                    if (revent == 0)
-                        continue;
+                // acquire lock
+                synchronized (this) {
 
-                    if ((revent & poll.POLLERR()) != 0) {
-                        // most likely the device has been disconnected;
-                        // remove from polled FD list to prevent further problems
-                        int fd = pollfd.fd$get(asyncPolls, i);
-                        removeFdFromAsyncIOCompletion(fd);
+                    if ((pollfd.revents$get(asyncPolls, n) & poll.POLLIN()) != 0) {
+                        // wakeup to refresh list of file descriptors
+                        res = IO.eventfd_read(asyncIOWakeUpEventFd, eventfdValueHolder, errnoState);
+                        if (res < 0)
+                            throwLastError(errnoState, "internal error (eventfd_read)");
                         continue;
                     }
 
-                    if (i != n) {
+                    // check for events
+                    for (int i = 0; i < n + 1; i++) {
+                        var revent = pollfd.revents$get(asyncPolls, i);
+                        if (revent == 0)
+                            continue;
+
+                        if ((revent & poll.POLLERR()) != 0) {
+                            // most likely the device has been disconnected;
+                            // remove from polled FD list to prevent further problems
+                            int fd = pollfd.fd$get(asyncPolls, i);
+                            removeFdFromAsyncIOCompletion(fd);
+                            continue;
+                        }
+
                         // reap URB
                         int fd = pollfd.fd$get(asyncPolls, i);
-                        reapURB(fd, urbPointerHolder, errnoState);
-
-                    } else {
-                        // wakeup to refresh list of file descriptors
-                        res = IO.eventfd_read(asyncIOUpdateEventFd, eventfdValueHolder, errnoState);
-                        if (res < 0)
-                            throwLastError(errnoState, "internal error (eventfd_read)");
+                        reapURBs(fd, urbPointerHolder, errnoState);
                     }
                 }
             }
         }
     }
 
     /**
-     * Reap URB and handle the completed transfer.
+     * Reap all pending URBs and handle the completed transfers.
      *
      * @param fd               file descriptor
      * @param urbPointerHolder native memory to receive the URB pointer
      * @param errnoState       native memory to receive the errno
      */
-    private void reapURB(int fd, MemorySegment urbPointerHolder, MemorySegment errnoState) {
-        int res;
-        res = IO.ioctl(fd, REAPURBNDELAY, urbPointerHolder, errnoState);
-        if (res < 0) {
-            var err = Linux.getErrno(errnoState);
-            if (err == errno.EAGAIN())
-                return; // retry
-            if (err == errno.EBADF())
-                return; // ignore, device might have been closed
-            throwException(err, "internal error (reap URB)");
-        }
+    private void reapURBs(int fd, MemorySegment urbPointerHolder, MemorySegment errnoState) {
+        while (true) {
+            int res = IO.ioctl(fd, REAPURBNDELAY, urbPointerHolder, errnoState);
+            if (res < 0) {
+                var err = Linux.getErrno(errnoState);
+                if (err == errno.EAGAIN())
+                    return; // no more pending URBs
+                if (err == errno.EBADF())
+                    return; // ignore, device might have been closed
+                throwException(err, "internal error (reap URB)");
+            }
 
-        // call completion handler
-        var urbAddr = urbPointerHolder.get(JAVA_LONG, 0);
-        var transfer = getTransferResult(urbAddr);
-        transfer.completion.completed(transfer);
+            // call completion handler
+            var urb = urbPointerHolder.get(ADDRESS, 0);
+            var transfer = getTransferResult(urb);
+            transfer.completion.completed(transfer);
+        }
     }
 
     /**
      * Notifies background process about changed FD list
      */
     private void notifyAsyncIOTask() {
         // start background process if needed
-        if (asyncIOUpdateEventFd == 0) {
+        if (asyncIOWakeUpEventFd == 0) {
             startAsyncIOTask();
             return;
         }
 
         try (var arena = Arena.openConfined()) {
             var errnoState = arena.allocate(Linux.ERRNO_STATE.layout());
-            if (IO.eventfd_write(asyncIOUpdateEventFd, 1, errnoState) < 0)
+            if (IO.eventfd_write(asyncIOWakeUpEventFd, 1, errnoState) < 0)
                 throwLastError(errnoState, "internal error (eventfd_write)");
         }
     }
@@ -268,15 +272,15 @@ private void addURB(LinuxTransfer transfer) {
         if (size > 0) {
             urb = availableURBs.remove(size - 1);
         } else {
-            urb = usbdevfs_urb.allocate(GLOBAL_ALLOCATOR);
+            urb = usbdevfs_urb.allocate(urbArena);
         }
 
         transfer.urb = urb;
-        transfersByURB.put(urb.address(), transfer);
+        transfersByURB.put(urb, transfer);
     }
 
-    private synchronized LinuxTransfer getTransferResult(long urbAddr) {
-        var transfer = transfersByURB.remove(urbAddr);
+    private synchronized LinuxTransfer getTransferResult(MemorySegment urb) {
+        var transfer = transfersByURB.remove(urb);
         if (transfer == null)
             throwException("internal error (unknown URB)");
 
@@ -291,27 +295,31 @@ private synchronized LinuxTransfer getTransferResult(long urbAddr) {
     synchronized void abortTransfers(LinuxUSBDevice device, byte endpointAddress) {
         int fd = device.fileDescriptor();
         try (var arena = Arena.openConfined()) {
+
             var errnoState = arena.allocate(Linux.ERRNO_STATE.layout());
 
-            for (var urbAddress : transfersByURB.keySet()) {
-                var urb = usbdevfs_urb.ofAddress(MemorySegment.ofAddress(urbAddress), SegmentScope.global());
+            // iterate all URBs and discard the ones for the specified endpoint
+            for (var urb : transfersByURB.keySet()) {
                 if (fd != (int) usbdevfs_urb.usercontext$get(urb).address())
                     continue;
                 if (endpointAddress != usbdevfs_urb.endpoint$get(urb))
                     continue;
 
-                if (IO.ioctl(fd, DISCARDURB, urb, errnoState) < 0)
-                    throwLastError(errnoState, "failed to cancel transfer");
+                if (IO.ioctl(fd, DISCARDURB, urb, errnoState) < 0) {
+                    // ignore EINVAL; it occurs if the URB has completed at the same time
+                    if (Linux.getErrno(errnoState) != errno.EINVAL())
+                        throwLastError(errnoState, "failed to abort transfer");
+                }
             }
         }
     }
 
     private void startAsyncIOTask() {
         try (var arena = Arena.openConfined()) {
             var errnoState = arena.allocate(Linux.ERRNO_STATE.layout());
-            asyncIOUpdateEventFd = IO.eventfd(0, 0, errnoState);
-            if (asyncIOUpdateEventFd == -1) {
-                asyncIOUpdateEventFd = 0;
+            asyncIOWakeUpEventFd = IO.eventfd(0, 0, errnoState);
+            if (asyncIOWakeUpEventFd == -1) {
+                asyncIOWakeUpEventFd = 0;
                 throwLastError(errnoState, "internal error (eventfd)");
             }
         }
diff --git a/java-does-usb/src/main/java/net/codecrete/usb/linux/LinuxUSBDevice.java b/java-does-usb/src/main/java/net/codecrete/usb/linux/LinuxUSBDevice.java
@@ -111,7 +111,6 @@ public synchronized void claimInterface(int interfaceNumber) {
             throwException("Interface %d has already been claimed", interfaceNumber);
 
         try (var arena = Arena.openConfined()) {
-            var intfNumSegment = arena.allocate(JAVA_INT, interfaceNumber);
             var errnoState = arena.allocate(Linux.ERRNO_STATE.layout());
 
             // claim interface (possibly disconnecting kernel driver)
@@ -324,15 +323,15 @@ public synchronized void abortTransfers(USBDirection direction, int endpointNumb
     }
 
     @Override
-    public InputStream openInputStream(int endpointNumber, int bufferSize) {
+    public synchronized InputStream openInputStream(int endpointNumber, int bufferSize) {
         // check that endpoint number is valid
         getEndpoint(USBDirection.IN, endpointNumber, USBTransferType.BULK, null);
 
         return new LinuxEndpointInputStream(this, endpointNumber, bufferSize);
     }
 
     @Override
-    public OutputStream openOutputStream(int endpointNumber, int bufferSize) {
+    public synchronized OutputStream openOutputStream(int endpointNumber, int bufferSize) {
         // check that endpoint number is valid
         getEndpoint(USBDirection.OUT, endpointNumber, USBTransferType.BULK, null);
 
diff --git a/java-does-usb/src/main/java/net/codecrete/usb/linux/gen/errno/errno.java b/java-does-usb/src/main/java/net/codecrete/usb/linux/gen/errno/errno.java
@@ -31,19 +31,19 @@ public static int EAGAIN() {
     }
     /**
      * {@snippet :
-     * #define EPIPE 32
+     * #define EINVAL 22
      * }
      */
-    public static int EPIPE() {
-        return (int)32L;
+    public static int EINVAL() {
+        return (int)22L;
     }
     /**
      * {@snippet :
-     * #define ETIMEDOUT 110
+     * #define EPIPE 32
      * }
      */
-    public static int ETIMEDOUT() {
-        return (int)110L;
+    public static int EPIPE() {
+        return (int)32L;
     }
 }
 
diff --git a/java-does-usb/src/main/java/net/codecrete/usb/macos/MacosUSBDevice.java b/java-does-usb/src/main/java/net/codecrete/usb/macos/MacosUSBDevice.java
@@ -97,10 +97,10 @@ public synchronized void close() {
         if (!isOpen())
             return;
 
-        for (InterfaceInfo interfaceInfo : claimedInterfaces) {
-            IoKitUSB.USBInterfaceClose(interfaceInfo.iokitInterface());
-            IoKitUSB.Release(interfaceInfo.iokitInterface());
-            setClaimed(interfaceInfo.interfaceNumber, false);
+        for (InterfaceInfo(MemorySegment iokitInterface, int interfaceNumber) : claimedInterfaces) {
+            IoKitUSB.USBInterfaceClose(iokitInterface);
+            IoKitUSB.Release(iokitInterface);
+            setClaimed(interfaceNumber, false);
         }
 
         claimedInterfaces = null;
diff --git a/java-does-usb/src/test/java/net/codecrete/usb/sample/LogicAnalyzer.java b/java-does-usb/src/test/java/net/codecrete/usb/sample/LogicAnalyzer.java
@@ -118,7 +118,7 @@ void sampleData(int sampleRate, int duration, String filename) {
         this.filename = filename;
         prepareSampling(sampleRate, duration);
         var acquirer = CompletableFuture.runAsync(this::saveSamples);
-        sleep(20); // give thread time to start
+        sleep(50); // give thread time to start
         startAcquisition();
         var watchdog = CompletableFuture.runAsync(this::detectBufferOverrun);
         CompletableFuture.allOf(acquirer, watchdog).join();
@@ -232,11 +232,12 @@ void stopAcquisition() {
     }
 
     void detectBufferOverrun() {
+        sleep(10);
         // if the 'activityValue' hasn't changed within 20ms, the logic analyzer
         // has stopped sending data (likely due to a buffer overrun)
         long lastValue = 0;
         while (true) {
-            sleep(20);
+            sleep(5);
 
             if (stopped)
                 break;
