From 3eed929d3c77b9e20e3f4778d73e57ab67519cbf Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Alen=20Vre=C4=8Dko?=
<332217+avrecko@users.noreply.github.com>
Date: Sun, 11 Feb 2024 20:19:51 +0100
Subject: [PATCH] Now we no longer depend on the GC to free native memory on
Linux.
JNA is freeing 192 KiB of per NuProcess native buffers on GC Finalizer / Reference Queue processing. In cases with very low GC activity, this is essentially a memory leak, potentially causing native memory issues.
* out/err/in buffers are freed onExit,
* event needed for Epoll #registerProcess and ##queueWrite is allocated and freed in the method,
* tweak in reusing of IntByReference for duration of epoll/kqueue processor,
* updated JNA library to 5.13.0, to get #close on Memory object (added in 5.12.0).
---
pom.xml | 2 +-
.../internal/BaseEventProcessor.java | 3 +
.../nuprocess/internal/BasePosixProcess.java | 6 +-
.../com/zaxxer/nuprocess/internal/Util.java | 21 +++++
.../zaxxer/nuprocess/linux/EpollEvent.java | 6 ++
.../zaxxer/nuprocess/linux/LinuxProcess.java | 15 +---
.../zaxxer/nuprocess/linux/ProcessEpoll.java | 25 +++---
.../zaxxer/nuprocess/osx/ProcessKqueue.java | 11 ++-
.../com/zaxxer/nuprocess/ThreadLoadTest.java | 80 +++++++++++++++++++
9 files changed, 141 insertions(+), 28 deletions(-)
create mode 100644 src/main/java/com/zaxxer/nuprocess/internal/Util.java
create mode 100644 src/test/java/com/zaxxer/nuprocess/ThreadLoadTest.java
diff --git a/pom.xml b/pom.xml
index 51cee582..65a14e05 100644
--- a/pom.xml
+++ b/pom.xml
@@ -68,7 +68,7 @@
net.java.dev.jna
jna
- 5.11.0
+ 5.13.0
diff --git a/src/main/java/com/zaxxer/nuprocess/internal/BaseEventProcessor.java b/src/main/java/com/zaxxer/nuprocess/internal/BaseEventProcessor.java
index 33ff02bc..df1bdb5f 100644
--- a/src/main/java/com/zaxxer/nuprocess/internal/BaseEventProcessor.java
+++ b/src/main/java/com/zaxxer/nuprocess/internal/BaseEventProcessor.java
@@ -48,6 +48,8 @@ public abstract class BaseEventProcessor 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);
}
/**
diff --git a/src/main/java/com/zaxxer/nuprocess/internal/BasePosixProcess.java b/src/main/java/com/zaxxer/nuprocess/internal/BasePosixProcess.java
index 5b8f19d5..b310581a 100644
--- a/src/main/java/com/zaxxer/nuprocess/internal/BasePosixProcess.java
+++ b/src/main/java/com/zaxxer/nuprocess/internal/BasePosixProcess.java
@@ -323,8 +323,9 @@ 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;
@@ -332,7 +333,6 @@ public void onExit(int statusCode)
errBuffer = null;
inBuffer = null;
processHandler = null;
- Memory.purge();
}
}
diff --git a/src/main/java/com/zaxxer/nuprocess/internal/Util.java b/src/main/java/com/zaxxer/nuprocess/internal/Util.java
new file mode 100644
index 00000000..81e485f4
--- /dev/null
+++ b/src/main/java/com/zaxxer/nuprocess/internal/Util.java
@@ -0,0 +1,21 @@
+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();
+ }
+ }
+
+}
diff --git a/src/main/java/com/zaxxer/nuprocess/linux/EpollEvent.java b/src/main/java/com/zaxxer/nuprocess/linux/EpollEvent.java
index bbe58e77..b3a7168e 100644
--- a/src/main/java/com/zaxxer/nuprocess/linux/EpollEvent.java
+++ b/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
{
/*
diff --git a/src/main/java/com/zaxxer/nuprocess/linux/LinuxProcess.java b/src/main/java/com/zaxxer/nuprocess/linux/LinuxProcess.java
index ee2dae01..4a092d8b 100644
--- a/src/main/java/com/zaxxer/nuprocess/linux/LinuxProcess.java
+++ b/src/main/java/com/zaxxer/nuprocess/linux/LinuxProcess.java
@@ -24,6 +24,7 @@
import com.zaxxer.nuprocess.internal.LibC;
import java.io.IOException;
+import java.nio.ByteBuffer;
import java.nio.file.Path;
import java.util.List;
import java.util.logging.Level;
@@ -35,7 +36,6 @@
*/
public class LinuxProcess extends BasePosixProcess
{
- private final EpollEvent epollEvent;
static {
LibEpoll.sigignore(LibEpoll.SIGPIPE);
@@ -58,7 +58,6 @@ private enum LaunchMechanism {
LinuxProcess(NuProcessHandler processListener) {
super(processListener);
- epollEvent = new EpollEvent();
}
@Override
@@ -117,17 +116,6 @@ public void run(List 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 command, String[] environment, Path cwd) throws IOException
{
String[] cmdarray = command.toArray(new String[0]);
@@ -230,4 +218,5 @@ private static byte[] toEnvironmentBlock(String[] environment) {
return block;
}
+
}
diff --git a/src/main/java/com/zaxxer/nuprocess/linux/ProcessEpoll.java b/src/main/java/com/zaxxer/nuprocess/linux/ProcessEpoll.java
index 54c09fc0..048134d1 100644
--- a/src/main/java/com/zaxxer/nuprocess/linux/ProcessEpoll.java
+++ b/src/main/java/com/zaxxer/nuprocess/linux/ProcessEpoll.java
@@ -21,10 +21,10 @@
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;
+import com.zaxxer.nuprocess.internal.Util;
import static com.zaxxer.nuprocess.internal.LibC.WIFEXITED;
import static com.zaxxer.nuprocess.internal.LibC.WEXITSTATUS;
@@ -86,6 +86,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 +97,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 +114,7 @@ public void registerProcess(LinuxProcess process)
}
}
finally {
+ event.close();
if (stdinFd != Integer.MIN_VALUE) {
process.getStdin().release();
}
@@ -133,13 +134,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());
@@ -154,6 +156,9 @@ public void queueWrite(LinuxProcess process)
}
finally {
process.getStdin().release();
+ if (event != null) {
+ event.close();
+ }
}
}
@@ -167,6 +172,9 @@ public void run()
// the handler's onExit is called before LinuxProcess.run returns.
waitForDeadPool();
}
+
+ Util.close(exitCodePointer);
+ triggeredEvent.close();
}
@Override
@@ -183,6 +191,7 @@ public void closeStdin(LinuxProcess process)
}
}
+
@Override
public boolean process()
{
@@ -306,8 +315,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 +324,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 +334,10 @@ private void checkDeadPool()
return;
}
- IntByReference ret = new IntByReference();
Iterator 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 +348,7 @@ private void checkDeadPool()
continue;
}
- handleExit(process, ret.getValue());
+ handleExit(process, exitCodePointer.getValue());
}
}
diff --git a/src/main/java/com/zaxxer/nuprocess/osx/ProcessKqueue.java b/src/main/java/com/zaxxer/nuprocess/osx/ProcessKqueue.java
index 210bf233..1d8188df 100644
--- a/src/main/java/com/zaxxer/nuprocess/osx/ProcessKqueue.java
+++ b/src/main/java/com/zaxxer/nuprocess/osx/ProcessKqueue.java
@@ -28,11 +28,12 @@
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
+import com.sun.jna.Memory;
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.internal.Util;
import com.zaxxer.nuprocess.osx.LibKevent.Kevent;
import com.zaxxer.nuprocess.osx.LibKevent.TimeSpec;
@@ -186,6 +187,12 @@ public void queueWrite(OsxProcess process)
LibC.kill(JAVA_PID, LibC.SIGUSR2);
}
+ @Override
+ public void run() {
+ super.run();
+ Util.close(exitCodePointer);
+ }
+
@Override
public void closeStdin(OsxProcess process)
{
@@ -393,7 +400,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());
diff --git a/src/test/java/com/zaxxer/nuprocess/ThreadLoadTest.java b/src/test/java/com/zaxxer/nuprocess/ThreadLoadTest.java
new file mode 100644
index 00000000..356827d3
--- /dev/null
+++ b/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();
+ }
+}