8365066: RecordingStream and RemoteRecordingStream do not terminate when the associated Recording is stopped or closed externally

Author: chiroito

src/jdk.jfr/share/classes/jdk/jfr/consumer/RecordingStream.java

@@ -33,12 +33,15 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Consumer;
 
 import jdk.jfr.Configuration;
 import jdk.jfr.Event;
 import jdk.jfr.EventSettings;
 import jdk.jfr.EventType;
+import jdk.jfr.FlightRecorder;
+import jdk.jfr.FlightRecorderListener;
 import jdk.jfr.Recording;
 import jdk.jfr.RecordingState;
 import jdk.jfr.internal.PlatformRecording;
@@ -82,6 +85,11 @@ public void accept(Long endNanos) {
     private final EventDirectoryStream directoryStream;
     private long maxSize;
     private Duration maxAge;
+    private final ReentrantLock lock = new ReentrantLock();
+    private volatile boolean closed;
+    private volatile boolean stopped;
+    private LocalStoppedListener localStoppedListener;
+    private LocalClosedListener localClosedListener;
 
     /**
      * Creates an event stream for the current JVM (Java Virtual Machine).
@@ -325,9 +333,41 @@ public void onError(Consumer<Throwable> action) {
 
     @Override
     public void close() {
-        directoryStream.setChunkCompleteHandler(null);
-        recording.close();
-        directoryStream.close();
+        try {
+            lock.lock();
+            if (closed) {
+                return;
+            }
+            if (localStoppedListener != null) {
+                FlightRecorder.removeListener(localStoppedListener);
+            }
+            if (localClosedListener != null) {
+                FlightRecorder.removeListener(localClosedListener);
+            }
+            recording.close();
+            closeInternal();
+        } finally {
+            lock.unlock();
+        }
+    }
+
+    private void closeInternal() {
+        final boolean isHeldByCurrentThread = lock.isHeldByCurrentThread();
+        try {
+            if (!isHeldByCurrentThread) {
+                lock.lock();
+            }
+            if (closed) {
+                return;
+            }
+            directoryStream.setChunkCompleteHandler(null);
+            directoryStream.close();
+            closed = true;
+        } finally {
+            if (!isHeldByCurrentThread) {
+                lock.unlock();
+            }
+        }
     }
 
     @Override
@@ -339,6 +379,10 @@ public boolean remove(Object action) {
     public void start() {
         PlatformRecording pr = PrivateAccess.getInstance().getPlatformRecording(recording);
         long startNanos = pr.start();
+        this.localStoppedListener = new LocalStoppedListener(pr.getId(), this);
+        this.localClosedListener = new LocalClosedListener(pr.getId(), this);
+        FlightRecorder.addListener(localStoppedListener);
+        FlightRecorder.addListener(localClosedListener);
         updateOnCompleteHandler();
         directoryStream.start(startNanos);
     }
@@ -362,6 +406,10 @@ public void start() {
     public void startAsync() {
         PlatformRecording pr = PrivateAccess.getInstance().getPlatformRecording(recording);
         long startNanos = pr.start();
+        this.localStoppedListener = new LocalStoppedListener(pr.getId(), this);
+        this.localClosedListener = new LocalClosedListener(pr.getId(), this);
+        FlightRecorder.addListener(localStoppedListener);
+        FlightRecorder.addListener(localClosedListener);
         updateOnCompleteHandler();
         directoryStream.startAsync(startNanos);
     }
@@ -389,20 +437,47 @@ public void startAsync() {
      * @since 20
      */
     public boolean stop() {
-        boolean stopped = false;
         try {
-            try (StreamBarrier sb = directoryStream.activateStreamBarrier()) {
-                stopped = recording.stop();
-                directoryStream.setCloseOnComplete(false);
-                sb.setStreamEnd(recording.getStopTime().toEpochMilli());
+            lock.lock();
+            if (stopped) {
+                return true;
             }
+            if (localStoppedListener != null) {
+                FlightRecorder.removeListener(localStoppedListener);
+            }
+            recording.stop();
+            stopInternal(recording.getStopTime().toEpochMilli());
             directoryStream.awaitTermination();
-        } catch (InterruptedException | IOException e) {
+        } catch (InterruptedException e) {
             // OK, return
+        } finally {
+            lock.unlock();
         }
         return stopped;
     }
 
+    private void stopInternal(long stopTime) {
+        final boolean isHeldByCurrentThread = lock.isHeldByCurrentThread();
+        try {
+            if (!isHeldByCurrentThread) {
+                lock.lock();
+            }
+            if (stopped) {
+                return;
+            }
+            try (StreamBarrier sb = directoryStream.activateStreamBarrier()) {
+                directoryStream.setCloseOnComplete(false);
+                sb.setStreamEnd(stopTime);
+                stopped = true;
+            } catch (Exception e) {
+            }
+        } finally {
+            if (!isHeldByCurrentThread) {
+                lock.unlock();
+            }
+        }
+    }
+
     /**
      * Writes recording data to a file.
      * <p>
@@ -481,4 +556,46 @@ private void updateOnCompleteHandler() {
             directoryStream.setChunkCompleteHandler(new ChunkConsumer(recording));
         }
     }
+
+    static final class LocalStoppedListener implements FlightRecorderListener {
+
+        private final long recordingId;
+        private final RecordingStream recordingStream;
+
+        public LocalStoppedListener(long recordingId, RecordingStream recordingStream) {
+            this.recordingId = recordingId;
+            this.recordingStream = recordingStream;
+        }
+
+        @Override
+        public void recordingStateChanged(Recording recording) {
+            if (this.recordingId != recording.getId()) {
+                return;
+            }
+            if (recording.getState() == RecordingState.STOPPED) {
+                recordingStream.stopInternal(recording.getStopTime().toEpochMilli());
+            }
+        }
+    }
+
+    static final class LocalClosedListener implements FlightRecorderListener {
+
+        private final long recordingId;
+        private final RecordingStream recordingStream;
+
+        public LocalClosedListener(long recordingId, RecordingStream recordingStream) {
+            this.recordingId = recordingId;
+            this.recordingStream = recordingStream;
+        }
+
+        @Override
+        public void recordingStateChanged(Recording recording) {
+            if (this.recordingId != recording.getId()) {
+                return;
+            }
+            if (recording.getState() == RecordingState.CLOSED) {
+                recordingStream.closeInternal();
+            }
+        }
+    }
 }

src/jdk.jfr/share/classes/jdk/jfr/internal/consumer/EventDirectoryStream.java

@@ -35,7 +35,6 @@
 import java.util.function.Consumer;
 
 import jdk.jfr.Configuration;
-import jdk.jfr.RecordingState;
 import jdk.jfr.consumer.RecordedEvent;
 import jdk.jfr.internal.JVM;
 import jdk.jfr.internal.LogLevel;
@@ -181,13 +180,6 @@ protected void processRecursionSafe() throws IOException {
                     return;
                 }
 
-                if (isRecordingStream()) {
-                    if (recording.getState() == RecordingState.STOPPED && !barrier.used()) {
-                        logStreamEnd("recording stopped externally.");
-                        return;
-                    }
-                }
-
                 if (repositoryFiles.hasFixedPath() && currentParser.isFinalChunk()) {
                     logStreamEnd("JVM process exited/crashed, or repository migrated to an unknown location.");
                     return;