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.consumer;

    27 

    28 import java.io.IOException;

    29 import java.nio.file.Path;

    30 import java.time.Instant;

    31 import java.util.Arrays;

    32 import java.util.HashSet;

    33 import java.util.Set;

    34 import java.util.concurrent.TimeUnit;

    35 import java.util.concurrent.atomic.AtomicInteger;

    36 import java.util.concurrent.locks.ReentrantLock;

    37 

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

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

    40 

    41 /**

    42  * Cache that represents all discovered events in a chunk.

    43  *

    44  */

    45 final class EventSet {

    46 

    47     public static final RecordedEvent[] END_OF_SET = new RecordedEvent[0];

    48     private static final AtomicInteger idCounter = new AtomicInteger(-1);

    49 

    50     private volatile Object[][] segments = new Object[1000][];

    51     private volatile boolean closed;

    52     private final long startTimeNanos;

    53     private final EventSetLocation location;

    54     private final Path path;

    55     private final int id;

    56 

    57     // Guarded by lock

    58     private boolean awaitNewEvents;

    59     private RecordingInput input;

    60     private ChunkParser chunkParser;

    61     private int referenceCount;

    62     private final ReentrantLock lock = new ReentrantLock();

    63     private final Set<InternalEventFilter> filters = new HashSet<>();

    64     private InternalEventFilter globalFilter = InternalEventFilter.ACCEPT_ALL;

    65     private boolean dirtyFilter = true;

    66 

    67     public void release(InternalEventFilter eventFilter) {

    68         try {

    69             lock.lock();

    70             filters.remove(eventFilter);

    71             updateGlobalFilter();

    72             referenceCount--;

    73             if (referenceCount == 0) {

    74                 closed = true;

    75                 if (input != null) {

    76                     try {

    77                         input.close();

    78                     } catch (IOException e) {

    79                         // TODO: Flie locked by other process?

    80                     }

    81                     chunkParser = null;

    82                     input = null;

    83                 }

    84             }

    85         } finally {

    86            lock.unlock();

    87         }

    88     }

    89 

    90     public EventSet(EventSetLocation location, EventSet previousEventSet, Path p) throws IOException {

    91         this.location = location;

    92         this.path = p;

    93         this.startTimeNanos = readStartTime(p);

    94         this.id = idCounter.incrementAndGet();

    95     }

    96 

    97     private long readStartTime(Path p) throws IOException {

    98         try (RecordingInput in = new RecordingInput(p.toFile(), 100)) {

    99             ChunkHeader c = new ChunkHeader(in);

   100             return c.getStartNanos();

   101         }

   102     }

   103 

   104     Path getPath() {

   105         return path;

   106     }

   107 

   108     // TODO: Use binary search, must use lock

   109     public int findIndex(Instant timestamp) {

   110         int index = 0;

   111         for (int i = 0; i < segments.length; i++) {

   112             RecordedEvent[] events = (RecordedEvent[]) segments[i];

   113             if (events == null || events.length == 0) {

   114                 return Math.max(index - 1, 0);

   115             }

   116             RecordedEvent e = events[0]; // May not be sorted.

   117             if (timestamp.isAfter(e.getEndTime())) {

   118                 return Math.max(index - 1, 0);

   119             }

   120         }

   121         return segments.length;

   122     }

   123 

   124     public void addFilter(InternalEventFilter filter) {

   125         try {

   126             lock.lock();

   127             filters.add(filter);

   128             updateGlobalFilter();

   129         } finally {

   130             lock.unlock();

   131         }

   132     }

   133 

   134     // held with lock

   135     private void updateGlobalFilter() {

   136         globalFilter = InternalEventFilter.merge(filters);

   137         dirtyFilter = true;

   138     }

   139 

   140     public RecordedEvent[] readEvents(int index) throws Exception {

   141         while (!closed) {

   142 

   143             RecordedEvent[] events = (RecordedEvent[]) segments[index];

   144             if (events != null) {

   145                 return events;

   146             }

   147             if (lock.tryLock(250, TimeUnit.MILLISECONDS)) {

   148                 try {

   149                     addSegment(index);

   150                 } finally {

   151                     lock.unlock();

   152                 }

   153             }

   154         }

   155         return null;

   156     }

   157 

   158     // held with lock

   159     private void addSegment(int index) throws IOException {

   160         if (chunkParser == null) {

   161             chunkParser = new ChunkParser(new RecordingInput(path.toFile()));

   162         }

   163         if (dirtyFilter) {

   164             chunkParser.setParserFilter(globalFilter);

   165         }

   166         if (segments[index] != null) {

   167             return;

   168         }

   169         if (index == segments.length - 2) {

   170             segments = Arrays.copyOf(segments, segments.length * 2);

   171         }

   172         RecordedEvent[] segment = new RecordedEvent[10];

   173         int i = 0;

   174         while (true) {

   175             RecordedEvent e = chunkParser.readStreamingEvent(awaitNewEvents);

   176             if (e == null) {

   177                 // wait for new event with next call to readStreamingEvent()

   178                 awaitNewEvents = true;

   179                 break;

   180             }

   181             awaitNewEvents = false;

   182             if (i == segment.length) {

   183                 segment = Arrays.copyOf(segment, segment.length * 2);

   184             }

   185             segment[i++] = e;

   186         }

   187 

   188         // no events found

   189         if (i == 0) {

   190             if (chunkParser.isChunkFinished()) {

   191                 segments[index] = END_OF_SET;

   192                 return;

   193             }

   194         }

   195         // at least 2 events, sort them

   196         if (i > 1) {

   197             Arrays.sort(segment, 0, i, (e1, e2) -> Long.compare(e1.endTime, e2.endTime));

   198         }

   199         segments[index] = segment;

   200         if (chunkParser.isChunkFinished()) {

   201             segments[index + 1] = END_OF_SET;

   202         }

   203     }

   204 

   205     public long getStartTimeNanos() {

   206         return startTimeNanos;

   207     }

   208 

   209     public EventSet next(InternalEventFilter filter) throws IOException {

   210         EventSet next = location.acquire(startTimeNanos + 1, this);

   211         if (next == null) {

   212             // closed

   213             return null;

   214         }

   215         next.addFilter(filter);

   216         release(filter);

   217         return next;

   218     }

   219 

   220     public void acquire() {

   221         try {

   222             lock.lock();

   223             referenceCount++;

   224         } finally {

   225             lock.unlock();

   226         }

   227     }

   228 

   229     public String toString() {

   230         return "Chunk:" + id + " (" + path + ")";

   231     }

   232 }