8248564: JFR: Remote Recording Stream

Reviewed-by: mgronlun

src/hotspot/share/jfr/recorder/repository/jfrChunkWriter.cpp

@@ -103,8 +103,8 @@ class JfrChunkHeadWriter : public StackObj {
     _writer->be_write(PAD);
   }
 
-  void write_next_generation() {
-    _writer->be_write(_chunk->next_generation());
+  void write_next_generation(bool finalize) {
+    _writer->be_write(finalize ? COMPLETE : _chunk->next_generation());
     _writer->be_write(PAD);
   }
 
@@ -199,9 +199,9 @@ int64_t JfrChunkWriter::write_chunk_header_checkpoint(bool flushpoint) {
   const int64_t chunk_size_offset = reserve(sizeof(int64_t)); // size to be decided when we are done
   be_write(event_size_offset); // last checkpoint offset will be this checkpoint
   head.write_metadata();
-  head.write_time(false);
+  head.write_time(!flushpoint);
   head.write_cpu_frequency();
-  head.write_next_generation();
+  head.write_next_generation(!flushpoint);
   head.write_flags();
   assert(current_offset() - header_content_pos == HEADER_SIZE, "invariant");
   const u4 checkpoint_size = current_offset() - event_size_offset;

src/jdk.jfr/share/classes/jdk/jfr/EventSettings.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2016, 2018, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2016, 2020, Oracle and/or its affiliates. All rights reserved.
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
  *
  * This code is free software; you can redistribute it and/or modify it
@@ -28,6 +28,8 @@
 import java.time.Duration;
 import java.util.Map;
 
+import jdk.jfr.internal.management.EventSettingsModifier;
+
 /**
  * Convenience class for applying event settings to a recording.
  * <p>
@@ -55,6 +57,26 @@
  */
 public abstract class EventSettings {
 
+    // Used to provide EventSettings for jdk.management.jfr module
+    static class DelegatedEventSettings extends EventSettings {
+        private final EventSettingsModifier delegate;
+
+        DelegatedEventSettings(EventSettingsModifier modifier) {
+            this.delegate = modifier;
+        }
+
+        @Override
+        public EventSettings with(String name, String value) {
+             delegate.with(name, value);
+             return this;
+        }
+
+        @Override
+        Map<String, String> toMap() {
+            return delegate.toMap();
+        }
+    }
+
     // package private
     EventSettings() {
     }

src/jdk.jfr/share/classes/jdk/jfr/FlightRecorderPermission.java

@@ -37,6 +37,7 @@
 import jdk.jfr.internal.PrivateAccess;
 import jdk.jfr.internal.Type;
 import jdk.jfr.internal.Utils;
+import jdk.jfr.internal.management.EventSettingsModifier;
 
 /**
  * Permission for controlling access to Flight Recorder.
@@ -197,6 +198,11 @@ public PlatformRecorder getPlatformRecorder() {
         public AccessControlContext getContext(SettingControl settingControl) {
             return settingControl.getContext();
         }
+
+        @Override
+        public EventSettings newEventSettings(EventSettingsModifier esm) {
+            return new EventSettings.DelegatedEventSettings(esm);
+        }
     }
 
     /**

src/jdk.jfr/share/classes/jdk/jfr/consumer/EventStream.java

@@ -31,6 +31,7 @@
 import java.security.AccessController;
 import java.time.Duration;
 import java.time.Instant;
+import java.util.Collections;
 import java.util.Objects;
 import java.util.function.Consumer;
 
@@ -137,7 +138,7 @@ public interface EventStream extends AutoCloseable {
      */
     public static EventStream openRepository() throws IOException {
         Utils.checkAccessFlightRecorder();
-        return new EventDirectoryStream(AccessController.getContext(), null, SecuritySupport.PRIVILEGED, null);
+        return new EventDirectoryStream(AccessController.getContext(), null, SecuritySupport.PRIVILEGED, null, Collections.emptyList());
     }
 
     /**
@@ -160,7 +161,7 @@ public static EventStream openRepository() throws IOException {
     public static EventStream openRepository(Path directory) throws IOException {
         Objects.nonNull(directory);
         AccessControlContext acc = AccessController.getContext();
-        return new EventDirectoryStream(acc, directory, FileAccess.UNPRIVILEGED, null);
+        return new EventDirectoryStream(acc, directory, FileAccess.UNPRIVILEGED, null, Collections.emptyList());
     }
 
     /**
@@ -182,6 +183,22 @@ static EventStream openFile(Path file) throws IOException {
         return new EventFileStream(AccessController.getContext(), file);
     }
 
+    /**
+     * Registers an action to perform when new metadata arrives in the stream.
+     *
+     * The event type of an event always arrives sometime before the actual event.
+     * The action must be registered before the stream is started.
+     *
+     * @implSpec The default implementation of this method is empty.
+     *
+     * @param action to perform, not {@code null}
+     *
+     * @throws IllegalStateException if an action is added after the stream has
+     *                               started
+     */
+     default void onMetadata(Consumer<MetadataEvent> action) {
+     }
+
     /**
      * Registers an action to perform on all events in the stream.
      *

src/jdk.jfr/share/classes/jdk/jfr/consumer/MetadataEvent.java

@@ -0,0 +1,118 @@
+/*
+ * Copyright (c) 2020, Oracle and/or its affiliates. All rights reserved.
+ * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
+ *
+ * This code is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License version 2 only, as
+ * published by the Free Software Foundation.  Oracle designates this
+ * particular file as subject to the "Classpath" exception as provided
+ * by Oracle in the LICENSE file that accompanied this code.
+ *
+ * This code is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
+ * version 2 for more details (a copy is included in the LICENSE file that
+ * accompanied this code).
+ *
+ * You should have received a copy of the GNU General Public License version
+ * 2 along with this work; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
+ *
+ * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
+ * or visit www.oracle.com if you need additional information or have any
+ * questions.
+ */
+package jdk.jfr.consumer;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import jdk.jfr.Configuration;
+import jdk.jfr.EventType;
+
+/**
+ * Event that contains information about event types and configurations.
+ *
+ * @since 16
+ */
+public final class MetadataEvent {
+    private final List<EventType> current;
+    private final List<EventType> previous;
+    private final List<Configuration> configurations;
+    private List<EventType> added;
+    private List<EventType> removed;
+
+    /* package private */
+    MetadataEvent(List<EventType> previous, List<EventType> current, List<Configuration> configs) {
+        this.previous = previous;
+        this.current = current;
+        this.configurations = configs;
+    }
+
+    /**
+     * Returns a list of the current event types being used.
+     *
+     * @return an immutable list of event types, not {@code null}
+     */
+    public final List<EventType> getEventTypes() {
+        return Collections.unmodifiableList(current);
+    }
+
+    /**
+     * Returns a list of added event types since the last metadata event.
+     * <p>
+     * The delta will be from the last metadata event. If no metadata event has been
+     * emitted earlier, all known event types will be in the list.
+     *
+     * @return an immutable list of added event types, not {@code null}
+     */
+    public final List<EventType> getAddedEventTypes() {
+        if (added == null) {
+            calculateDelta();
+        }
+        return added;
+    }
+
+    /**
+     * Returns a list of removed event types since the last metadata event.
+     * <p>
+     * The delta will be from the last metadata event. If no metadata event has been
+     * emitted earlier, the list will be empty.
+     *
+     * @return an immutable list of added event types, not {@code null}
+     */
+    public final List<EventType> getRemovedEventTypes() {
+        if (removed == null) {
+            calculateDelta();
+        }
+        return removed;
+    }
+
+    /**
+     * Returns a list of configurations.
+     *
+     * @return an immutable list of configurations, not {@code null}
+     */
+    public List<Configuration> getConfigurations() {
+        return configurations;
+    }
+
+    private void calculateDelta() {
+        List<EventType> added = new ArrayList<>();
+        Map<Long, EventType> previousSet = new HashMap<>(previous.size());
+        for (EventType eventType : previous) {
+            previousSet.put(eventType.getId(), eventType);
+        }
+        for (EventType eventType : current) {
+            EventType t = previousSet.remove(eventType.getId());
+            if (t == null) {
+                added.add(eventType);
+            }
+        }
+        this.removed = Collections.unmodifiableList(new ArrayList<>(previousSet.values()));
+        this.added = Collections.unmodifiableList(added);
+    }
+}

src/jdk.jfr/share/classes/jdk/jfr/consumer/RecordedObject.java

@@ -34,15 +34,19 @@
 import java.util.Comparator;
 import java.util.List;
 import java.util.Objects;
+import java.util.function.Consumer;
 
+import jdk.jfr.Configuration;
+import jdk.jfr.EventType;
 import jdk.jfr.Timespan;
 import jdk.jfr.Timestamp;
 import jdk.jfr.ValueDescriptor;
-import jdk.jfr.internal.consumer.JdkJfrConsumer;
-import jdk.jfr.internal.consumer.ObjectFactory;
 import jdk.jfr.internal.PrivateAccess;
 import jdk.jfr.internal.Type;
+import jdk.jfr.internal.consumer.EventDirectoryStream;
+import jdk.jfr.internal.consumer.JdkJfrConsumer;
 import jdk.jfr.internal.consumer.ObjectContext;
+import jdk.jfr.internal.consumer.ObjectFactory;
 import jdk.jfr.internal.tool.PrettyWriter;
 
 /**
@@ -136,6 +140,12 @@ public void setEndTicks(RecordedEvent event, long endTicks) {
             public Object[] eventValues(RecordedEvent event) {
                 return event.objects;
             }
+
+            @Override
+            public MetadataEvent newMetadataEvent(List<EventType> previous, List<EventType> current,
+                    List<Configuration> configurations) {
+                return new MetadataEvent(previous, current, configurations);
+            }
         };
         JdkJfrConsumer.setAccess(access);
     }

src/jdk.jfr/share/classes/jdk/jfr/consumer/RecordingStream.java

@@ -30,6 +30,8 @@
 import java.security.AccessController;
 import java.time.Duration;
 import java.time.Instant;
+import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import java.util.function.Consumer;
 
@@ -43,6 +45,7 @@
 import jdk.jfr.internal.SecuritySupport;
 import jdk.jfr.internal.Utils;
 import jdk.jfr.internal.consumer.EventDirectoryStream;
+import jdk.jfr.internal.consumer.JdkJfrConsumer;
 
 /**
  * A recording stream produces events from the current JVM (Java Virtual
@@ -85,13 +88,21 @@ public RecordingStream() {
         this.recording = new Recording();
         try {
             PlatformRecording pr = PrivateAccess.getInstance().getPlatformRecording(recording);
-            this.directoryStream = new EventDirectoryStream(acc, null, SecuritySupport.PRIVILEGED, pr);
+            this.directoryStream = new EventDirectoryStream(acc, null, SecuritySupport.PRIVILEGED, pr, configurations());
         } catch (IOException ioe) {
             this.recording.close();
             throw new IllegalStateException(ioe.getMessage());
         }
     }
 
+    private List<Configuration> configurations() {
+        try {
+            return Configuration.getConfigurations();
+        } catch (Exception e) {
+            return Collections.emptyList();
+        }
+    }
+
     /**
      * Creates a recording stream using settings from a configuration.
      * <p>
@@ -361,4 +372,9 @@ public void awaitTermination(Duration timeout) throws InterruptedException {
     public void awaitTermination() throws InterruptedException {
         directoryStream.awaitTermination();
     }
+
+    @Override
+    public void onMetadata(Consumer<MetadataEvent> action) {
+        directoryStream.onMetadata(action);
+    }
 }

src/jdk.jfr/share/classes/jdk/jfr/internal/PlatformRecording.java

@@ -86,6 +86,7 @@ public final class PlatformRecording implements AutoCloseable {
     private boolean shouldWriteActiveRecordingEvent = true;
     private Duration flushInterval = Duration.ofSeconds(1);
     private long finalStartChunkNanos = Long.MIN_VALUE;
+    private long startNanos = -1;
 
     PlatformRecording(PlatformRecorder recorder, long id) {
         // Typically the access control context is taken
@@ -103,7 +104,6 @@ public final class PlatformRecording implements AutoCloseable {
     public long start() {
         RecordingState oldState;
         RecordingState newState;
-        long startNanos = -1;
         synchronized (recorder) {
             oldState = getState();
             if (!Utils.isBefore(state, RecordingState.RUNNING)) {
@@ -826,11 +826,29 @@ public long getStreamIntervalMillis() {
         }
     }
 
+    public long getStartNanos() {
+        return startNanos;
+    }
+
     public long getFinalChunkStartNanos() {
         return finalStartChunkNanos;
     }
 
     public void setFinalStartnanos(long chunkStartNanos) {
        this.finalStartChunkNanos = chunkStartNanos;
     }
+
+    public void removeBefore(Instant timestamp) {
+        synchronized (recorder) {
+            while (!chunks.isEmpty()) {
+                RepositoryChunk oldestChunk = chunks.peek();
+                if (!oldestChunk.getEndTime().isBefore(timestamp)) {
+                    return;
+                }
+                chunks.removeFirst();
+                removed(oldestChunk);
+            }
+        }
+
+    }
 }