8274315: JFR: One closed state per file or stream

Reviewed-by: mgronlun

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

@@ -41,6 +41,7 @@
 import jdk.jfr.internal.consumer.ChunkHeader;
 import jdk.jfr.internal.consumer.ChunkParser;
 import jdk.jfr.internal.consumer.FileAccess;
+import jdk.jfr.internal.consumer.ParserState;
 import jdk.jfr.internal.consumer.RecordingInput;
 
 /**
@@ -61,6 +62,7 @@
  */
 public final class RecordingFile implements Closeable {
 
+    private final ParserState parserState = new ParserState();
     private boolean isLastEventInChunk;
     private final File file;
     private RecordingInput input;
@@ -247,7 +249,7 @@ boolean isLastEventInChunk() {
     private void findNext() throws IOException {
         while (nextEvent == null) {
             if (chunkParser == null) {
-                chunkParser = new ChunkParser(input);
+                chunkParser = new ChunkParser(input, parserState);
             } else if (!chunkParser.isLastChunk()) {
                 chunkParser = chunkParser.nextChunkParser();
             } else {

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

@@ -65,7 +65,7 @@ public abstract class AbstractEventStream implements EventStream {
     private volatile Thread thread;
     private Dispatcher dispatcher;
 
-    private volatile boolean closed;
+    protected final ParserState parserState = new ParserState();
 
     private boolean daemon = false;
 
@@ -215,12 +215,12 @@ public final void awaitTermination(Duration timeout) throws InterruptedException
 
     protected abstract void process() throws IOException;
 
-    protected final void setClosed(boolean closed) {
-        this.closed = closed;
+    protected final void closeParser() {
+        parserState.close();
     }
 
     protected final boolean isClosed() {
-        return closed;
+        return parserState.isClosed();
     }
 
     public final void startAsync(long startNanos) {

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

@@ -97,32 +97,32 @@ private boolean is(int flags) {
     private final RecordingInput input;
     private final ChunkHeader chunkHeader;
     private final TimeConverter timeConverter;
-
+    private final ParserState parserState;
     private final LongMap<ConstantLookup> constantLookups;
 
     private LongMap<Type> typeMap;
     private LongMap<Parser> parsers;
     private boolean chunkFinished;
 
     private ParserConfiguration configuration;
-    private volatile boolean closed;
     private MetadataDescriptor previousMetadata;
     private MetadataDescriptor metadata;
     private boolean staleMetadata = true;
 
-    public ChunkParser(RecordingInput input) throws IOException {
-        this(input, new ParserConfiguration());
+    public ChunkParser(RecordingInput input, ParserState ps) throws IOException {
+        this(input, new ParserConfiguration(), ps);
     }
 
-    ChunkParser(RecordingInput input, ParserConfiguration pc) throws IOException {
-       this(new ChunkHeader(input), null, pc);
+    ChunkParser(RecordingInput input, ParserConfiguration pc, ParserState ps) throws IOException {
+       this(new ChunkHeader(input), null, pc, ps);
     }
 
-    private ChunkParser(ChunkParser previous) throws IOException {
-        this(new ChunkHeader(previous.input), previous, new ParserConfiguration());
+    private ChunkParser(ChunkParser previous, ParserState ps) throws IOException {
+        this(new ChunkHeader(previous.input), previous, new ParserConfiguration(), ps);
      }
 
-    private ChunkParser(ChunkHeader header, ChunkParser previous, ParserConfiguration pc) throws IOException {
+    private ChunkParser(ChunkHeader header, ChunkParser previous, ParserConfiguration pc, ParserState ps) throws IOException {
+        this.parserState = ps;
         this.configuration = pc;
         this.input = header.getInput();
         this.chunkHeader = header;
@@ -155,7 +155,7 @@ private ChunkParser(ChunkHeader header, ChunkParser previous, ParserConfiguratio
     }
 
     public ChunkParser nextChunkParser() throws IOException {
-        return new ChunkParser(chunkHeader.nextHeader(), this, configuration);
+        return new ChunkParser(chunkHeader.nextHeader(), this, configuration, parserState);
     }
 
     private void updateConfiguration() {
@@ -280,7 +280,7 @@ private boolean awaitUpdatedHeader(long absoluteChunkEnd, long filterEnd) throws
             Logger.log(LogTag.JFR_SYSTEM_PARSER, LogLevel.INFO, "Waiting for more data (streaming). Read so far: " + chunkHeader.getChunkSize() + " bytes");
         }
         while (true) {
-            if (closed) {
+            if (parserState.isClosed()) {
                 return true;
             }
             if (chunkHeader.getLastNanos() > filterEnd)  {
@@ -437,7 +437,7 @@ public boolean isLastChunk() throws IOException {
     }
 
     ChunkParser newChunkParser() throws IOException {
-        return new ChunkParser(this);
+        return new ChunkParser(this, parserState);
     }
 
     public boolean isChunkFinished() {
@@ -457,7 +457,7 @@ public boolean isFinalChunk() {
     }
 
     public void close() {
-        this.closed = true;
+        parserState.close();
         try {
             input.close();
         } catch(IOException e) {

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

@@ -83,7 +83,7 @@ public EventDirectoryStream(
 
     @Override
     public void close() {
-        setClosed(true);
+        closeParser();
         dispatcher().runCloseActions();
         repositoryFiles.close();
         if (currentParser != null) {
@@ -148,7 +148,7 @@ protected void processRecursionSafe() throws IOException {
         }
         currentChunkStartNanos = repositoryFiles.getTimestamp(path);
         try (RecordingInput input = new RecordingInput(path.toFile(), fileAccess)) {
-            currentParser = new ChunkParser(input, disp.parserConfiguration);
+            currentParser = new ChunkParser(input, disp.parserConfiguration, parserState);
             long segmentStart = currentParser.getStartNanos() + currentParser.getChunkDuration();
             long filterStart = validStartTime ? disp.startNanos : segmentStart;
             long filterEnd = disp.endTime != null ? disp.endNanos : Long.MAX_VALUE;

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

@@ -64,7 +64,7 @@ public void startAsync() {
 
     @Override
     public void close() {
-        setClosed(true);
+        closeParser();
         dispatcher().runCloseActions();
         try {
             input.close();
@@ -85,7 +85,7 @@ protected void process() throws IOException {
             end = disp.endNanos;
         }
 
-        currentParser = new ChunkParser(input, disp.parserConfiguration);
+        currentParser = new ChunkParser(input, disp.parserConfiguration, parserState);
         while (!isClosed()) {
             onMetadata(currentParser);
             if (currentParser.getStartNanos() > end) {

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

@@ -0,0 +1,38 @@
+/*
+ * Copyright (c) 2021, 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.internal.consumer;
+
+public final class ParserState {
+    private volatile boolean closed;
+
+    public boolean isClosed() {
+        return closed;
+    }
+
+    public void close() {
+        closed = true;
+    }
+}