src/jdk.jfr/share/classes/jdk/jfr/internal/consumer/EventDirectoryStream.java
author egahlin
Fri, 22 Nov 2019 17:20:43 +0100
changeset 59226 a0f39cc47387
parent 58863 c16ac7a2eba4
child 59310 72f3dd43dd28
permissions -rw-r--r--
8233700: EventStream not closed Reviewed-by: mgronlun, mseledtsov
Ignore whitespace changes - Everywhere: Within whitespace: At end of lines:
58863
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
     1
/*
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
     2
 * Copyright (c) 2019, Oracle and/or its affiliates. All rights reserved.
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
     3
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
     4
 *
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
     5
 * This code is free software; you can redistribute it and/or modify it
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
     6
 * under the terms of the GNU General Public License version 2 only, as
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
     7
 * published by the Free Software Foundation.  Oracle designates this
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
     8
 * particular file as subject to the "Classpath" exception as provided
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
     9
 * by Oracle in the LICENSE file that accompanied this code.
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    10
 *
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    11
 * This code is distributed in the hope that it will be useful, but WITHOUT
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    12
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    13
 * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    14
 * version 2 for more details (a copy is included in the LICENSE file that
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    15
 * accompanied this code).
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    16
 *
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    17
 * You should have received a copy of the GNU General Public License version
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    18
 * 2 along with this work; if not, write to the Free Software Foundation,
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    19
 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    20
 *
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    21
 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    22
 * or visit www.oracle.com if you need additional information or have any
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    23
 * questions.
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    24
 */
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    25
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    26
package jdk.jfr.internal.consumer;
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    27
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    28
import java.io.IOException;
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    29
import java.nio.file.Path;
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    30
import java.security.AccessControlContext;
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    31
import java.time.Instant;
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    32
import java.util.Arrays;
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    33
import java.util.Comparator;
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    34
import java.util.Objects;
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    35
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    36
import jdk.jfr.consumer.RecordedEvent;
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    37
import jdk.jfr.internal.JVM;
59226
a0f39cc47387 8233700: EventStream not closed
egahlin
parents: 58863
diff changeset
    38
import jdk.jfr.internal.PlatformRecording;
58863
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    39
import jdk.jfr.internal.Utils;
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    40
import jdk.jfr.internal.consumer.ChunkParser.ParserConfiguration;
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    41
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    42
/**
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    43
 * Implementation of an {@code EventStream}} that operates against a directory
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    44
 * with chunk files.
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    45
 *
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    46
 */
59226
a0f39cc47387 8233700: EventStream not closed
egahlin
parents: 58863
diff changeset
    47
public class EventDirectoryStream extends AbstractEventStream {
58863
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    48
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    49
    private final static Comparator<? super RecordedEvent> EVENT_COMPARATOR = JdkJfrConsumer.instance().eventComparator();
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    50
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    51
    private final RepositoryFiles repositoryFiles;
59226
a0f39cc47387 8233700: EventStream not closed
egahlin
parents: 58863
diff changeset
    52
    private final PlatformRecording recording;
58863
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    53
    private final FileAccess fileAccess;
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    54
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    55
    private ChunkParser currentParser;
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    56
    private long currentChunkStartNanos;
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    57
    private RecordedEvent[] sortedCache;
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    58
    private int threadExclusionLevel = 0;
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    59
59226
a0f39cc47387 8233700: EventStream not closed
egahlin
parents: 58863
diff changeset
    60
    public EventDirectoryStream(AccessControlContext acc, Path p, FileAccess fileAccess, PlatformRecording recording) throws IOException {
a0f39cc47387 8233700: EventStream not closed
egahlin
parents: 58863
diff changeset
    61
        super(acc, recording);
58863
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    62
        this.fileAccess = Objects.requireNonNull(fileAccess);
59226
a0f39cc47387 8233700: EventStream not closed
egahlin
parents: 58863
diff changeset
    63
        this.recording = recording;
58863
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    64
        this.repositoryFiles = new RepositoryFiles(fileAccess, p);
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    65
    }
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    66
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    67
    @Override
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    68
    public void close() {
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    69
        setClosed(true);
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    70
        dispatcher().runCloseActions();
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    71
        repositoryFiles.close();
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    72
    }
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    73
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    74
    @Override
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    75
    public void start() {
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    76
        start(Utils.timeToNanos(Instant.now()));
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    77
    }
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    78
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    79
    @Override
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    80
    public void startAsync() {
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    81
        startAsync(Utils.timeToNanos(Instant.now()));
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    82
    }
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    83
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    84
    @Override
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    85
    protected void process() throws IOException {
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    86
        JVM jvm = JVM.getJVM();
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    87
        Thread t = Thread.currentThread();
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    88
        try {
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    89
            if (jvm.isExcluded(t)) {
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    90
                threadExclusionLevel++;
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    91
            } else {
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    92
                jvm.exclude(t);
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    93
            }
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    94
            processRecursionSafe();
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    95
        } finally {
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    96
            if (threadExclusionLevel > 0) {
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    97
                threadExclusionLevel--;
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    98
            } else {
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
    99
                jvm.include(t);
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   100
            }
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   101
        }
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   102
    }
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   103
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   104
    protected void processRecursionSafe() throws IOException {
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   105
        Dispatcher disp = dispatcher();
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   106
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   107
        Path path;
59226
a0f39cc47387 8233700: EventStream not closed
egahlin
parents: 58863
diff changeset
   108
        boolean validStartTime = recording != null || disp.startTime != null;
58863
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   109
        if (validStartTime) {
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   110
            path = repositoryFiles.firstPath(disp.startNanos);
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   111
        } else {
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   112
            path = repositoryFiles.lastPath();
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   113
        }
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   114
        if (path == null) { // closed
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   115
            return;
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   116
        }
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   117
        currentChunkStartNanos = repositoryFiles.getTimestamp(path);
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   118
        try (RecordingInput input = new RecordingInput(path.toFile(), fileAccess)) {
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   119
            currentParser = new ChunkParser(input, disp.parserConfiguration);
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   120
            long segmentStart = currentParser.getStartNanos() + currentParser.getChunkDuration();
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   121
            long filterStart = validStartTime ? disp.startNanos : segmentStart;
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   122
            long filterEnd = disp.endTime != null ? disp.endNanos: Long.MAX_VALUE;
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   123
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   124
            while (!isClosed()) {
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   125
                boolean awaitnewEvent = false;
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   126
                while (!isClosed() && !currentParser.isChunkFinished()) {
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   127
                    disp = dispatcher();
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   128
                    ParserConfiguration pc = disp.parserConfiguration;
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   129
                    pc.filterStart = filterStart;
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   130
                    pc.filterEnd = filterEnd;
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   131
                    currentParser.updateConfiguration(pc, true);
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   132
                    currentParser.setFlushOperation(getFlushOperation());
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   133
                    if (pc.isOrdered()) {
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   134
                        awaitnewEvent = processOrdered(disp, awaitnewEvent);
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   135
                    } else {
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   136
                        awaitnewEvent = processUnordered(disp, awaitnewEvent);
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   137
                    }
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   138
                    if (currentParser.getStartNanos() + currentParser.getChunkDuration() > filterEnd) {
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   139
                        close();
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   140
                        return;
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   141
                    }
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   142
                }
59226
a0f39cc47387 8233700: EventStream not closed
egahlin
parents: 58863
diff changeset
   143
                if (isLastChunk()) {
a0f39cc47387 8233700: EventStream not closed
egahlin
parents: 58863
diff changeset
   144
                    // Recording was stopped/closed externally, and no more data to process.
a0f39cc47387 8233700: EventStream not closed
egahlin
parents: 58863
diff changeset
   145
                    return;
a0f39cc47387 8233700: EventStream not closed
egahlin
parents: 58863
diff changeset
   146
                }
58863
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   147
59226
a0f39cc47387 8233700: EventStream not closed
egahlin
parents: 58863
diff changeset
   148
                if (repositoryFiles.hasFixedPath() && currentParser.isFinalChunk()) {
a0f39cc47387 8233700: EventStream not closed
egahlin
parents: 58863
diff changeset
   149
                    // JVM process exited/crashed, or repository migrated to an unknown location
a0f39cc47387 8233700: EventStream not closed
egahlin
parents: 58863
diff changeset
   150
                    return;
a0f39cc47387 8233700: EventStream not closed
egahlin
parents: 58863
diff changeset
   151
                }
58863
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   152
                if (isClosed()) {
59226
a0f39cc47387 8233700: EventStream not closed
egahlin
parents: 58863
diff changeset
   153
                    // Stream was closed
58863
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   154
                    return;
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   155
                }
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   156
                long durationNanos = currentParser.getChunkDuration();
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   157
                if (durationNanos == 0) {
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   158
                    // Avoid reading the same chunk again and again if
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   159
                    // duration is 0 ns
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   160
                    durationNanos++;
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   161
                }
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   162
                path = repositoryFiles.nextPath(currentChunkStartNanos + durationNanos);
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   163
                if (path == null) {
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   164
                    return; // stream closed
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   165
                }
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   166
                currentChunkStartNanos = repositoryFiles.getTimestamp(path);
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   167
                input.setFile(path);
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   168
                currentParser = currentParser.newChunkParser();
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   169
                // TODO: Optimization. No need filter when we reach new chunk
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   170
                // Could set start = 0;
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   171
            }
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   172
        }
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   173
    }
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   174
59226
a0f39cc47387 8233700: EventStream not closed
egahlin
parents: 58863
diff changeset
   175
    private boolean isLastChunk() {
a0f39cc47387 8233700: EventStream not closed
egahlin
parents: 58863
diff changeset
   176
        if (recording == null) {
a0f39cc47387 8233700: EventStream not closed
egahlin
parents: 58863
diff changeset
   177
            return false;
a0f39cc47387 8233700: EventStream not closed
egahlin
parents: 58863
diff changeset
   178
        }
a0f39cc47387 8233700: EventStream not closed
egahlin
parents: 58863
diff changeset
   179
        return recording.getFinalChunkStartNanos() >= currentParser.getStartNanos();
a0f39cc47387 8233700: EventStream not closed
egahlin
parents: 58863
diff changeset
   180
    }
a0f39cc47387 8233700: EventStream not closed
egahlin
parents: 58863
diff changeset
   181
58863
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   182
    private boolean processOrdered(Dispatcher c, boolean awaitNewEvents) throws IOException {
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   183
        if (sortedCache == null) {
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   184
            sortedCache = new RecordedEvent[100_000];
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   185
        }
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   186
        int index = 0;
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   187
        while (true) {
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   188
            RecordedEvent e = currentParser.readStreamingEvent(awaitNewEvents);
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   189
            if (e == null) {
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   190
                // wait for new event with next call to
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   191
                // readStreamingEvent()
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   192
                awaitNewEvents = true;
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   193
                break;
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   194
            }
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   195
            awaitNewEvents = false;
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   196
            if (index == sortedCache.length) {
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   197
                sortedCache = Arrays.copyOf(sortedCache, sortedCache.length * 2);
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   198
            }
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   199
            sortedCache[index++] = e;
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   200
        }
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   201
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   202
        // no events found
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   203
        if (index == 0 && currentParser.isChunkFinished()) {
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   204
            return awaitNewEvents;
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   205
        }
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   206
        // at least 2 events, sort them
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   207
        if (index > 1) {
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   208
            Arrays.sort(sortedCache, 0, index, EVENT_COMPARATOR);
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   209
        }
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   210
        for (int i = 0; i < index; i++) {
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   211
            c.dispatch(sortedCache[i]);
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   212
        }
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   213
        return awaitNewEvents;
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   214
    }
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   215
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   216
    private boolean processUnordered(Dispatcher c, boolean awaitNewEvents) throws IOException {
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   217
        while (true) {
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   218
            RecordedEvent e = currentParser.readStreamingEvent(awaitNewEvents);
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   219
            if (e == null) {
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   220
                return true;
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   221
            } else {
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   222
                c.dispatch(e);
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   223
            }
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   224
        }
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   225
    }
59226
a0f39cc47387 8233700: EventStream not closed
egahlin
parents: 58863
diff changeset
   226
58863
c16ac7a2eba4 8226511: Implement JFR Event Streaming
mgronlun
parents:
diff changeset
   227
}