infernap12/jdk
1
0
Fork 0
mirror of https://github.com/openjdk/jdk.git synced 2026-03-18 11:53:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

8233700: EventStream not closed

Browse Source 
Reviewed-by: mgronlun, mseledtsov
This commit is contained in:
Erik Gahlin 2019-11-22 17:20:43 +01:00
parent 008bdefad4
commit 8a5e087ed8
30 changed files with 577 additions and 79 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
src/hotspot/share/jfr/jni/jfrJniMethod.cpp
View File

@ -210,6 +210,10 @@ JVM_ENTRY_NO_ENV(void, jfr_begin_recording(JNIEnv* env, jobject jvm))
JfrRecorder::start_recording();
JVM_END
JVM_ENTRY_NO_ENV(jboolean, jfr_is_recording(JNIEnv * env, jobject jvm))
return JfrRecorder::is_recording() ? JNI_TRUE : JNI_FALSE;
JVM_END
JVM_ENTRY_NO_ENV(void, jfr_end_recording(JNIEnv* env, jobject jvm))
if (!JfrRecorder::is_recording()) {
return;
@ -217,6 +221,9 @@ JVM_ENTRY_NO_ENV(void, jfr_end_recording(JNIEnv* env, jobject jvm))
JfrRecorder::stop_recording();
JVM_END
JVM_ENTRY_NO_ENV(void, jfr_mark_chunk_final(JNIEnv * env, jobject jvm))
JfrRepository::mark_chunk_final();
JVM_END
JVM_ENTRY_NO_ENV(jboolean, jfr_emit_event(JNIEnv* env, jobject jvm, jlong eventTypeId, jlong timeStamp, jlong when))
JfrPeriodicEventSet::requestEvent((JfrEventId)eventTypeId);

4
src/hotspot/share/jfr/jni/jfrJniMethod.hpp
View File

@ -49,8 +49,12 @@ jboolean JNICALL jfr_destroy_jfr(JNIEnv* env, jobject jvm);
void JNICALL jfr_begin_recording(JNIEnv* env, jobject jvm);
jboolean JNICALL jfr_is_recording(JNIEnv* env, jobject jvm);
void JNICALL jfr_end_recording(JNIEnv* env, jobject jvm);
void JNICALL jfr_mark_chunk_final(JNIEnv* env, jobject jvm);
jboolean JNICALL jfr_emit_event(JNIEnv* env, jobject jvm, jlong eventTypeId, jlong timeStamp, jlong when);
jobject JNICALL jfr_get_all_event_classes(JNIEnv* env, jobject jvm);

2
src/hotspot/share/jfr/jni/jfrJniMethodRegistration.cpp
View File

@ -36,7 +36,9 @@ JfrJniMethodRegistration::JfrJniMethodRegistration(JNIEnv* env) {
if (jfr_clz != NULL) {
JNINativeMethod method[] = {
(char*)"beginRecording", (char*)"()V", (void*)jfr_begin_recording,
(char*)"isRecording", (char*)"()Z", (void*)jfr_is_recording,
(char*)"endRecording", (char*)"()V", (void*)jfr_end_recording,
(char*)"markChunkFinal", (char*)"()V", (void*)jfr_mark_chunk_final,
(char*)"counterTime", (char*)"()J", (void*)jfr_elapsed_counter,
(char*)"createJFR", (char*)"(Z)Z", (void*)jfr_create_jfr,
(char*)"destroyJFR", (char*)"()Z", (void*)jfr_destroy_jfr,

19
src/hotspot/share/jfr/recorder/repository/jfrChunk.cpp
View File

@ -59,7 +59,8 @@ JfrChunk::JfrChunk() :
_last_update_nanos(0),
_last_checkpoint_offset(0),
_last_metadata_offset(0),
_generation(1) {}
_generation(1),
_final(false) {}
JfrChunk::~JfrChunk() {
reset();
@ -86,10 +87,20 @@ u2 JfrChunk::minor_version() const {
return JFR_VERSION_MINOR;
}
u2 JfrChunk::capabilities() const {
void JfrChunk::mark_final() {
_final = true;
}
u2 JfrChunk::flags() const {
// chunk capabilities, CompressedIntegers etc
static bool compressed_integers = JfrOptionSet::compressed_integers();
return compressed_integers;
u2 flags = 0;
if (JfrOptionSet::compressed_integers()) {
flags |= 1 << 0;
}
if (_final) {
flags |= 1 << 1;
}
return flags;
}
int64_t JfrChunk::cpu_frequency() const {

5
src/hotspot/share/jfr/recorder/repository/jfrChunk.hpp
View File

@ -44,6 +44,7 @@ class JfrChunk : public JfrCHeapObj {
int64_t _last_checkpoint_offset;
int64_t _last_metadata_offset;
mutable u1 _generation;
bool _final;
JfrChunk();
~JfrChunk();
@ -53,7 +54,9 @@ class JfrChunk : public JfrCHeapObj {
u2 major_version() const;
u2 minor_version() const;
int64_t cpu_frequency() const;
u2 capabilities() const;
u2 flags() const;
void mark_final();
void update_start_ticks();
void update_start_nanos();

19
src/hotspot/share/jfr/recorder/repository/jfrChunkWriter.cpp
View File

@ -41,8 +41,8 @@ static const int64_t DURATION_NANOS_OFFSET = START_NANOS_OFFSET + SLOT_SIZE;
static const int64_t START_TICKS_OFFSET = DURATION_NANOS_OFFSET + SLOT_SIZE;
static const int64_t CPU_FREQUENCY_OFFSET = START_TICKS_OFFSET + SLOT_SIZE;
static const int64_t GENERATION_OFFSET = CPU_FREQUENCY_OFFSET + SLOT_SIZE;
static const int64_t CAPABILITY_OFFSET = GENERATION_OFFSET + 2;
static const int64_t HEADER_SIZE = CAPABILITY_OFFSET + 2;
static const int64_t FLAG_OFFSET = GENERATION_OFFSET + 2;
static const int64_t HEADER_SIZE = FLAG_OFFSET + 2;
static fio_fd open_chunk(const char* path) {
return path != NULL ? os::open(path, O_CREAT | O_RDWR, S_IREAD | S_IWRITE) : invalid_fd;
@ -117,8 +117,8 @@ class JfrChunkHeadWriter : public StackObj {
_writer->flush();
}
void write_capabilities() {
_writer->be_write(_chunk->capabilities());
void write_flags() {
_writer->be_write(_chunk->flags());
}
void write_size_to_generation(int64_t size, bool finalize) {
@ -135,7 +135,7 @@ class JfrChunkHeadWriter : public StackObj {
assert(_chunk != NULL, "invariant");
DEBUG_ONLY(assert_writer_position(_writer, SIZE_OFFSET);)
write_size_to_generation(size, finalize);
// no need to write capabilities
write_flags();
_writer->seek(size); // implicit flush
}
@ -146,7 +146,7 @@ class JfrChunkHeadWriter : public StackObj {
write_magic();
write_version();
write_size_to_generation(HEADER_SIZE, false);
write_capabilities();
write_flags();
DEBUG_ONLY(assert_writer_position(_writer, HEADER_SIZE);)
_writer->flush();
}
@ -201,7 +201,7 @@ int64_t JfrChunkWriter::write_chunk_header_checkpoint(bool flushpoint) {
head.write_time(false);
head.write_cpu_frequency();
head.write_next_generation();
head.write_capabilities();
head.write_flags();
assert(current_offset() - header_content_pos == HEADER_SIZE, "invariant");
const u4 checkpoint_size = current_offset() - event_size_offset;
write_padded_at_offset<u4>(checkpoint_size, event_size_offset);
@ -211,6 +211,11 @@ int64_t JfrChunkWriter::write_chunk_header_checkpoint(bool flushpoint) {
return sz_written;
}
void JfrChunkWriter::mark_chunk_final() {
assert(_chunk != NULL, "invariant");
_chunk->mark_final();
}
int64_t JfrChunkWriter::flush_chunk(bool flushpoint) {
assert(_chunk != NULL, "invariant");
const int64_t sz_written = write_chunk_header_checkpoint(flushpoint);

1
src/hotspot/share/jfr/recorder/repository/jfrChunkWriter.hpp
View File

@ -59,6 +59,7 @@ class JfrChunkWriter : public JfrChunkWriterBase {
bool has_metadata() const;
void set_time_stamp();
void mark_chunk_final();
};
#endif // SHARE_JFR_RECORDER_REPOSITORY_JFRCHUNKWRITER_HPP

4
src/hotspot/share/jfr/recorder/repository/jfrRepository.cpp
View File

@ -115,6 +115,10 @@ void JfrRepository::set_chunk_path(const char* path) {
chunkwriter().set_path(path);
}
void JfrRepository::mark_chunk_final() {
chunkwriter().mark_chunk_final();
}
jlong JfrRepository::current_chunk_start_nanos() {
return chunkwriter().current_chunk_start_nanos();
}

1
src/hotspot/share/jfr/recorder/repository/jfrRepository.hpp
View File

@ -70,6 +70,7 @@ class JfrRepository : public JfrCHeapObj {
public:
static void set_path(jstring location, JavaThread* jt);
static void set_chunk_path(jstring path, JavaThread* jt);
static void mark_chunk_final();
static void flush(JavaThread* jt);
static jlong current_chunk_start_nanos();
};

1
src/hotspot/share/jfr/recorder/service/jfrRecorderService.cpp
View File

@ -432,6 +432,7 @@ void JfrRecorderService::vm_error_rotation() {
if (_chunkwriter.is_valid()) {
Thread* const t = Thread::current();
_storage.flush_regular_buffer(t->jfr_thread_local()->native_buffer(), t);
_chunkwriter.mark_chunk_final();
invoke_flush();
_chunkwriter.set_time_stamp();
_repository.close_chunk();

4
src/jdk.jfr/share/classes/jdk/jfr/consumer/EventStream.java
View File

@ -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);
}
/**

3
src/jdk.jfr/share/classes/jdk/jfr/consumer/RecordingStream.java
View File

@ -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());

46
src/jdk.jfr/share/classes/jdk/jfr/internal/JVM.java
View File

@ -43,7 +43,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();
@ -68,6 +67,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
*
@ -75,6 +83,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
*
@ -97,13 +118,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 jdk.internal.event.Event}
@ -354,20 +369,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
@ -558,4 +559,5 @@ public final class JVM {
*@return start time of the recording in nanos, -1 in case of in-memory
*/
public native long getChunkStartNanos();
}

47
src/jdk.jfr/share/classes/jdk/jfr/internal/PlatformRecorder.java
View File

@ -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;
@ -53,6 +54,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;
@ -70,6 +72,7 @@ public final class PlatformRecorder {
private long recordingCounter = 0;
private RepositoryChunk currentChunk;
private boolean inShutdown;
public PlatformRecorder() throws Exception {
repository = Repository.getRepository();
@ -176,6 +179,10 @@ public final class PlatformRecorder {
}
}
synchronized void setInShutDown() {
this.inShutdown = true;
}
// called by shutdown hook
synchronized void destroy() {
try {
@ -198,7 +205,7 @@ public final class PlatformRecorder {
if (jvm.hasNativeJFR()) {
if (jvm.isRecording()) {
jvm.endRecording_();
jvm.endRecording();
}
jvm.destroyNativeJFR();
}
@ -236,7 +243,7 @@ public final class PlatformRecorder {
MetadataRepository.getInstance().setOutput(null);
}
currentChunk = newChunk;
jvm.beginRecording_();
jvm.beginRecording();
startNanos = jvm.getChunkStartNanos();
recording.setState(RecordingState.RUNNING);
updateSettings();
@ -289,11 +296,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;
@ -302,7 +313,7 @@ public final class PlatformRecorder {
// last memory
dumpMemoryToDestination(recording);
}
jvm.endRecording_();
jvm.endRecording();
disableEvents();
} else {
RepositoryChunk newChunk = null;
@ -327,7 +338,6 @@ public final class PlatformRecorder {
} else {
RequestEngine.setFlushInterval(Long.MAX_VALUE);
}
recording.setState(RecordingState.STOPPED);
}
@ -357,17 +367,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();
@ -584,6 +584,19 @@ public final class PlatformRecorder {
target.setInternalDuration(Duration.between(startTime, endTime));
}
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();
}
}
}

9
src/jdk.jfr/share/classes/jdk/jfr/internal/PlatformRecording.java
View File

@ -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
@ -811,4 +812,12 @@ public final class PlatformRecording implements AutoCloseable {
return Long.MAX_VALUE;
}
}
public long getFinalChunkStartNanos() {
return finalStartChunkNanos;
}
public void setFinalStartnanos(long chunkStartNanos) {
this.finalStartChunkNanos = chunkStartNanos;
}
}

5
src/jdk.jfr/share/classes/jdk/jfr/internal/Repository.java
View File

@ -85,6 +85,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);
@ -115,9 +116,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 {

4
src/jdk.jfr/share/classes/jdk/jfr/internal/ShutdownHook.java
View File

@ -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 @@ 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);

9
src/jdk.jfr/share/classes/jdk/jfr/internal/consumer/AbstractEventStream.java
View File

@ -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);

14
src/jdk.jfr/share/classes/jdk/jfr/internal/consumer/ChunkHeader.java
View File

@ -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;
}

4
src/jdk.jfr/share/classes/jdk/jfr/internal/consumer/ChunkParser.java
View File

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

30
src/jdk.jfr/share/classes/jdk/jfr/internal/consumer/EventDirectoryStream.java
View File

@ -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 {
}
}
}
}

2
src/jdk.jfr/share/classes/jdk/jfr/internal/consumer/EventFileStream.java
View File

@ -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);
}

4
src/jdk.jfr/share/classes/jdk/jfr/internal/consumer/RepositoryFiles.java
View File

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

8
src/jdk.jfr/share/classes/jdk/jfr/internal/dcmd/DCmdConfigure.java
View File

@ -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;
@ -89,11 +90,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);
}

63
test/jdk/jdk/jfr/api/consumer/recordingstream/TestStoppedRecording.java Normal file
View File

@ -0,0 +1,63 @@
/*
* 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
* @requires vm.hasJFR
* @library /test/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();
}
}
}

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

@ -39,14 +39,14 @@ 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
* @requires vm.hasJFR
* @library /test/lib /test/jdk
* @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;
}

68
test/jdk/jdk/jfr/api/consumer/streaming/TestJVMCrash.java Normal file
View File

@ -0,0 +1,68 @@
/*
* 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
* @requires vm.hasJFR
* @library /test/lib /test/jdk
* @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...");
}
}
}
}

58
test/jdk/jdk/jfr/api/consumer/streaming/TestJVMExit.java Normal file
View File

@ -0,0 +1,58 @@
/*
* 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
* @requires vm.hasJFR
* @library /test/lib /test/jdk
* @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();
}
}
}

69
test/jdk/jdk/jfr/api/consumer/streaming/TestOutOfProcessMigration.java Normal file
View File

@ -0,0 +1,69 @@
/*
* 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
* @requires vm.hasJFR
* @library /test/lib /test/jdk
* @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();
}
}
}

138
test/jdk/jdk/jfr/api/consumer/streaming/TestProcess.java Normal file
View File

@ -0,0 +1,138 @@
/*
* 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.io.IOException;
import java.io.RandomAccessFile;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.Properties;
import jdk.internal.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 Path path;
public TestProcess(String name) throws IOException {
this.path = Paths.get("action-" + System.currentTimeMillis()).toAbsolutePath();
String[] args = {
"--add-exports",
"java.base/jdk.internal.misc=ALL-UNNAMED",
"-XX:StartFlightRecording:settings=none",
TestProcess.class.getName(), path.toString()
};
ProcessBuilder pb = ProcessTools.createJavaProcessBuilder(false, args);
process = ProcessTools.startProcess(name, pb);
}
public static void main(String... args) throws Exception {
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 = Files.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(process.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 {
Files.writeString(path, "crash");
} catch (IOException ioe) {
ioe.printStackTrace();
}
}
public void exit() {
try {
Files.writeString(path, "exit");
} catch (IOException ioe) {
ioe.printStackTrace();
}
}
public long pid() {
return process.pid();
}
}