macOS: handle all messages on disconnect

java-does-usb/jextract/macos/gen_macos.sh

@@ -27,6 +27,12 @@ $JEXTRACT --output ../../src/main/java \
   --include-function CFRunLoopAddSource \
   --include-function CFRunLoopRemoveSource \
   --include-function CFRunLoopRun \
+  --include-function CFMessagePortCreateLocal \
+  --include-function CFMessagePortCreateRunLoopSource \
+  --include-function CFMessagePortCreateRemote \
+  --include-function CFMessagePortSendRequest \
+  --include-function CFDataCreate \
+  --include-function CFDataGetBytePtr \
   --include-function CFUUIDGetUUIDBytes \
   --include-constant kCFNumberSInt32Type \
   cf_helper.h

java-does-usb/src/main/java/net/codecrete/usb/macos/MacosAsyncTask.java

@@ -22,8 +22,9 @@
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.ReentrantLock;
 
-import static java.lang.foreign.ValueLayout.ADDRESS;
-import static java.lang.foreign.ValueLayout.JAVA_INT;
+import static java.lang.foreign.MemorySegment.NULL;
+import static java.lang.foreign.ValueLayout.*;
+
 
 /**
  * Background task for handling asynchronous transfers.
@@ -54,6 +55,7 @@ enum TaskState {
     private TaskState state = TaskState.NOT_STARTED;
     private MemorySegment asyncIoRunLoop;
     private MemorySegment completionUpcallStub;
+    private MemorySegment messagePort;
     private long lastTransferId;
     private final Map<Long, MacosTransfer> transfersById = new HashMap<>();
 
@@ -67,15 +69,9 @@ void addEventSource(MemorySegment source) {
             asyncIoLock.lock();
 
             if (state != TaskState.RUNNING) {
-                if (state == TaskState.NOT_STARTED) {
-                    startAsyncIOThread(source);
-                    waitForRunLoopReady();
-                    return;
-
-                } else {
-                    // special case: run loop is not ready yet but background process is already starting
-                    waitForRunLoopReady();
-                }
+                if (state == TaskState.NOT_STARTED)
+                    startAsyncIOThread();
+                waitForRunLoopReady();
             }
 
             CoreFoundation.CFRunLoopAddSource(asyncIoRunLoop, source, IOKit.kCFRunLoopDefaultMode());
@@ -92,34 +88,60 @@ private void waitForRunLoopReady() {
 
     /**
      * Removes an event source from this background task.
-     *
+     * <p>
+     * The event source is not immediately removed. Instead, it is posted to a message queue
+     * processed by the same background thread processing the completion callbacks. This ensures
+     * that the events from releasing interfaces and closing devices are processed.
+     * </p>
      * @param source event source
      */
     void removeEventSource(MemorySegment source) {
-        CoreFoundation.CFRunLoopRemoveSource(asyncIoRunLoop, source, IOKit.kCFRunLoopDefaultMode());
+        try (var arena = Arena.ofConfined()) {
+            var eventSourceRef = arena.allocate(JAVA_LONG, 1);
+            eventSourceRef.set(JAVA_LONG, 0, source.address());
+            var dataRef = CoreFoundation.CFDataCreate(NULL, eventSourceRef, eventSourceRef.byteSize());
+            CoreFoundation.CFMessagePortSendRequest(messagePort, 0, dataRef, 0, 0, NULL, NULL);
+            CoreFoundation.CFRelease(dataRef);
+        }
     }
 
     /**
      * Starts the background thread.
-     *
-     * @param firstSource first event source
      */
-    private void startAsyncIOThread(MemorySegment firstSource) {
+    @SuppressWarnings("java:S125")
+    private void startAsyncIOThread() {
+        MemorySegment messagePortSource;
+
         try {
             state = TaskState.STARTING;
+
+            // create descriptor for completion callback function
             var completionHandlerFuncDesc = FunctionDescriptor.ofVoid(ADDRESS, JAVA_INT, ADDRESS);
             var asyncIOCompletedMH = MethodHandles.lookup().findVirtual(MacosAsyncTask.class, "asyncIOCompleted",
                     MethodType.methodType(void.class, MemorySegment.class, int.class, MemorySegment.class));
 
             var methodHandle = asyncIOCompletedMH.bindTo(this);
-            completionUpcallStub = Linker.nativeLinker().upcallStub(methodHandle, completionHandlerFuncDesc,
-                    Arena.global());
+            completionUpcallStub = Linker.nativeLinker().upcallStub(methodHandle, completionHandlerFuncDesc, Arena.global());
+
+            // create descriptor for message port callback function
+            var messagePortCallbackFuncDec = FunctionDescriptor.of(ADDRESS, ADDRESS, JAVA_INT, ADDRESS, ADDRESS);
+            var messagePortCallbackMH = MethodHandles.lookup().findVirtual(MacosAsyncTask.class, "messagePortCallback",
+                    MethodType.methodType(MemorySegment.class, MemorySegment.class, int.class, MemorySegment.class, MemorySegment.class));
+            var messagePortCallbackHandle = messagePortCallbackMH.bindTo(this);
+            var messagePortCallbackStub = Linker.nativeLinker().upcallStub(messagePortCallbackHandle, messagePortCallbackFuncDec, Arena.global());
+
+            // create local and remote message ports
+            var pid = ProcessHandle.current().pid();
+            var portName = CoreFoundationHelper.createCFStringRef("net.codecrete.usb.macos.eventsource." + pid, Arena.global());
+            var localPort = CoreFoundation.CFMessagePortCreateLocal(NULL, portName, messagePortCallbackStub, NULL, NULL);
+            messagePortSource = CoreFoundation.CFMessagePortCreateRunLoopSource(NULL, localPort, 0);
+            messagePort = CoreFoundation.CFMessagePortCreateRemote(NULL, portName);
 
         } catch (IllegalAccessException | NoSuchMethodException e) {
             throw new UsbException("internal error (creating method handle)", e);
         }
 
-        var thread = new Thread(() -> asyncIOCompletionTask(firstSource), "USB async IO");
+        var thread = new Thread(() -> asyncIOCompletionTask(messagePortSource), "USB async IO");
         thread.setDaemon(true);
         thread.start();
     }
@@ -184,6 +206,20 @@ private void asyncIOCompleted(MemorySegment refcon, int result, MemorySegment ar
         transfer.completion().completed(transfer);
     }
 
+    /**
+     * Callback function called when a message is received on the message port.
+     * <p>
+     * All messages are related to removing event sources. They just contain the run loop source reference.
+     * </p>
+     */
+    @SuppressWarnings({"java:S1144", "unused"})
+    private MemorySegment messagePortCallback(MemorySegment local, int msgid, MemorySegment data, MemorySegment info) {
+        var runloopSourceRefPtr = CoreFoundation.CFDataGetBytePtr(data);
+        var runloopSourceRef = MemorySegment.ofAddress(runloopSourceRefPtr.get(JAVA_LONG_UNALIGNED, 0));
+        CoreFoundation.CFRunLoopRemoveSource(asyncIoRunLoop, runloopSourceRef, IOKit.kCFRunLoopDefaultMode());
+        return NULL;
+    }
+
     /**
      * Gets the native IO completion callback function for asynchronous transfers
      * to be handled by this background task.

java-does-usb/src/main/java/net/codecrete/usb/macos/MacosUsbDevice.java

@@ -143,22 +143,21 @@ public synchronized void close() {
             return;
 
         for (var interfaceInfo : claimedInterfaces) {
+            setClaimed(interfaceInfo.interfaceNumber, false);
             var source = IoKitUsb.GetInterfaceAsyncEventSource(interfaceInfo.iokitInterface());
-            if (source.address() != 0)
-                asyncTask.removeEventSource(source);
             IoKitUsb.USBInterfaceClose(interfaceInfo.iokitInterface);
             IoKitUsb.Release(interfaceInfo.iokitInterface);
-            setClaimed(interfaceInfo.interfaceNumber, false);
+            if (source.address() != 0)
+                asyncTask.removeEventSource(source);
         }
 
         claimedInterfaces = null;
         endpoints = null;
 
         var source = IoKitUsb.GetDeviceAsyncEventSource(device);
+        IoKitUsb.USBDeviceClose(device);
         if (source.address() != 0)
             asyncTask.removeEventSource(source);
-
-        IoKitUsb.USBDeviceClose(device);
     }
 
     synchronized void closeFully() {

java-does-usb/src/main/java/net/codecrete/usb/macos/gen/corefoundation/CFMessagePortCreateLocal$callout.java

@@ -0,0 +1,71 @@
+// Generated by jextract
+
+package net.codecrete.usb.macos.gen.corefoundation;
+
+import java.lang.invoke.*;
+import java.lang.foreign.*;
+import java.nio.ByteOrder;
+import java.util.*;
+import java.util.function.*;
+import java.util.stream.*;
+
+import static java.lang.foreign.ValueLayout.*;
+import static java.lang.foreign.MemoryLayout.PathElement.*;
+
+/**
+ * {@snippet lang=c :
+ * CFMessagePortCallBack callout
+ * }
+ */
+public class CFMessagePortCreateLocal$callout {
+
+    CFMessagePortCreateLocal$callout() {
+        // Should not be called directly
+    }
+
+    /**
+     * The function pointer signature, expressed as a functional interface
+     */
+    public interface Function {
+        MemorySegment apply(MemorySegment _x0, int _x1, MemorySegment _x2, MemorySegment _x3);
+    }
+
+    private static final FunctionDescriptor $DESC = FunctionDescriptor.of(
+        CoreFoundation.C_POINTER,
+        CoreFoundation.C_POINTER,
+        CoreFoundation.C_INT,
+        CoreFoundation.C_POINTER,
+        CoreFoundation.C_POINTER
+    );
+
+    /**
+     * The descriptor of this function pointer
+     */
+    public static FunctionDescriptor descriptor() {
+        return $DESC;
+    }
+
+    private static final MethodHandle UP$MH = CoreFoundation.upcallHandle(CFMessagePortCreateLocal$callout.Function.class, "apply", $DESC);
+
+    /**
+     * Allocates a new upcall stub, whose implementation is defined by {@code fi}.
+     * The lifetime of the returned segment is managed by {@code arena}
+     */
+    public static MemorySegment allocate(CFMessagePortCreateLocal$callout.Function fi, Arena arena) {
+        return Linker.nativeLinker().upcallStub(UP$MH.bindTo(fi), $DESC, arena);
+    }
+
+    private static final MethodHandle DOWN$MH = Linker.nativeLinker().downcallHandle($DESC);
+
+    /**
+     * Invoke the upcall stub {@code funcPtr}, with given parameters
+     */
+    public static MemorySegment invoke(MemorySegment funcPtr,MemorySegment _x0, int _x1, MemorySegment _x2, MemorySegment _x3) {
+        try {
+            return (MemorySegment) DOWN$MH.invokeExact(funcPtr, _x0, _x1, _x2, _x3);
+        } catch (Throwable ex$) {
+            throw new AssertionError("should not reach here", ex$);
+        }
+    }
+}
+