1 /*

     2  * Copyright (c) 2019, Oracle and/or its affiliates. All rights reserved.

     3  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.

     4  *

     5  * This code is free software; you can redistribute it and/or modify it

     6  * under the terms of the GNU General Public License version 2 only, as

     7  * published by the Free Software Foundation.  Oracle designates this

     8  * particular file as subject to the "Classpath" exception as provided

     9  * by Oracle in the LICENSE file that accompanied this code.

    10  *

    11  * This code is distributed in the hope that it will be useful, but WITHOUT

    12  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or

    13  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License

    14  * version 2 for more details (a copy is included in the LICENSE file that

    15  * accompanied this code).

    16  *

    17  * You should have received a copy of the GNU General Public License version

    18  * 2 along with this work; if not, write to the Free Software Foundation,

    19  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.

    20  *

    21  * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA

    22  * or visit www.oracle.com if you need additional information or have any

    23  * questions.

    24  */

    25 

    26 package jdk.jfr.internal.consumer;

    27 

    28 import java.io.IOException;

    29 import java.nio.file.Path;

    30 import java.security.AccessControlContext;

    31 import java.util.Arrays;

    32 import java.util.Comparator;

    33 import java.util.Objects;

    34 

    35 import jdk.jfr.consumer.RecordedEvent;

    36 import jdk.jfr.internal.consumer.Dispatcher;

    37 import jdk.jfr.internal.consumer.FileAccess;

    38 import jdk.jfr.internal.consumer.RecordingInput;

    39 

    40 /**

    41  * Implementation of an event stream that operates against a recording file.

    42  *

    43  */

    44 public final class EventFileStream extends AbstractEventStream {

    45     private final static Comparator<? super RecordedEvent> EVENT_COMPARATOR = JdkJfrConsumer.instance().eventComparator();

    46 

    47     private final RecordingInput input;

    48     private ChunkParser chunkParser;

    49     private RecordedEvent[] sortedList;

    50 

    51     public EventFileStream(AccessControlContext acc, Path path) throws IOException {

    52         super(acc, false);

    53         Objects.requireNonNull(path);

    54         this.input = new RecordingInput(path.toFile(), FileAccess.UNPRIVILIGED);

    55     }

    56 

    57     @Override

    58     public void start() {

    59         start(0);

    60     }

    61 

    62     @Override

    63     public void startAsync() {

    64         startAsync(0);

    65     }

    66 

    67     @Override

    68     public void close() {

    69         setClosed(true);

    70         dispatcher().runCloseActions();

    71         try {

    72             input.close();

    73         } catch (IOException e) {

    74             // ignore

    75         }

    76     }

    77 

    78     @Override

    79     protected void process() throws IOException {

    80         Dispatcher disp = dispatcher();

    81         long start = 0;

    82         long end = Long.MAX_VALUE;

    83         if (disp.startTime != null) {

    84             start = disp.startNanos;

    85         }

    86         if (disp.endTime != null) {

    87             end = disp.endNanos;

    88         }

    89 

    90         chunkParser = new ChunkParser(input, disp.parserConfiguration);

    91         while (!isClosed()) {

    92             if (chunkParser.getStartNanos() > end) {

    93                 close();

    94                 return;

    95             }

    96             disp = dispatcher();

    97             disp.parserConfiguration.filterStart = start;

    98             disp.parserConfiguration.filterEnd = end;

    99             chunkParser.updateConfiguration(disp.parserConfiguration, true);

   100             chunkParser.setFlushOperation(getFlushOperation());

   101             if (disp.parserConfiguration.ordered) {

   102                 processOrdered(disp);

   103             } else {

   104                 processUnordered(disp);

   105             }

   106             if (isClosed() || chunkParser.isLastChunk()) {

   107                 return;

   108             }

   109             chunkParser = chunkParser.nextChunkParser();

   110         }

   111     }

   112 

   113     private void processOrdered(Dispatcher c) throws IOException {

   114         if (sortedList == null) {

   115             sortedList = new RecordedEvent[10_000];

   116         }

   117         RecordedEvent event;

   118         int index = 0;

   119         while (true) {

   120             event = chunkParser.readEvent();

   121             if (event == null) {

   122                 Arrays.sort(sortedList, 0, index, EVENT_COMPARATOR);

   123                 for (int i = 0; i < index; i++) {

   124                     c.dispatch(sortedList[i]);

   125                 }

   126                 return;

   127             }

   128             if (index == sortedList.length) {

   129                 RecordedEvent[] tmp = sortedList;

   130                 sortedList = new RecordedEvent[2 * tmp.length];

   131                 System.arraycopy(tmp, 0, sortedList, 0, tmp.length);

   132             }

   133             sortedList[index++] = event;

   134         }

   135     }

   136 

   137     private void processUnordered(Dispatcher c) throws IOException {

   138         while (!isClosed()) {

   139             RecordedEvent event = chunkParser.readEvent();

   140             if (event == null) {

   141                 return;

   142             }

   143             c.dispatch(event);

   144         }

   145     }

   146 }