Don't depend on the GC to free native memory on Linux

pom.xml

@@ -68,7 +68,7 @@
         <dependency>
             <groupId>net.java.dev.jna</groupId>
             <artifactId>jna</artifactId>
-            <version>5.11.0</version>
+            <version>5.13.0</version>
         </dependency>
 
         <dependency>

src/main/java/com/zaxxer/nuprocess/internal/BaseEventProcessor.java

@@ -48,6 +48,8 @@ public abstract class BaseEventProcessor<T extends BasePosixProcess> implements
    private CyclicBarrier startBarrier;
    private AtomicBoolean isRunning;
 
+   protected final IntByReference exitCodePointer = new IntByReference();
+
    static {
       LINGER_TIME_MS = Math.max(1000, Integer.getInteger("com.zaxxer.nuprocess.lingerTimeMs", 2500));
 
@@ -128,6 +130,7 @@ public void shutdown()
          process.onExit(Integer.MAX_VALUE - 1);
          LibC.waitpid(process.getPid(), exitCode, LibC.WNOHANG);
       }
+      Util.close(exitCode);
    }
 
    /**

src/main/java/com/zaxxer/nuprocess/internal/BasePosixProcess.java

@@ -323,16 +323,16 @@ public void onExit(int statusCode)
       }
       finally {
          exitPending.countDown();
-         // Once the last reference to the buffer is gone, Java will finalize the buffer
-         // and release the native memory we allocated in initializeBuffers().
+         outBufferMemory.close();
+         errBufferMemory.close();
+         inBufferMemory.close();
          outBufferMemory = null;
          errBufferMemory = null;
          inBufferMemory = null;
          outBuffer = null;
          errBuffer = null;
          inBuffer = null;
          processHandler = null;
-         Memory.purge();
       }
    }
 

src/main/java/com/zaxxer/nuprocess/internal/Util.java

@@ -0,0 +1,22 @@
+package com.zaxxer.nuprocess.internal;
+
+import com.sun.jna.Memory;
+import com.sun.jna.Pointer;
+import com.sun.jna.PointerType;
+
+public class Util {
+
+    public static void close(PointerType pointerType) {
+        if (pointerType == null) {
+            return;
+        }
+        close(pointerType.getPointer());
+    }
+
+    public static void close(Pointer p) {
+        if (p instanceof Memory) {
+            ((Memory) p).close();
+        }
+    }
+
+}

src/main/java/com/zaxxer/nuprocess/linux/EpollEvent.java

@@ -20,6 +20,7 @@
 import java.util.List;
 
 import com.sun.jna.*;
+import com.zaxxer.nuprocess.internal.Util;
 
 class EpollEvent
 {
@@ -64,6 +65,11 @@ int size() {
       return size;
    }
 
+   public void close()
+   {
+      Util.close(getPointer());
+   }
+
    public static class EpollEventPrototype extends Structure
    {
       /*

src/main/java/com/zaxxer/nuprocess/linux/LinuxProcess.java

@@ -35,7 +35,6 @@
  */
 public class LinuxProcess extends BasePosixProcess
 {
-   private final EpollEvent epollEvent;
 
    static {
       LibEpoll.sigignore(LibEpoll.SIGPIPE);
@@ -58,7 +57,6 @@ private enum LaunchMechanism {
    LinuxProcess(NuProcessHandler processListener) {
       super(processListener);
 
-      epollEvent = new EpollEvent();
    }
 
    @Override
@@ -117,17 +115,6 @@ public void run(List<String> command, String[] environment, Path cwd)
       }
    }
 
-   /**
-    * An {@link EpollEvent} struct, which may be used when registering for events for this process. Each process has
-    * its own struct to avoid concurrency issues in {@link ProcessEpoll#registerProcess} when multiple processes are
-    * registered at once (e.g. multiple threads are all starting new processes concurrently).
-    *
-    * @return this process's {@link EpollEvent} struct
-    */
-   EpollEvent getEpollEvent() {
-      return epollEvent;
-   }
-
    private void prepareProcess(List<String> command, String[] environment, Path cwd) throws IOException
    {
       String[] cmdarray = command.toArray(new String[0]);

src/main/java/com/zaxxer/nuprocess/linux/ProcessEpoll.java

@@ -21,7 +21,6 @@
 import java.util.List;
 
 import com.sun.jna.Native;
-import com.sun.jna.ptr.IntByReference;
 import com.zaxxer.nuprocess.NuProcess;
 import com.zaxxer.nuprocess.internal.BaseEventProcessor;
 import com.zaxxer.nuprocess.internal.LibC;
@@ -86,6 +85,7 @@ public void registerProcess(LinuxProcess process)
       int stdinFd = Integer.MIN_VALUE;
       int stdoutFd = Integer.MIN_VALUE;
       int stderrFd = Integer.MIN_VALUE;
+      EpollEvent event = new EpollEvent();
       try {
          stdinFd = process.getStdin().acquire();
          stdoutFd = process.getStdout().acquire();
@@ -96,7 +96,6 @@ public void registerProcess(LinuxProcess process)
          fildesToProcessMap.put(stdoutFd, process);
          fildesToProcessMap.put(stderrFd, process);
 
-         EpollEvent event = process.getEpollEvent();
          event.setEvents(LibEpoll.EPOLLIN);
          event.setFileDescriptor(stdoutFd);
          int rc = LibEpoll.epoll_ctl(epoll, LibEpoll.EPOLL_CTL_ADD, stdoutFd, event.getPointer());
@@ -114,6 +113,7 @@ public void registerProcess(LinuxProcess process)
          }
       }
       finally {
+         event.close();
          if (stdinFd != Integer.MIN_VALUE) {
             process.getStdin().release();
          }
@@ -133,13 +133,14 @@ public void queueWrite(LinuxProcess process)
          return;
       }
 
+      EpollEvent event = null;
       try {
          int stdin = process.getStdin().acquire();
          if (stdin == -1) {
            return;
          }
 
-         EpollEvent event = process.getEpollEvent();
+         event = new EpollEvent();
          event.setEvents(LibEpoll.EPOLLOUT | LibEpoll.EPOLLONESHOT | LibEpoll.EPOLLRDHUP | LibEpoll.EPOLLHUP);
          event.setFileDescriptor(stdin);
          int rc = LibEpoll.epoll_ctl(epoll, LibEpoll.EPOLL_CTL_MOD, stdin, event.getPointer());
@@ -153,6 +154,9 @@ public void queueWrite(LinuxProcess process)
          }
       }
       finally {
+         if (event != null) {
+            event.close();
+         }
          process.getStdin().release();
       }
    }
@@ -306,8 +310,7 @@ private void cleanupProcess(LinuxProcess linuxProcess, int stdinFd, int stdoutFd
          return;
       }
 
-      IntByReference ret = new IntByReference();
-      int rc = LibC.waitpid(linuxProcess.getPid(), ret, LibC.WNOHANG);
+      int rc = LibC.waitpid(linuxProcess.getPid(), exitCodePointer, LibC.WNOHANG);
 
       if (rc == 0) {
          deadPool.add(linuxProcess);
@@ -316,7 +319,7 @@ else if (rc < 0) {
          linuxProcess.onExit((Native.getLastError() == LibC.ECHILD) ? Integer.MAX_VALUE : Integer.MIN_VALUE);
       }
       else {
-         handleExit(linuxProcess, ret.getValue());
+         handleExit(linuxProcess, exitCodePointer.getValue());
       }
    }
 
@@ -326,11 +329,10 @@ private void checkDeadPool()
          return;
       }
 
-      IntByReference ret = new IntByReference();
       Iterator<LinuxProcess> iterator = deadPool.iterator();
       while (iterator.hasNext()) {
          LinuxProcess process = iterator.next();
-         int rc = LibC.waitpid(process.getPid(), ret, LibC.WNOHANG);
+         int rc = LibC.waitpid(process.getPid(), exitCodePointer, LibC.WNOHANG);
          if (rc == 0) {
             continue;
          }
@@ -341,7 +343,7 @@ private void checkDeadPool()
             continue;
          }
 
-         handleExit(process, ret.getValue());
+         handleExit(process, exitCodePointer.getValue());
       }
    }
 

src/main/java/com/zaxxer/nuprocess/osx/ProcessKqueue.java

@@ -30,7 +30,6 @@
 
 import com.sun.jna.Native;
 import com.sun.jna.Pointer;
-import com.sun.jna.ptr.IntByReference;
 import com.zaxxer.nuprocess.internal.BaseEventProcessor;
 import com.zaxxer.nuprocess.internal.LibC;
 import com.zaxxer.nuprocess.osx.LibKevent.Kevent;
@@ -393,7 +392,7 @@ private void checkWaitWrites()
 
    private void cleanupProcess(OsxProcess osxProcess)
    {
-      LibC.waitpid(osxProcess.getPid(), new IntByReference(), LibC.WNOHANG);
+      LibC.waitpid(osxProcess.getPid(), exitCodePointer, LibC.WNOHANG);
 
       // If this is the last process in the map, this thread will cleanly shut down.
       pidToProcessMap.remove(osxProcess.getPid());

src/test/java/com/zaxxer/nuprocess/ThreadLoadTest.java

@@ -0,0 +1,80 @@
+package com.zaxxer.nuprocess;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.nio.ByteBuffer;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Yet another threaded test.
+ */
+public class ThreadLoadTest {
+
+    @Test
+    public void testStartLoad() throws InterruptedException {
+        int durationInMs = 15_000;
+        long cutOfTime = System.currentTimeMillis() + durationInMs;
+        int nrOfThreads = (Runtime.getRuntime().availableProcessors() * 2);
+        CountDownLatch latch = new CountDownLatch(nrOfThreads);
+        AtomicLong runCountCode0 = new AtomicLong();
+        AtomicLong runCountCodeNon0 = new AtomicLong();
+        AtomicLong problems = new AtomicLong();
+        for (int i = 0; i < nrOfThreads; i++) {
+            startNewThread(latch, cutOfTime, runCountCode0, runCountCodeNon0, problems);
+        }
+
+        Assert.assertTrue(latch.await(durationInMs + 1_000, TimeUnit.MILLISECONDS));
+        System.out.println("runCount 0 = " + runCountCode0.get());
+        System.out.println("runCount non-0 = " + runCountCodeNon0.get());
+        System.out.println("problems  = " + problems.get());
+    }
+
+    private void startNewThread(final CountDownLatch latch, final long cutOfTime, final AtomicLong zeroExit, final AtomicLong nonZeroExit, AtomicLong problems) {
+        new Thread(new Runnable() {
+            public void run() {
+                while (System.currentTimeMillis() < cutOfTime) {
+                    final int randomInt = ThreadLocalRandom.current().nextInt(10_000);
+                    final String text = "foo" + randomInt;
+                    long startTime = System.nanoTime();
+
+                    final NuProcess start = new NuProcessBuilder(new NuAbstractProcessHandler() {
+                        public void onPreStart(final NuProcess nuProcess) {
+                            super.onPreStart(nuProcess);
+                        }
+
+                        @Override
+                        public void onStart(NuProcess nuProcess) {
+                        }
+
+                        @Override
+                        public void onStdout(ByteBuffer buffer, boolean closed) {
+
+                        }
+
+                        public void onExit(final int statusCode) {
+                            if (statusCode == 0) {
+                                zeroExit.incrementAndGet();
+                            } else {
+                                nonZeroExit.incrementAndGet();
+                            }
+                        }
+                    }, "echo", text).start();
+                    try {
+                        start.waitFor(10, TimeUnit.DAYS);
+
+//                        System.out.println("Took " + ((System.nanoTime() - startTime) / 1000000));
+//                        start.wantWrite();
+                    } catch (InterruptedException e) {
+                        throw new RuntimeException(e);
+                    }
+                }
+
+                latch.countDown();
+            }
+        }).start();
+    }
+}