[Backport] 8233700: EventStream not closed

Summary:

Test Plan: jdk/jfr

Reviewed-by: yuleil

Issue: https://github.com/alibaba/dragonwell8/issues/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 interface EventStream extends AutoCloseable {
     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 final class RecordingStream implements AutoCloseable, EventStream {
         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,6 +68,15 @@ public final class JVM {
     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
      *
@@ -76,6 +84,19 @@ public final class JVM {
      */
     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 @@ public final class 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 @@ public final class 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 final class JVM {
      *@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.LogLevel.WARN;
 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.Recording;
 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 final class PlatformRecorder {
         }
     }
 
+    synchronized void setInShutDown() {
+        this.inShutdown = true;
+    }
+
     // called by shutdown hook
     synchronized void destroy() {
         try {
@@ -200,7 +207,7 @@ public final class PlatformRecorder {
 
         if (jvm.hasNativeJFR()) {
             if (jvm.isRecording()) {
-                jvm.endRecording_();
+                jvm.endRecording();
             }
             jvm.destroyNativeJFR();
         }
@@ -238,7 +245,7 @@ public final class PlatformRecorder {
                 MetadataRepository.getInstance().setOutput(null);
             }
             currentChunk = newChunk;
-            jvm.beginRecording_();
+            jvm.beginRecording();
             startNanos = jvm.getChunkStartNanos();
             recording.setState(RecordingState.RUNNING);
             updateSettings();
@@ -291,11 +298,15 @@ public final class PlatformRecorder {
             }
         }
         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 @@ public final class PlatformRecorder {
                 // last memory
                 dumpMemoryToDestination(recording);
             }
-            jvm.endRecording_();
+            jvm.endRecording();
             disableEvents();
         } else {
             RepositoryChunk newChunk = null;
@@ -329,7 +340,6 @@ public final class PlatformRecorder {
         } else {
             RequestEngine.setFlushInterval(Long.MAX_VALUE);
         }
-
         recording.setState(RecordingState.STOPPED);
     }
 
@@ -359,17 +369,7 @@ public final class PlatformRecorder {
         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 @@ public final class PlatformRecorder {
     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 final class PlatformRecording implements AutoCloseable {
             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 @@ public final class Repository {
             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 @@ public final class Repository {
         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

 /*
- * 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 @@ final class ShutdownHook implements Runnable {
         // 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.consumer.RecordedEvent;
 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 @@ abstract class AbstractEventStream implements EventStream {
             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 @@ public final class ChunkHeader {
         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 @@ public final class ChunkHeader {
             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 @@ public final class ChunkHeader {
                     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 final class ChunkHeader {
         return input.getFileSize() == absoluteChunkEnd;
    }
 
+    public boolean isFinalChunk() {
+        return finalChunk;
+    }
+
     public boolean isFinished() throws IOException {
         return isFinished;
     }

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

@@ -448,4 +448,8 @@ public final class ChunkParser {
         return chunkHeader.getStartNanos();
     }
 
+    public boolean isFinalChunk() {
+        return chunkHeader.isFinalChunk();
+    }
+
 }

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

@@ -35,6 +35,7 @@ import java.util.Objects;
 
 import jdk.jfr.consumer.RecordedEvent;
 import jdk.jfr.internal.JVM;
+import jdk.jfr.internal.PlatformRecording;
 import jdk.jfr.internal.Utils;
 import jdk.jfr.internal.consumer.ChunkParser.ParserConfiguration;
 
@@ -43,12 +44,12 @@ import jdk.jfr.internal.consumer.ChunkParser.ParserConfiguration;
  * with chunk files.
  *
  */
-public final class EventDirectoryStream extends AbstractEventStream {
+public class EventDirectoryStream extends AbstractEventStream {
 
     private final static Comparator<? super RecordedEvent> EVENT_COMPARATOR = JdkJfrConsumer.instance().eventComparator();
 
     private final RepositoryFiles repositoryFiles;
-    private final boolean active;
+    private final PlatformRecording recording;
     private final FileAccess fileAccess;
 
     private ChunkParser currentParser;
@@ -56,10 +57,10 @@ public final class EventDirectoryStream extends AbstractEventStream {
     private RecordedEvent[] sortedCache;
     private int threadExclusionLevel = 0;
 
-    public EventDirectoryStream(AccessControlContext acc, Path p, FileAccess fileAccess, boolean active) throws IOException {
-        super(acc, active);
+    public EventDirectoryStream(AccessControlContext acc, Path p, FileAccess fileAccess, PlatformRecording recording) throws IOException {
+        super(acc, recording);
         this.fileAccess = Objects.requireNonNull(fileAccess);
-        this.active = active;
+        this.recording = recording;
         this.repositoryFiles = new RepositoryFiles(fileAccess, p);
     }
 
@@ -104,7 +105,7 @@ public final class EventDirectoryStream extends AbstractEventStream {
         Dispatcher disp = dispatcher();
 
         Path path;
-        boolean validStartTime = active || disp.startTime != null;
+        boolean validStartTime = recording != null || disp.startTime != null;
         if (validStartTime) {
             path = repositoryFiles.firstPath(disp.startNanos);
         } else {
@@ -139,8 +140,17 @@ public final class EventDirectoryStream extends AbstractEventStream {
                         return;
                     }
                 }
+                if (isLastChunk()) {
+                    // Recording was stopped/closed externally, and no more data to process.
+                    return;
+                }
 
+                if (repositoryFiles.hasFixedPath() && currentParser.isFinalChunk()) {
+                    // JVM process exited/crashed, or repository migrated to an unknown location
+                    return;
+                }
                 if (isClosed()) {
+                    // Stream was closed
                     return;
                 }
                 long durationNanos = currentParser.getChunkDuration();
@@ -162,6 +172,13 @@ public final class EventDirectoryStream extends AbstractEventStream {
         }
     }
 
+    private boolean isLastChunk() {
+        if (recording == null) {
+            return false;
+        }
+        return recording.getFinalChunkStartNanos() >= currentParser.getStartNanos();
+    }
+
     private boolean processOrdered(Dispatcher c, boolean awaitNewEvents) throws IOException {
         if (sortedCache == null) {
             sortedCache = new RecordedEvent[100_000];
@@ -206,4 +223,5 @@ public final class EventDirectoryStream extends AbstractEventStream {
             }
         }
     }
+
 }

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

@@ -50,7 +50,7 @@ public final class EventFileStream extends AbstractEventStream {
     private RecordedEvent[] cacheSorted;
 
     public EventFileStream(AccessControlContext acc, Path path) throws IOException {
-        super(acc, false);
+        super(acc, null);
         Objects.requireNonNull(path);
         this.input = new RecordingInput(path.toFile(), FileAccess.UNPRIVILIGED);
     }

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

@@ -227,4 +227,8 @@ public final class RepositoryFiles {
             waitObject.notify();
         }
     }
+
+    public boolean hasFixedPath() {
+        return repository != null;
+    }
 }

src/share/classes/jdk/jfr/internal/dcmd/DCmdConfigure.java

@@ -32,6 +32,7 @@ import jdk.jfr.internal.LogLevel;
 import jdk.jfr.internal.LogTag;
 import jdk.jfr.internal.Logger;
 import jdk.jfr.internal.Options;
+import jdk.jfr.internal.PlatformRecorder;
 import jdk.jfr.internal.PrivateAccess;
 import jdk.jfr.internal.Repository;
 import jdk.jfr.internal.SecuritySupport.SafePath;
@@ -101,11 +102,12 @@ final class DCmdConfigure extends AbstractDCmd {
         if (repositoryPath != null) {
             try {
                 SafePath s = new SafePath(repositoryPath);
-                Repository.getRepository().setBasePath(s);
-                Logger.log(LogTag.JFR, LogLevel.INFO, "Base repository path set to " + repositoryPath);
                 if (FlightRecorder.isInitialized()) {
-                    PrivateAccess.getInstance().getPlatformRecorder().rotateIfRecordingToDisk();;
+                    PrivateAccess.getInstance().getPlatformRecorder().migrate(s);
+                } else {
+                    Repository.getRepository().setBasePath(s);
                 }
+                Logger.log(LogTag.JFR, LogLevel.INFO, "Base repository path set to " + repositoryPath);
             } catch (Exception e) {
                 throw new DCmdException("Could not use " + repositoryPath + " as repository. " + e.getMessage(), e);
             }

test/jdk/jfr/api/consumer/recordingstream/TestStoppedRecording.java

+/*
+ * Copyright (c) 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
+ * under the terms of the GNU General Public License version 2 only, as
+ * published by the Free Software Foundation.  Oracle designates this
+ * particular file as subject to the "Classpath" exception as provided
+ * by Oracle in the LICENSE file that accompanied this code.
+ *
+ * This code is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
+ * version 2 for more details (a copy is included in the LICENSE file that
+ * accompanied this code).
+ *
+ * You should have received a copy of the GNU General Public License version
+ * 2 along with this work; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
+ *
+ * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
+ * or visit www.oracle.com if you need additional information or have any
+ * questions.
+ */
+
+package jdk.jfr.api.consumer.recordingstream;
+
+import java.util.concurrent.CountDownLatch;
+
+import jdk.jfr.Event;
+import jdk.jfr.FlightRecorder;
+import jdk.jfr.consumer.RecordingStream;
+
+/**
+ * @test
+ * @summary Tests that a RecordingStream is closed if the underlying Recording
+ *          is stopped.
+ * @key jfr
+ * @library /lib /
+ * @run main/othervm jdk.jfr.api.consumer.recordingstream.TestStoppedRecording
+ */
+public class TestStoppedRecording {
+
+    private static final class StopEvent extends Event {
+    }
+
+    public static void main(String... args) throws Exception {
+        CountDownLatch latch = new CountDownLatch(1);
+        try (RecordingStream rs = new RecordingStream()) {
+            rs.onEvent(e -> {
+                FlightRecorder.getFlightRecorder().getRecordings().get(0).stop();
+            });
+            rs.onClose(() -> {
+                latch.countDown();
+            });
+            rs.startAsync();
+            StopEvent stop = new StopEvent();
+            stop.commit();
+            latch.await();
+        }
+    }
+}

test/jdk/jfr/api/consumer/streaming/TestRepositoryMigration.java → test/jdk/jfr/api/consumer/streaming/TestInProcessMigration.java

@@ -39,13 +39,13 @@ import jdk.jfr.jcmd.JcmdHelper;
 
 /**
  * @test
- * @summary Verifies that is possible to stream from a repository that is being
- *          moved.
+ * @summary Verifies that is possible to stream from an in-process repository
+ *          that is being moved.
  * @key jfr
  * @library /lib /
- * @run main/othervm jdk.jfr.api.consumer.streaming.TestRepositoryMigration
+ * @run main/othervm jdk.jfr.api.consumer.streaming.TestInProcessMigration
  */
-public class TestRepositoryMigration {
+public class TestInProcessMigration {
     static class MigrationEvent extends Event {
         int id;
     }

test/jdk/jfr/api/consumer/streaming/TestJVMCrash.java

+/*
+ * Copyright (c) 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
+ * under the terms of the GNU General Public License version 2 only, as
+ * published by the Free Software Foundation.  Oracle designates this
+ * particular file as subject to the "Classpath" exception as provided
+ * by Oracle in the LICENSE file that accompanied this code.
+ *
+ * This code is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
+ * version 2 for more details (a copy is included in the LICENSE file that
+ * accompanied this code).
+ *
+ * You should have received a copy of the GNU General Public License version
+ * 2 along with this work; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
+ *
+ * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
+ * or visit www.oracle.com if you need additional information or have any
+ * questions.
+ */
+package jdk.jfr.api.consumer.streaming;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import jdk.jfr.consumer.EventStream;
+
+/**
+ * @test
+ * @summary Test that a stream ends/closes when an application crashes.
+ * @key jfr
+ * @library /lib /
+ * @modules jdk.jfr jdk.attach java.base/jdk.internal.misc
+ *
+ * @run main/othervm jdk.jfr.api.consumer.streaming.TestJVMCrash
+ */
+public class TestJVMCrash {
+
+    public static void main(String... args) throws Exception {
+        int id = 1;
+        while (true) {
+            TestProcess process = new TestProcess("crash-application-" + id++);
+            AtomicInteger eventCounter = new AtomicInteger();
+            try (EventStream es = EventStream.openRepository(process.getRepository())) {
+                // Start from first event in repository
+                es.setStartTime(Instant.EPOCH);
+                es.onEvent(e -> {
+                    if (eventCounter.incrementAndGet() == TestProcess.NUMBER_OF_EVENTS) {
+                        process.crash();
+                    }
+                });
+                es.startAsync();
+                // If crash corrupts chunk in repository, retry in 30 seconds
+                es.awaitTermination(Duration.ofSeconds(30));
+                if (eventCounter.get() == TestProcess.NUMBER_OF_EVENTS) {
+                    return;
+                }
+                System.out.println("Incorrect event count. Retrying...");
+            }
+        }
+    }
+}

test/jdk/jfr/api/consumer/streaming/TestJVMExit.java

+/*
+ * Copyright (c) 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
+ * under the terms of the GNU General Public License version 2 only, as
+ * published by the Free Software Foundation.  Oracle designates this
+ * particular file as subject to the "Classpath" exception as provided
+ * by Oracle in the LICENSE file that accompanied this code.
+ *
+ * This code is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
+ * version 2 for more details (a copy is included in the LICENSE file that
+ * accompanied this code).
+ *
+ * You should have received a copy of the GNU General Public License version
+ * 2 along with this work; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
+ *
+ * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
+ * or visit www.oracle.com if you need additional information or have any
+ * questions.
+ */
+package jdk.jfr.api.consumer.streaming;
+
+import java.time.Instant;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import jdk.jfr.consumer.EventStream;
+
+/**
+ * @test
+ * @summary Test that a stream ends/closes when an application exists.
+ * @key jfr
+ * @library /lib /
+ * @modules jdk.jfr jdk.attach java.base/jdk.internal.misc
+ *
+ * @run main/othervm jdk.jfr.api.consumer.streaming.TestJVMExit
+ */
+public class TestJVMExit {
+
+    public static void main(String... args) throws Exception {
+        TestProcess process = new TestProcess("exit-application");
+        AtomicInteger eventCounter = new AtomicInteger();
+        try (EventStream es = EventStream.openRepository(process.getRepository())) {
+            // Start from first event in repository
+            es.setStartTime(Instant.EPOCH);
+            es.onEvent(e -> {
+                if (eventCounter.incrementAndGet() == TestProcess.NUMBER_OF_EVENTS) {
+                    process.exit();
+                }
+            });
+            es.start();
+        }
+    }
+}

test/jdk/jfr/api/consumer/streaming/TestOutOfProcessMigration.java

+/*
+ * Copyright (c) 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
+ * under the terms of the GNU General Public License version 2 only, as
+ * published by the Free Software Foundation.  Oracle designates this
+ * particular file as subject to the "Classpath" exception as provided
+ * by Oracle in the LICENSE file that accompanied this code.
+ *
+ * This code is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
+ * version 2 for more details (a copy is included in the LICENSE file that
+ * accompanied this code).
+ *
+ * You should have received a copy of the GNU General Public License version
+ * 2 along with this work; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
+ *
+ * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
+ * or visit www.oracle.com if you need additional information or have any
+ * questions.
+ */
+
+package jdk.jfr.api.consumer.streaming;
+
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.time.Instant;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import jdk.jfr.consumer.EventStream;
+import jdk.test.lib.dcmd.CommandExecutor;
+import jdk.test.lib.dcmd.PidJcmdExecutor;
+import jdk.test.lib.process.OutputAnalyzer;
+
+/**
+ * @test
+ * @summary Verifies that a out-of-process stream is closed when the repository
+ *          is changed.
+ * @key jfr
+ * @library /lib /
+ * @modules jdk.jfr jdk.attach java.base/jdk.internal.misc
+ * @run main/othervm jdk.jfr.api.consumer.streaming.TestOutOfProcessMigration
+ */
+public class TestOutOfProcessMigration {
+    public static void main(String... args) throws Exception {
+        Path newRepo = Paths.get("new-repository").toAbsolutePath();
+
+        TestProcess process = new TestProcess("application");
+        AtomicInteger eventCounter = new AtomicInteger();
+        try (EventStream es = EventStream.openRepository(process.getRepository())) {
+            // Start from first event in repository
+            es.setStartTime(Instant.EPOCH);
+            es.onEvent(e -> {
+                if (eventCounter.incrementAndGet() == TestProcess.NUMBER_OF_EVENTS) {
+                    System.out.println("Changing repository to " + newRepo + " ...");
+                    CommandExecutor executor = new PidJcmdExecutor(String.valueOf(process.pid()));
+                    // This should close stream
+                    OutputAnalyzer oa = executor.execute("JFR.configure repositorypath=" + newRepo);
+                    System.out.println(oa);
+                }
+            });
+            es.start();
+        }
+    }
+}

test/jdk/jfr/api/consumer/streaming/TestProcess.java

+/*
+ * Copyright (c) 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
+ * under the terms of the GNU General Public License version 2 only, as
+ * published by the Free Software Foundation.  Oracle designates this
+ * particular file as subject to the "Classpath" exception as provided
+ * by Oracle in the LICENSE file that accompanied this code.
+ *
+ * This code is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
+ * version 2 for more details (a copy is included in the LICENSE file that
+ * accompanied this code).
+ *
+ * You should have received a copy of the GNU General Public License version
+ * 2 along with this work; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
+ *
+ * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
+ * or visit www.oracle.com if you need additional information or have any
+ * questions.
+ */
+package jdk.jfr.api.consumer.streaming;
+
+import static jdk.test.lib.Asserts.assertTrue;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.OutputStream;
+import java.io.RandomAccessFile;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Properties;
+
+import sun.misc.Unsafe;
+import jdk.jfr.Event;
+import jdk.test.lib.process.ProcessTools;
+
+import com.sun.tools.attach.VirtualMachine;
+
+/**
+ * Class that emits a NUMBER_OF_EVENTS and then awaits crash or exit
+ *
+ * Requires jdk.attach module.
+ *
+ */
+public final class TestProcess {
+
+    private static class TestEvent extends Event {
+    }
+
+    public final static int NUMBER_OF_EVENTS = 10;
+
+    private final Process process;
+    private final long pid;
+    private final Path path;
+
+    public TestProcess(String name) throws IOException {
+        this.path = Paths.get("action-" + System.currentTimeMillis()).toAbsolutePath();
+        Path pidPath = Paths.get("pid-" + System.currentTimeMillis()).toAbsolutePath();
+        String[] args = {
+                "-XX:StartFlightRecording:settings=none",
+                TestProcess.class.getName(), path.toString(),
+                pidPath.toString()
+            };
+        ProcessBuilder pb = ProcessTools.createJavaProcessBuilder(false, args);
+        process = ProcessTools.startProcess(name, pb);
+        do {
+            takeNap();
+        } while (!pidPath.toFile().exists());
+
+        String pidStr;
+        do {
+            pidStr = readString(pidPath);
+        } while (pidStr == null || !pidStr.endsWith("@"));
+
+        pid = Long.valueOf(pidStr.substring(0, pidStr.length()-1));
+    }
+
+    private static String readString(Path path) throws IOException {
+        try (InputStream in = new FileInputStream(path.toFile())) {
+            byte[] bytes = new byte[32];
+            int length = in.read(bytes);
+            assertTrue(length < bytes.length, "bytes array to small");
+            if (length == -1) {
+                return null;
+            }
+            return new String(bytes, 0, length);
+        }
+    }
+
+    private static void writeString(Path path, String content) throws IOException {
+        try (OutputStream out = new FileOutputStream(path.toFile())) {
+            out.write(content.getBytes());
+        }
+    }
+
+    public static void main(String... args) throws Exception {
+        Path pidPath = Paths.get(args[1]);
+        writeString(pidPath, ProcessTools.getProcessId() + "@");
+
+        for (int i = 0; i < NUMBER_OF_EVENTS; i++) {
+            TestEvent e = new TestEvent();
+            e.commit();
+        }
+
+        Path path = Paths.get(args[0]);
+        while (true) {
+            try {
+                String action = readString(path);
+                if ("crash".equals(action)) {
+                    System.out.println("About to crash...");
+                    Unsafe.getUnsafe().putInt(0L, 0);
+                }
+                if ("exit".equals(action)) {
+                    System.out.println("About to exit...");
+                    System.exit(0);
+                }
+            } catch (Exception ioe) {
+                // Ignore
+            }
+            takeNap();
+        }
+    }
+
+    public Path getRepository() {
+        while (true) {
+            try {
+                VirtualMachine vm = VirtualMachine.attach(String.valueOf(pid));
+                Properties p = vm.getSystemProperties();
+                vm.detach();
+                String repo = (String) p.get("jdk.jfr.repository");
+                if (repo != null) {
+                    return Paths.get(repo);
+                }
+            } catch (Exception e) {
+                System.out.println("Attach failed: " + e.getMessage());
+                System.out.println("Retrying...");
+            }
+            takeNap();
+        }
+    }
+
+    private static void takeNap() {
+        try {
+            Thread.sleep(10);
+        } catch (InterruptedException ie) {
+            // ignore
+        }
+    }
+
+    public void crash() {
+        try {
+            writeString(path, "crash");
+        } catch (IOException ioe) {
+            ioe.printStackTrace();
+        }
+    }
+
+    public void exit() {
+        try {
+            writeString(path, "exit");
+        } catch (IOException ioe) {
+            ioe.printStackTrace();
+        }
+    }
+
+    public long pid() {
+        return pid;
+    }
+}