[Backport] 8233700: EventStream not closed

Summary:

Test Plan: jdk/jfr

Reviewed-by: yuleil

Issue: dragonwell-project/dragonwell8#112

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

@@ -139,7 +139,7 @@ public interface EventStream extends AutoCloseable {
      */
     public static EventStream openRepository() throws IOException {
         Utils.checkAccessFlightRecorder();
-        return new EventDirectoryStream(AccessController.getContext(), null, SecuritySupport.PRIVILIGED, false);
+        return new EventDirectoryStream(AccessController.getContext(), null, SecuritySupport.PRIVILIGED, null);
     }
 
     /**
@@ -162,7 +162,7 @@ public static EventStream openRepository() throws IOException {
     public static EventStream openRepository(Path directory) throws IOException {
         Objects.nonNull(directory);
         AccessControlContext acc = AccessController.getContext();
-        return new EventDirectoryStream(acc, directory, FileAccess.UNPRIVILIGED, false);
+        return new EventDirectoryStream(acc, directory, FileAccess.UNPRIVILIGED, null);
     }
 
     /**

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

@@ -88,7 +88,8 @@ public RecordingStream() {
         this.recording = new Recording();
         this.recording.setFlushInterval(Duration.ofMillis(1000));
         try {
-            this.directoryStream = new EventDirectoryStream(acc, null, SecuritySupport.PRIVILIGED, true);
+            PlatformRecording pr = PrivateAccess.getInstance().getPlatformRecording(recording);
+            this.directoryStream = new EventDirectoryStream(acc, null, SecuritySupport.PRIVILIGED, pr);
         } catch (IOException ioe) {
             this.recording.close();
             throw new IllegalStateException(ioe.getMessage());

src/share/classes/jdk/jfr/internal/JVM.java

@@ -42,7 +42,6 @@ public final class JVM {
 
     static final long RESERVED_CLASS_ID_LIMIT = 400;
 
-    private volatile boolean recording;
     private volatile boolean nativeOK;
 
     private static native void registerNatives();
@@ -69,13 +68,35 @@ public static JVM getJVM() {
     private JVM() {
     }
 
+    /**
+     * Marks current chunk as final
+     * <p>
+     * This allows streaming clients to read the chunk header and
+     * close the stream when no more data will be written into
+     * the current repository.
+     */
+    public native void markChunkFinal();
+
     /**
      * Begin recording events
      *
      * Requires that JFR has been started with {@link #createNativeJFR()}
      */
     public native void beginRecording();
 
+    /**
+     * Return true if the JVM is recording
+     */
+    public native boolean isRecording();
+
+    /**
+     * End recording events, which includes flushing data in thread buffers
+     *
+     * Requires that JFR has been started with {@link #createNativeJFR()}
+     *
+     */
+    public native void endRecording();
+
     /**
      * Return ticks
      *
@@ -98,13 +119,7 @@ private JVM() {
      */
     public native boolean emitEvent(long eventTypeId, long timestamp, long when);
 
-    /**
-     * End recording events, which includes flushing data in thread buffers
-     *
-     * Requires that JFR has been started with {@link #createNativeJFR()}
-     *
-     */
-    public native void endRecording();
+
 
     /**
      * Return a list of all classes deriving from {@link Event}
@@ -369,20 +384,6 @@ private JVM() {
      */
     public native void storeMetadataDescriptor(byte[] bytes);
 
-    public void endRecording_() {
-        endRecording();
-        recording = false;
-    }
-
-    public void beginRecording_() {
-        beginRecording();
-        recording = true;
-    }
-
-    public boolean isRecording() {
-        return recording;
-    }
-
     /**
      * If the JVM supports JVM TI and retransformation has not been disabled this
      * method will return true. This flag can not change during the lifetime of
@@ -573,4 +574,5 @@ public boolean hasNativeJFR() {
      *@return start time of the recording in nanos, -1 in case of in-memory
      */
     public native long getChunkStartNanos();
+
 }

src/share/classes/jdk/jfr/internal/PlatformRecorder.java

@@ -31,6 +31,7 @@
 import static jdk.jfr.internal.LogTag.JFR;
 import static jdk.jfr.internal.LogTag.JFR_SYSTEM;
 
+import java.io.IOException;
 import java.security.AccessControlContext;
 import java.security.AccessController;
 import java.time.Duration;
@@ -54,6 +55,7 @@
 import jdk.jfr.RecordingState;
 import jdk.jfr.events.ActiveRecordingEvent;
 import jdk.jfr.events.ActiveSettingEvent;
+import jdk.jfr.internal.SecuritySupport.SafePath;
 import jdk.jfr.internal.SecuritySupport.SecureRecorderListener;
 import jdk.jfr.internal.instrument.JDKEvents;
 
@@ -71,6 +73,7 @@ public final class PlatformRecorder {
 
     private long recordingCounter = 0;
     private RepositoryChunk currentChunk;
+    private boolean inShutdown;
 
     public PlatformRecorder() throws Exception {
         repository = Repository.getRepository();
@@ -178,6 +181,10 @@ public static void notifyRecorderInitialized(FlightRecorder recorder) {
         }
     }
 
+    synchronized void setInShutDown() {
+        this.inShutdown = true;
+    }
+
     // called by shutdown hook
     synchronized void destroy() {
         try {
@@ -200,7 +207,7 @@ synchronized void destroy() {
 
         if (jvm.hasNativeJFR()) {
             if (jvm.isRecording()) {
-                jvm.endRecording_();
+                jvm.endRecording();
             }
             jvm.destroyNativeJFR();
         }
@@ -238,7 +245,7 @@ synchronized long start(PlatformRecording recording) {
                 MetadataRepository.getInstance().setOutput(null);
             }
             currentChunk = newChunk;
-            jvm.beginRecording_();
+            jvm.beginRecording();
             startNanos = jvm.getChunkStartNanos();
             recording.setState(RecordingState.RUNNING);
             updateSettings();
@@ -291,11 +298,15 @@ synchronized void stop(PlatformRecording recording) {
             }
         }
         OldObjectSample.emit(recording);
+        recording.setFinalStartnanos(jvm.getChunkStartNanos());
 
         if (endPhysical) {
             RequestEngine.doChunkEnd();
             if (recording.isToDisk()) {
                 if (currentChunk != null) {
+                    if (inShutdown) {
+                        jvm.markChunkFinal();
+                    }
                     MetadataRepository.getInstance().setOutput(null);
                     finishChunk(currentChunk, now, null);
                     currentChunk = null;
@@ -304,7 +315,7 @@ synchronized void stop(PlatformRecording recording) {
                 // last memory
                 dumpMemoryToDestination(recording);
             }
-            jvm.endRecording_();
+            jvm.endRecording();
             disableEvents();
         } else {
             RepositoryChunk newChunk = null;
@@ -329,7 +340,6 @@ synchronized void stop(PlatformRecording recording) {
         } else {
             RequestEngine.setFlushInterval(Long.MAX_VALUE);
         }
-
         recording.setState(RecordingState.STOPPED);
     }
 
@@ -359,17 +369,7 @@ void updateSettingsButIgnoreRecording(PlatformRecording ignoreMe) {
         MetadataRepository.getInstance().setSettings(list);
     }
 
-    public synchronized void rotateIfRecordingToDisk() {
-        boolean disk = false;
-        for (PlatformRecording s : getRecordings()) {
-            if (RecordingState.RUNNING == s.getState() && s.isToDisk()) {
-                disk = true;
-            }
-        }
-        if (disk) {
-            rotateDisk();
-        }
-    }
+
 
     synchronized void rotateDisk() {
         Instant now = Instant.now();
@@ -589,4 +589,20 @@ private void fillWithDiskChunks(PlatformRecording target) {
     public boolean isEnabled(String eventName) {
         return MetadataRepository.getInstance().isEnabled(eventName);
     }
+
+    public synchronized void migrate(SafePath repo) throws IOException {
+        // Must set repository while holding recorder lock so
+        // the final chunk in repository gets marked correctly
+        Repository.getRepository().setBasePath(repo);
+        boolean disk = false;
+        for (PlatformRecording s : getRecordings()) {
+            if (RecordingState.RUNNING == s.getState() && s.isToDisk()) {
+                disk = true;
+            }
+        }
+        if (disk) {
+            jvm.markChunkFinal();
+            rotateDisk();
+        }
+    }
 }

src/share/classes/jdk/jfr/internal/PlatformRecording.java

@@ -85,6 +85,7 @@ public final class PlatformRecording implements AutoCloseable {
     private AccessControlContext noDestinationDumpOnExitAccessControlContext;
     private boolean shuoldWriteActiveRecordingEvent = true;
     private Duration flushInterval = Duration.ofSeconds(1);
+    private long finalStartChunkNanos = Long.MIN_VALUE;
 
     PlatformRecording(PlatformRecorder recorder, long id) {
         // Typically the access control context is taken
@@ -807,4 +808,12 @@ public long getStreamIntervalMillis() {
             return Long.MAX_VALUE;
         }
     }
+
+    public long getFinalChunkStartNanos() {
+        return finalStartChunkNanos;
+    }
+
+    public void setFinalStartnanos(long chunkStartNanos) {
+       this.finalStartChunkNanos = chunkStartNanos;
+    }
 }

src/share/classes/jdk/jfr/internal/Repository.java

@@ -81,6 +81,7 @@ synchronized RepositoryChunk newChunk(Instant timestamp) {
             if (!SecuritySupport.existDirectory(repository)) {
                 this.repository = createRepository(baseLocation);
                 jvm.setRepositoryLocation(repository.toString());
+                SecuritySupport.setProperty(JFR_REPOSITORY_LOCATION_PROPERTY, repository.toString());
                 cleanupDirectories.add(repository);
             }
             return new RepositoryChunk(repository, timestamp);
@@ -111,9 +112,7 @@ private static SafePath createRepository(SafePath basePath) throws IOException {
         if (i == MAX_REPO_CREATION_RETRIES) {
             throw new IOException("Unable to create JFR repository directory using base location (" + basePath + ")");
         }
-        SafePath canonicalRepositoryPath = SecuritySupport.toRealPath(f);
-        SecuritySupport.setProperty(JFR_REPOSITORY_LOCATION_PROPERTY, canonicalRepositoryPath.toString());
-        return canonicalRepositoryPath;
+        return SecuritySupport.toRealPath(f);
     }
 
     private static SafePath createRealBasePath(SafePath safePath) throws IOException {

src/share/classes/jdk/jfr/internal/ShutdownHook.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2016, 2018, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2016, 2019, Oracle and/or its affiliates. All rights reserved.
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
  *
  * This code is free software; you can redistribute it and/or modify it
@@ -51,7 +51,7 @@ public void run() {
         // starting any "real" operations. In low memory situations,
         // we would like to take an OOM as early as possible.
         tlabDummyObject = new Object();
-
+        recorder.setInShutDown();
         for (PlatformRecording recording : recorder.getRecordings()) {
             if (recording.getDumpOnExit() && recording.getState() == RecordingState.RUNNING) {
                 dump(recording);

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

@@ -40,6 +40,7 @@
 import jdk.jfr.internal.LogLevel;
 import jdk.jfr.internal.LogTag;
 import jdk.jfr.internal.Logger;
+import jdk.jfr.internal.PlatformRecording;
 import jdk.jfr.internal.SecuritySupport;
 
 /*
@@ -50,19 +51,19 @@ abstract class AbstractEventStream implements EventStream {
     private final static AtomicLong counter = new AtomicLong(1);
 
     private final Object terminated = new Object();
-    private final boolean active;
     private final Runnable flushOperation = () -> dispatcher().runFlushActions();
     private final AccessControlContext accessControllerContext;
     private final StreamConfiguration configuration = new StreamConfiguration();
+    private final PlatformRecording recording;
 
     private volatile Thread thread;
     private Dispatcher dispatcher;
 
     private volatile boolean closed;
 
-    AbstractEventStream(AccessControlContext acc, boolean active) throws IOException {
+    AbstractEventStream(AccessControlContext acc, PlatformRecording recording) throws IOException {
         this.accessControllerContext = Objects.requireNonNull(acc);
-        this.active = active;
+        this.recording = recording;
     }
 
     @Override
@@ -229,7 +230,7 @@ private void startInternal(long startNanos) {
             if (configuration.started) {
                 throw new IllegalStateException("Event stream can only be started once");
             }
-            if (active && configuration.startTime == null) {
+            if (recording != null && configuration.startTime == null) {
                 configuration.setStartNanos(startNanos);
             }
             configuration.setStarted(true);

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

@@ -39,8 +39,10 @@ public final class ChunkHeader {
     private static final long CHUNK_SIZE_POSITION = 8;
     private static final long DURATION_NANOS_POSITION = 40;
     private static final long FILE_STATE_POSITION = 64;
+    private static final long FLAG_BYTE_POSITION = 67;
     private static final long METADATA_TYPE_ID = 0;
     private static final byte[] FILE_MAGIC = { 'F', 'L', 'R', '\0' };
+    private static final int MASK_FINAL_CHUNK = 1 << 1;
 
     private final short major;
     private final short minor;
@@ -58,6 +60,7 @@ public final class ChunkHeader {
     private long absoluteChunkEnd;
     private boolean isFinished;
     private boolean finished;
+    private boolean finalChunk;
 
     public ChunkHeader(RecordingInput input) throws IOException {
         this(input, 0, 0);
@@ -101,8 +104,7 @@ private ChunkHeader(RecordingInput input, long absoluteChunkStart, long id) thro
         Logger.log(LogTag.JFR_SYSTEM_PARSER, LogLevel.INFO, "Chunk: startTicks=" + chunkStartTicks);
         ticksPerSecond = input.readRawLong();
         Logger.log(LogTag.JFR_SYSTEM_PARSER, LogLevel.INFO, "Chunk: ticksPerSecond=" + ticksPerSecond);
-        input.readRawInt(); // features, not used
-
+        input.readRawInt(); // ignore file state and flag bits
         refresh();
         input.position(absoluteEventStart);
     }
@@ -123,6 +125,8 @@ void refresh() throws IOException {
             long durationNanos = input.readPhysicalLong();
             input.positionPhysical(absoluteChunkStart + FILE_STATE_POSITION);
             byte fileState2 =  input.readPhysicalByte();
+            input.positionPhysical(absoluteChunkStart + FLAG_BYTE_POSITION);
+            int flagByte = input.readPhysicalByte();
             if (fileState1 == fileState2) { // valid header
                 finished = fileState1 == 0;
                 if (metadataPosition != 0) {
@@ -150,6 +154,8 @@ void refresh() throws IOException {
                     Logger.log(LogTag.JFR_SYSTEM_PARSER, LogLevel.INFO, "Chunk: generation=" + fileState2);
                     Logger.log(LogTag.JFR_SYSTEM_PARSER, LogLevel.INFO, "Chunk: finished=" + isFinished);
                     Logger.log(LogTag.JFR_SYSTEM_PARSER, LogLevel.INFO, "Chunk: fileSize=" + input.size());
+                    this.finalChunk = (flagByte & MASK_FINAL_CHUNK) != 0;
+                    Logger.log(LogTag.JFR_SYSTEM_PARSER, LogLevel.INFO, "Chunk: finalChunk=" + finalChunk);
                     absoluteChunkEnd = absoluteChunkStart + chunkSize;
                     return;
                 }
@@ -183,6 +189,10 @@ public boolean isLastChunk() throws IOException {
         return input.getFileSize() == absoluteChunkEnd;
    }
 
+    public boolean isFinalChunk() {
+        return finalChunk;
+    }
+
     public boolean isFinished() throws IOException {
         return isFinished;
     }