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

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

    41 

    42 /**

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

    44  *

    45  */

    46 final class EventSet {

    47 

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

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

    50 

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

    52     private volatile boolean closed;

    53     private final long startTimeNanos;

    54     private final EventSetLocation location;

    55     private final Path path;

    56     private final int id;

    57 

    58     // Guarded by lock

    59     private boolean awaitNewEvents;

    60     private RecordingInput input;

    61     private ChunkParser chunkParser;

    62     private int referenceCount;

    63     private final ReentrantLock lock = new ReentrantLock();

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

    65     private InternalEventFilter globalFilter = InternalEventFilter.ACCEPT_ALL;

    66     private boolean dirtyFilter = true;

    67 

    68     public void release(InternalEventFilter eventFilter) {

    69         try {

    70             lock.lock();

    71             filters.remove(eventFilter);

    72             updateGlobalFilter();

    73             referenceCount--;

    74             if (referenceCount == 0) {

    75                 closed = true;

    76                 if (input != null) {

    77                     try {

    78                         input.close();

    79                     } catch (IOException e) {

    80                         // TODO: Flie locked by other process?

    81                     }

    82                     chunkParser = null;

    83                     input = null;

    84                 }

    85             }

    86         } finally {

    87            lock.unlock();

    88         }

    89     }

    90 

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

    92         this.location = location;

    93         this.path = p;

    94         this.startTimeNanos = readStartTime(p);

    95         this.id = idCounter.incrementAndGet();

    96     }

    97 

    98     private long readStartTime(Path p) throws IOException {

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

   100             ChunkHeader c = new ChunkHeader(in);

   101             return c.getStartNanos();

   102         }

   103     }

   104 

   105     Path getPath() {

   106         return path;

   107     }

   108 

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

   110     public int findIndex(Instant timestamp) {

   111         int index = 0;

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

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

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

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

   116             }

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

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

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

   120             }

   121         }

   122         return segments.length;

   123     }

   124 

   125     public void addFilter(InternalEventFilter filter) {

   126         try {

   127             lock.lock();

   128             filters.add(filter);

   129             updateGlobalFilter();

   130         } finally {

   131             lock.unlock();

   132         }

   133     }

   134 

   135     // held with lock

   136     private void updateGlobalFilter() {

   137         globalFilter = InternalEventFilter.merge(filters);

   138         dirtyFilter = true;

   139     }

   140 

   141     public RecordedEvent[] readEvents(int index) throws IOException {

   142         while (!closed) {

   143 

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

   145             if (events != null) {

   146                 return events;

   147             }

   148             if (await()) {

   149                 try {

   150                     addSegment(index);

   151                 } finally {

   152                     lock.unlock();

   153                 }

   154             }

   155         }

   156         return null;

   157     }

   158 

   159     private boolean await()  {

   160         try {

   161             return lock.tryLock(250, TimeUnit.MILLISECONDS);

   162         } catch (InterruptedException e) {

   163             return false;

   164         }

   165     }

   166 

   167     // held with lock

   168     private void addSegment(int index) throws IOException {

   169         if (chunkParser == null) {

   170             chunkParser = new ChunkParser(new RecordingInput(path.toFile()), false);

   171         }

   172         if (dirtyFilter) {

   173             chunkParser.setParserFilter(globalFilter);

   174         }

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

   176             return;

   177         }

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

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

   180         }

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

   182         int i = 0;

   183         while (true) {

   184             RecordedEvent e = chunkParser.readStreamingEvent(awaitNewEvents);

   185             if (e == null) {

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

   187                 awaitNewEvents = true;

   188                 break;

   189             }

   190             awaitNewEvents = false;

   191             if (i == segment.length) {

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

   193             }

   194             segment[i++] = e;

   195         }

   196 

   197         // no events found

   198         if (i == 0) {

   199             if (chunkParser.isChunkFinished()) {

   200                 segments[index] = END_OF_SET;

   201                 return;

   202             }

   203         }

   204         // at least 2 events, sort them

   205         if (i > 1) {

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

   207         }

   208         segments[index] = segment;

   209         if (chunkParser.isChunkFinished()) {

   210             segments[index + 1] = END_OF_SET;

   211         }

   212     }

   213 

   214     public long getStartTimeNanos() {

   215         return startTimeNanos;

   216     }

   217 

   218     public EventSet next(InternalEventFilter filter) throws IOException {

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

   220         if (next == null) {

   221             // closed

   222             return null;

   223         }

   224         next.addFilter(filter);

   225         release(filter);

   226         return next;

   227     }

   228 

   229     public void acquire() {

   230         try {

   231             lock.lock();

   232             referenceCount++;

   233         } finally {

   234             lock.unlock();

   235         }

   236     }

   237 

   238     public String toString() {

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

   240     }

   241 }