JVM hangs during shutdown (#287)

zhengyu123 · web-flow · commit 292144fd88aa · 2025-11-06T11:55:05.000-05:00
diff --git a/ddprof-lib/src/main/cpp/flightRecorder.cpp b/ddprof-lib/src/main/cpp/flightRecorder.cpp
@@ -40,7 +40,7 @@
 #include <vector>
 #include <unistd.h>
 
-static SpinLock _rec_lock(1);
+static SpinLock _rec_lock(0);
 
 static const char *const SETTING_RING[] = {NULL, "kernel", "user", "any"};
 static const char *const SETTING_CSTACK[] = {NULL, "no", "fp", "dwarf", "lbr"};
@@ -1460,6 +1460,7 @@ void Recording::addThread(int lock_index, int tid) {
 }
 
 Error FlightRecorder::start(Arguments &args, bool reset) {
+  ExclusiveLockGuard locker(&_rec_lock);
   const char *file = args.file();
   if (file == NULL || file[0] == 0) {
     _filename = "";
@@ -1473,7 +1474,6 @@ Error FlightRecorder::start(Arguments &args, bool reset) {
   }
 
   Error ret = newRecording(reset);
-  _rec_lock.unlock();
   return ret;
 }
 
@@ -1484,34 +1484,35 @@ Error FlightRecorder::newRecording(bool reset) {
     return Error("Could not open Flight Recorder output file");
   }
 
-  _rec = new Recording(fd, _args);
+  // Given some of reads are not protected by _rec_lock,
+  // we want to publish _rec with full fence, so that read
+  // side cannot see partially initialized recording.
+  Recording* tmp = new Recording(fd, _args);
+  __atomic_store_n(&_rec, tmp, __ATOMIC_SEQ_CST);
   return Error::OK;
 }
 
 void FlightRecorder::stop() {
+  ExclusiveLockGuard locker(&_rec_lock);
   if (_rec != NULL) {
-    _rec_lock.lock();
-
-    Recording *tmp = _rec;
+    volatile Recording *tmp = _rec;
     // NULL first, deallocate later
-    _rec = NULL;
+    __atomic_store_n(&_rec, nullptr, __ATOMIC_RELAXED);
     delete tmp;
   }
 }
 
 Error FlightRecorder::dump(const char *filename, const int length) {
+  ExclusiveLockGuard locker(&_rec_lock);
   if (_rec != NULL) {
-    _rec_lock.lock();
     if (_filename.length() != length ||
         strncmp(filename, _filename.c_str(), length) != 0) {
       // if the filename to dump the recording to is specified move the current
       // working file there
       int copy_fd = open(filename, O_CREAT | O_RDWR | O_TRUNC, 0644);
       _rec->switchChunk(copy_fd);
       close(copy_fd);
-      _rec_lock.unlock();
     } else {
-      _rec_lock.unlock();
       return Error(
           "Can not dump recording to itself. Provide a different file name!");
     }
@@ -1523,9 +1524,8 @@ Error FlightRecorder::dump(const char *filename, const int length) {
 }
 
 void FlightRecorder::flush() {
+  ExclusiveLockGuard locker(&_rec_lock);
   if (_rec != NULL) {
-    _rec_lock.lock();
-
     jvmtiEnv *jvmti = VM::jvmti();
     JNIEnv *env = VM::jni();
 
@@ -1542,7 +1542,6 @@ void FlightRecorder::flush() {
         jvmti->Deallocate((unsigned char *)classes[i]);
       }
     }
-    _rec_lock.unlock();
   }
 }
 
diff --git a/ddprof-lib/src/main/cpp/flightRecorder.h b/ddprof-lib/src/main/cpp/flightRecorder.h
@@ -291,7 +291,7 @@ class FlightRecorder {
 private:
   std::string _filename;
   Arguments _args;
-  Recording *_rec;
+  Recording* volatile _rec;
 
   Error newRecording(bool reset);
 
diff --git a/ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/stresstest/scenarios/counters/ThreadFilterBenchmark.java b/ddprof-stresstest/src/jmh/java/com/datadoghq/profiler/stresstest/scenarios/counters/ThreadFilterBenchmark.java
@@ -10,224 +10,79 @@
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.atomic.AtomicIntegerArray;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLongArray;
 
 @State(Scope.Benchmark)
 public class ThreadFilterBenchmark extends Configuration {
 
     @Param({"true", "false"}) // Parameterize the filter usage
     public boolean useThreadFilters;
 
-    private static final int NUM_THREADS = 15;
-    private ExecutorService executorService;
     private JavaProfiler profiler;
-    private AtomicBoolean running;
-    private CountDownLatch startLatch;
-    private CountDownLatch stopLatch;
-    private AtomicLong operationCount;
-    private long startTime;
-    private long stopTime;
-    private PrintWriter logWriter;
     private static final int ARRAY_SIZE = 1024; // Larger array to stress memory
-    private static final int[] sharedArray = new int[ARRAY_SIZE];
-    private static final AtomicIntegerArray atomicArray = new AtomicIntegerArray(ARRAY_SIZE);
+    private static final long[] sharedArray = new long[ARRAY_SIZE];
+    private static final AtomicLongArray atomicArray = new AtomicLongArray(ARRAY_SIZE);
     private static final int CACHE_LINE_SIZE = 64; // Typical cache line size
     private static final int STRIDE = CACHE_LINE_SIZE / Integer.BYTES; // Elements per cache line
-    private AtomicLong addThreadCount = new AtomicLong(0);
-    private AtomicLong removeThreadCount = new AtomicLong(0);
 
     @Setup(Level.Trial)
     public void setup() throws IOException {
-        System.out.println("Setting up benchmark...");
-        System.out.println("Thread filters enabled: " + useThreadFilters);
-        System.out.println("Creating thread pool with " + NUM_THREADS + " threads");
-        executorService = Executors.newFixedThreadPool(NUM_THREADS);
-        System.out.println("Getting profiler instance");
         profiler = JavaProfiler.getInstance();
-        
-        // Stop the profiler if it's already running
-        try {
-            profiler.stop();
-        } catch (IllegalStateException e) {
-            System.out.println("Profiler was not active at setup.");
-        }
-        
-        String config = "start,wall=10ms,filter=1,file=/tmp/thread_filter_profile.jfr";
-        System.out.println("Starting profiler with " + config);
-        profiler.execute(config);
-        System.out.println("Started profiler with output file");
-        
-        running = new AtomicBoolean(true);
-        operationCount = new AtomicLong(0);
-        startTime = System.currentTimeMillis();
-        stopTime = startTime + 30000; // Run for 30 seconds
-        System.out.println("Benchmark setup completed at " + startTime);
-        
-        try {
-            String logFile = "/tmp/thread_filter_benchmark.log";
-            System.out.println("Attempting to create log file at: " + logFile);
-            logWriter = new PrintWriter(new FileWriter(logFile));
-            logWriter.printf("Benchmark started at %d%n", startTime);
-            logWriter.flush();
-            System.out.println("Successfully created and wrote to log file");
-        } catch (IOException e) {
-            System.err.println("Failed to create log file: " + e.getMessage());
-            e.printStackTrace();
-            throw e;
-        }
-    }
-
-    @TearDown(Level.Trial)
-    public void tearDown() {
-        System.out.println("Tearing down benchmark...");
-        running.set(false);
-        
-        // Wait for all threads to finish with a timeout
-        try {
-            if (stopLatch != null) {
-                if (!stopLatch.await(30, TimeUnit.SECONDS)) {
-                    System.err.println("Warning: Some threads did not finish within timeout");
-                }
-            }
-        } catch (InterruptedException e) {
-            Thread.currentThread().interrupt();
-        }
-
-        // Shutdown executor with timeout
-        executorService.shutdown();
-        try {
-            if (!executorService.awaitTermination(30, TimeUnit.SECONDS)) {
-                executorService.shutdownNow();
-                if (!executorService.awaitTermination(30, TimeUnit.SECONDS)) {
-                    System.err.println("Warning: Executor did not terminate");
-                }
-            }
-        } catch (InterruptedException e) {
-            executorService.shutdownNow();
-            Thread.currentThread().interrupt();
-        }
-
-        long endTime = System.currentTimeMillis();
-        long totalOps = operationCount.get();
-        double durationSecs = (endTime - startTime) / 1000.0;
-        double opsPerSec = totalOps / durationSecs;
-        double addOpsPerSec = addThreadCount.get() / durationSecs;
-        double removeOpsPerSec = removeThreadCount.get() / durationSecs;
-        
-        String stats = String.format("Thread Filter Stats:%n" +
-                "Total operations: %,d%n" +
-                "Duration: %.2f seconds%n" +
-                "Operations/second: %,.0f%n" +
-                "Operations/second/thread: %,.0f%n" +
-                "AddThread operations/second: %,.0f%n" +
-                "RemoveThread operations/second: %,.0f%n",
-                totalOps, durationSecs, opsPerSec, opsPerSec / NUM_THREADS, addOpsPerSec, removeOpsPerSec);
-        
-        System.out.print(stats);
-        if (logWriter != null) {
-            try {
-                logWriter.print(stats);
-                logWriter.flush();
-                logWriter.close();
-                System.out.println("Successfully closed log file");
-            } catch (Exception e) {
-                System.err.println("Error closing log file: " + e.getMessage());
-                e.printStackTrace();
-            }
-        }
-    }
-
-    public void setUseThreadFilters(boolean useThreadFilters) {
-        this.useThreadFilters = useThreadFilters;
     }
 
     @Benchmark
     @BenchmarkMode(Mode.Throughput)
     @Fork(value = 1, warmups = 1)
     @Warmup(iterations = 1, time = 1)
     @Measurement(iterations = 1, time = 2)
-    @Threads(1)
+    @Threads(15)
     @OutputTimeUnit(TimeUnit.MILLISECONDS)
-    public long threadFilterStress() throws InterruptedException {
-        System.out.println("Starting benchmark iteration...");
-        startLatch = new CountDownLatch(NUM_THREADS);
-        stopLatch = new CountDownLatch(NUM_THREADS);
+    public void threadFilterStress() throws InterruptedException {
+       long threadId = Thread.currentThread().getId();
+       // Memory-intensive operations that would be sensitive to false sharing
+       for (int j = 0; j < ARRAY_SIZE; j += STRIDE) {
+           if (useThreadFilters) {
+               // Register thread at the start of each cache line operation
+               profiler.addThread();
+           }
 
-        // Start all worker threads[]
-        for (int i = 0; i < NUM_THREADS; i++) {
-            final int threadId = i;
-            executorService.submit(() -> {
-                try {
-                    startLatch.countDown();
-                    startLatch.await(30, TimeUnit.SECONDS);
-                    String startMsg = String.format("Thread %d started%n", threadId);
-                    System.out.print(startMsg);
-                    if (logWriter != null) {
-                        logWriter.print(startMsg);
-                        logWriter.flush();
-                    }
-                    
-                    while (running.get() && System.currentTimeMillis() < stopTime) {
-                        // Memory-intensive operations that would be sensitive to false sharing
-                        for (int j = 0; j < ARRAY_SIZE; j += STRIDE) {
-                            if (useThreadFilters) {
-                                // Register thread at the start of each cache line operation
-                                profiler.addThread();
-                                addThreadCount.incrementAndGet();
-                            }
-                            
-                            // Each thread writes to its own cache line
-                            int baseIndex = (threadId * STRIDE) % ARRAY_SIZE;
-                            for (int k = 0; k < STRIDE; k++) {
-                                int index = (baseIndex + k) % ARRAY_SIZE;
-                                // Write to shared array
-                                sharedArray[index] = threadId;
-                                // Read and modify
-                                int value = sharedArray[index] + 1;
-                                // Atomic operation
-                                atomicArray.set(index, value);
-                            }
-                            
-                            if (useThreadFilters) {
-                                // Remove thread after cache line operation
-                                profiler.removeThread();
-                                removeThreadCount.incrementAndGet();
-                            }
-                            operationCount.incrementAndGet();
-                        }
+           // Each thread writes to its own cache line
+           long baseIndex = (threadId * STRIDE) % ARRAY_SIZE;
+           for (int k = 0; k < STRIDE; k++) {
+               int index = (int)(baseIndex + k) % ARRAY_SIZE;
+               // Write to shared array
+               sharedArray[index] = threadId;
+               // Read and modify
+               long value = sharedArray[index] + 1;
+               // Atomic operation
+               atomicArray.set(index, value);
+           }
 
-                        // More memory operations with thread registration
-                        for (int j = 0; j < ARRAY_SIZE; j += STRIDE) {
-                            if (useThreadFilters) {
-                                // Register thread at the start of each cache line operation
-                                profiler.addThread();
-                                addThreadCount.incrementAndGet();
-                            }
-                            
-                            int baseIndex = (threadId * STRIDE) % ARRAY_SIZE;
-                            for (int k = 0; k < STRIDE; k++) {
-                                int index = (baseIndex + k) % ARRAY_SIZE;
-                                int value = atomicArray.get(index);
-                                sharedArray[index] = value * 2;
-                            }
-                            
-                            if (useThreadFilters) {
-                                // Remove thread after cache line operation
-                                profiler.removeThread();
-                                removeThreadCount.incrementAndGet();
-                            }
-                            operationCount.incrementAndGet();
-                        }
-                    }
-                } catch (InterruptedException e) {
-                    Thread.currentThread().interrupt();
-                } finally {
-                    stopLatch.countDown();
-                }
-            });
-        }
+           if (useThreadFilters) {
+               // Remove thread after cache line operation
+               profiler.removeThread();
+           }
+       }
+
+       // More memory operations with thread registration
+       for (int j = 0; j < ARRAY_SIZE; j += STRIDE) {
+            if (useThreadFilters) {
+                // Register thread at the start of each cache line operation
+                profiler.addThread();
+            }
 
-        stopLatch.await();
-        return operationCount.get();
+            long baseIndex = (threadId * STRIDE) % ARRAY_SIZE;
+            for (int k = 0; k < STRIDE; k++) {
+                int index = (int)(baseIndex + k) % ARRAY_SIZE;
+                long value = atomicArray.get(index);
+                sharedArray[index] = value * 2;
+            }
+
+            if (useThreadFilters) {
+                // Remove thread after cache line operation
+                profiler.removeThread();
+            }
+       }
     }
 }
