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

    35 import java.util.Objects;

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

    37 import java.util.function.Consumer;

    38 

    39 import jdk.jfr.internal.JVM;

    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. In particular, it handles:

    48  *

    49  * - configuration storage

    50  * - atomic updates to a configuration

    51  * - dispatch mechanism

    52  * - error handling

    53  * - security

    54  *

    55  */

    56 abstract class AbstractEventStream implements EventStream {

    57 

    58     static final Comparator<? super RecordedEvent> END_TIME = (e1, e2) -> Long.compare(e1.endTimeTicks, e2.endTimeTicks);

    59 

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

    61     private final Object terminated = new Object();

    62     private final boolean active;

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

    64     private final AccessControlContext accessControllerContext;

    65     private final StreamConfiguration configuration = new StreamConfiguration();

    66 

    67     private volatile Thread thread;

    68     private Dispatcher dispatcher;

    69 

    70     private volatile boolean closed;

    71 

    72     public AbstractEventStream(AccessControlContext acc, boolean active) throws IOException {

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

    74         this.active = active;

    75     }

    76 

    77     @Override

    78     abstract public void start();

    79 

    80     @Override

    81     abstract public void startAsync();

    82 

    83     @Override

    84     abstract public void close();

    85 

    86     protected final Dispatcher dispatcher() {

    87         if (configuration.hasChanged()) {

    88             synchronized (configuration) {

    89                 dispatcher = new Dispatcher(configuration);

    90             }

    91         }

    92         return dispatcher;

    93     }

    94 

    95     @Override

    96     public final void setOrdered(boolean ordered) {

    97         configuration.setOrdered(ordered);

    98     }

    99 

   100     @Override

   101     public final void setReuse(boolean reuse) {

   102         configuration.setReuse(reuse);

   103     }

   104 

   105     @Override

   106     public final void setStartTime(Instant startTime) {

   107         Objects.nonNull(startTime);

   108         synchronized (configuration) {

   109             if (configuration.started) {

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

   111             }

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

   113                 startTime = Instant.EPOCH;

   114             }

   115             configuration.setStartTime(startTime);

   116         }

   117     }

   118 

   119     @Override

   120     public final void setEndTime(Instant endTime) {

   121         Objects.requireNonNull(endTime);

   122         synchronized (configuration) {

   123             if (configuration.started) {

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

   125             }

   126             configuration.setEndTime(endTime);

   127         }

   128     }

   129 

   130     @Override

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

   132         Objects.requireNonNull(action);

   133         configuration.addEventAction(action);

   134     }

   135 

   136     @Override

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

   138         Objects.requireNonNull(eventName);

   139         Objects.requireNonNull(action);

   140         configuration.addEventAction(eventName, action);

   141     }

   142 

   143     @Override

   144     public final void onFlush(Runnable action) {

   145         Objects.requireNonNull(action);

   146         configuration.addFlushAction(action);

   147     }

   148 

   149     @Override

   150     public final void onClose(Runnable action) {

   151         Objects.requireNonNull(action);

   152         configuration.addCloseAction(action);

   153     }

   154 

   155     @Override

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

   157         Objects.requireNonNull(action);

   158         configuration.addErrorAction(action);

   159     }

   160 

   161     @Override

   162     public final boolean remove(Object action) {

   163         Objects.requireNonNull(action);

   164         return configuration.remove(action);

   165     }

   166 

   167     @Override

   168     public final void awaitTermination() throws InterruptedException {

   169         awaitTermination(Duration.ofMillis(0));

   170     }

   171 

   172     @Override

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

   174         Objects.requireNonNull(timeout);

   175         if (timeout.isNegative()) {

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

   177         }

   178 

   179         long base = System.currentTimeMillis();

   180         long now = 0;

   181 

   182         long millis;

   183         try {

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

   185         } catch (ArithmeticException a) {

   186             millis = Long.MAX_VALUE;

   187         }

   188         int nanos = timeout.toNanosPart();

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

   190             synchronized (terminated) {

   191                 while (!isClosed()) {

   192                     terminated.wait(0);

   193                 }

   194             }

   195         } else {

   196             while (!isClosed()) {

   197                 long delay = millis - now;

   198                 if (delay <= 0) {

   199                     break;

   200                 }

   201                 synchronized (terminated) {

   202                     terminated.wait(delay, nanos);

   203                 }

   204                 now = System.currentTimeMillis() - base;

   205             }

   206         }

   207     }

   208 

   209     protected abstract void process() throws Exception;

   210 

   211     protected final void setClosed(boolean closed) {

   212         this.closed = closed;

   213     }

   214 

   215     protected final boolean isClosed() {

   216         return closed;

   217     }

   218 

   219     protected final void startAsync(long startNanos) {

   220         startInternal(startNanos);

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

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

   223         thread.start();

   224     }

   225 

   226     protected final void start(long startNanos) {

   227         startInternal(startNanos);

   228         thread = Thread.currentThread();

   229         run(accessControllerContext);

   230     }

   231 

   232     protected final Runnable getFlushOperation() {

   233         return flushOperation;

   234     }

   235 

   236     private void startInternal(long startNanos) {

   237         synchronized (configuration) {

   238             if (configuration.started) {

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

   240             }

   241             if (active) {

   242                 configuration.setStartNanos(startNanos);

   243             }

   244             configuration.setStarted(true);

   245         }

   246     }

   247 

   248     private void execute() {

   249         JVM.getJVM().exclude(Thread.currentThread());

   250         try {

   251             process();

   252         } catch (IOException ioe) {

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

   254             // a file is access that has been closed

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

   256             // stream will be closed when this happens

   257         } catch (Exception e) {

   258             // TODO: Remove before integrating

   259             e.printStackTrace();

   260         } finally {

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

   262             try {

   263                 close();

   264             } finally {

   265                 synchronized (terminated) {

   266                     terminated.notifyAll();

   267                 }

   268             }

   269         }

   270     }

   271 

   272     private void run(AccessControlContext accessControlContext) {

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

   274             @Override

   275             public Void run() {

   276                 execute();

   277                 return null;

   278             }

   279         }, accessControlContext);

   280     }

   281 

   282     private String nextThreadName() {

   283         counter.incrementAndGet();

   284         return "JFR Event Stream " + counter;

   285     }

   286 }