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.security.AccessControlContext;

    30 import java.security.AccessController;

    31 import java.security.PrivilegedAction;

    32 import java.time.Duration;

    33 import java.time.Instant;

    34 import java.util.Objects;

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

    36 import java.util.function.Consumer;

    37 

    38 import jdk.jfr.consumer.EventStream;

    39 import jdk.jfr.consumer.RecordedEvent;

    40 import jdk.jfr.internal.LogLevel;

    41 import jdk.jfr.internal.LogTag;

    42 import jdk.jfr.internal.Logger;

    43 import jdk.jfr.internal.SecuritySupport;

    44 

    45 /*

    46  * Purpose of this class is to simplify the implementation of

    47  * an event stream.

    48  */

    49 abstract class AbstractEventStream implements EventStream {

    50     private final static AtomicLong counter = new AtomicLong(1);

    51 

    52     private final Object terminated = new Object();

    53     private final boolean active;

    54     private final Runnable flushOperation = () -> dispatcher().runFlushActions();

    55     private final AccessControlContext accessControllerContext;

    56     private final StreamConfiguration configuration = new StreamConfiguration();

    57 

    58     private volatile Thread thread;

    59     private Dispatcher dispatcher;

    60 

    61     private volatile boolean closed;

    62 

    63     AbstractEventStream(AccessControlContext acc, boolean active) throws IOException {

    64         this.accessControllerContext = Objects.requireNonNull(acc);

    65         this.active = active;

    66     }

    67 

    68     @Override

    69     abstract public void start();

    70 

    71     @Override

    72     abstract public void startAsync();

    73 

    74     @Override

    75     abstract public void close();

    76 

    77     protected final Dispatcher dispatcher() {

    78         if (configuration.hasChanged()) {

    79             synchronized (configuration) {

    80                 dispatcher = new Dispatcher(configuration);

    81             }

    82         }

    83         return dispatcher;

    84     }

    85 

    86     @Override

    87     public final void setOrdered(boolean ordered) {

    88         configuration.setOrdered(ordered);

    89     }

    90 

    91     @Override

    92     public final void setReuse(boolean reuse) {

    93         configuration.setReuse(reuse);

    94     }

    95 

    96     @Override

    97     public final void setStartTime(Instant startTime) {

    98         Objects.nonNull(startTime);

    99         synchronized (configuration) {

   100             if (configuration.started) {

   101                 throw new IllegalStateException("Stream is already started");

   102             }

   103             if (startTime.isBefore(Instant.EPOCH)) {

   104                 startTime = Instant.EPOCH;

   105             }

   106             configuration.setStartTime(startTime);

   107         }

   108     }

   109 

   110     @Override

   111     public final void setEndTime(Instant endTime) {

   112         Objects.requireNonNull(endTime);

   113         synchronized (configuration) {

   114             if (configuration.started) {

   115                 throw new IllegalStateException("Stream is already started");

   116             }

   117             configuration.setEndTime(endTime);

   118         }

   119     }

   120 

   121     @Override

   122     public final void onEvent(Consumer<RecordedEvent> action) {

   123         Objects.requireNonNull(action);

   124         configuration.addEventAction(action);

   125     }

   126 

   127     @Override

   128     public final void onEvent(String eventName, Consumer<RecordedEvent> action) {

   129         Objects.requireNonNull(eventName);

   130         Objects.requireNonNull(action);

   131         configuration.addEventAction(eventName, action);

   132     }

   133 

   134     @Override

   135     public final void onFlush(Runnable action) {

   136         Objects.requireNonNull(action);

   137         configuration.addFlushAction(action);

   138     }

   139 

   140     @Override

   141     public final void onClose(Runnable action) {

   142         Objects.requireNonNull(action);

   143         configuration.addCloseAction(action);

   144     }

   145 

   146     @Override

   147     public final void onError(Consumer<Throwable> action) {

   148         Objects.requireNonNull(action);

   149         configuration.addErrorAction(action);

   150     }

   151 

   152     @Override

   153     public final boolean remove(Object action) {

   154         Objects.requireNonNull(action);

   155         return configuration.remove(action);

   156     }

   157 

   158     @Override

   159     public final void awaitTermination() throws InterruptedException {

   160         awaitTermination(Duration.ofMillis(0));

   161     }

   162 

   163     @Override

   164     public final void awaitTermination(Duration timeout) throws InterruptedException {

   165         Objects.requireNonNull(timeout);

   166         if (timeout.isNegative()) {

   167             throw new IllegalArgumentException("timeout value is negative");

   168         }

   169 

   170         long base = System.currentTimeMillis();

   171         long now = 0;

   172 

   173         long millis;

   174         try {

   175             millis = Math.multiplyExact(timeout.getSeconds(), 1000);

   176         } catch (ArithmeticException a) {

   177             millis = Long.MAX_VALUE;

   178         }

   179         int nanos = timeout.toNanosPart();

   180         if (nanos == 0 && millis == 0) {

   181             synchronized (terminated) {

   182                 while (!isClosed()) {

   183                     terminated.wait(0);

   184                 }

   185             }

   186         } else {

   187             while (!isClosed()) {

   188                 long delay = millis - now;

   189                 if (delay <= 0) {

   190                     break;

   191                 }

   192                 synchronized (terminated) {

   193                     terminated.wait(delay, nanos);

   194                 }

   195                 now = System.currentTimeMillis() - base;

   196             }

   197         }

   198     }

   199 

   200     protected abstract void process() throws IOException;

   201 

   202     protected final void setClosed(boolean closed) {

   203         this.closed = closed;

   204     }

   205 

   206     protected final boolean isClosed() {

   207         return closed;

   208     }

   209 

   210     public final void startAsync(long startNanos) {

   211         startInternal(startNanos);

   212         Runnable r = () -> run(accessControllerContext);

   213         thread = SecuritySupport.createThreadWitNoPermissions(nextThreadName(), r);

   214         thread.start();

   215     }

   216 

   217     public final void start(long startNanos) {

   218         startInternal(startNanos);

   219         thread = Thread.currentThread();

   220         run(accessControllerContext);

   221     }

   222 

   223     protected final Runnable getFlushOperation() {

   224         return flushOperation;

   225     }

   226 

   227     private void startInternal(long startNanos) {

   228         synchronized (configuration) {

   229             if (configuration.started) {

   230                 throw new IllegalStateException("Event stream can only be started once");

   231             }

   232             if (active && configuration.startTime == null) {

   233                 configuration.setStartNanos(startNanos);

   234             }

   235             configuration.setStarted(true);

   236         }

   237     }

   238 

   239     private void execute() {

   240         try {

   241             process();

   242         } catch (IOException ioe) {

   243             // This can happen if a chunk file is removed, or

   244             // a file is access that has been closed

   245             // This is "normal" behavior for streaming and the

   246             // stream will be closed when this happens.

   247         } finally {

   248             Logger.log(LogTag.JFR_SYSTEM_STREAMING, LogLevel.DEBUG, "Execution of stream ended.");

   249             try {

   250                 close();

   251             } finally {

   252                 synchronized (terminated) {

   253                     terminated.notifyAll();

   254                 }

   255             }

   256         }

   257     }

   258 

   259     private void run(AccessControlContext accessControlContext) {

   260         AccessController.doPrivileged(new PrivilegedAction<Void>() {

   261             @Override

   262             public Void run() {

   263                 execute();

   264                 return null;

   265             }

   266         }, accessControlContext);

   267     }

   268 

   269     private String nextThreadName() {

   270         counter.incrementAndGet();

   271         return "JFR Event Stream " + counter;

   272     }

   273 }