[WFCORE-7041] Reporting of virtual thread pinning

.github/workflows/dep-diff-pull_request.yml

@@ -38,7 +38,7 @@ jobs:
         uses: actions/setup-java@v4
         with:
           distribution: 'temurin'
-          java-version: 11
+          java-version: 17
 
       # Run the caching against the base version only
       - name: Cache local Maven repository

README.md

@@ -14,7 +14,7 @@ Building
 
 Prerequisites:
 
-* JDK 11 or newer - check `java -version`
+* JDK 17 or newer - check `java -version`
 * Maven 3.6.0 or newer - check `mvn -v`
 
 To build with your own Maven installation:
@@ -63,7 +63,7 @@ Contributing
 Using Eclipse
 -------------
 1. Install the latest version of Eclipse.
-2. Make sure Xmx in Eclipse.ini is at least 1280M, and it's using java 11
+2. Make sure Xmx in Eclipse.ini is at least 1280M, and it's using java 17
 3. Launch Eclipse and install the m2e plugin, make sure it uses your repo configs
    (get it from: https://www.eclipse.org/m2e/
    or install "Maven Integration for Eclipse" from the Eclipse Marketplace).

core-feature-pack/common/src/main/resources/modules/system/layers/base/org/wildfly/extension/core-management/main/module.xml

@@ -17,6 +17,9 @@
     <dependencies>
         <!-- for java.beans -->
         <module name="java.desktop"/>
+        <!-- Needed by the service=thread-pinning-recorder resource,
+             but if the VM doesn't provide it, the resource can deal with that -->
+        <module name="jdk.jfr" optional="true"/>
         <module name="io.smallrye.jandex"/>
         <module name="org.jboss.as.controller"/>
         <module name="org.jboss.as.server" />

core-management/core-management-subsystem/src/main/java/org/wildfly/extension/core/management/CoreManagementRootResourceDefinition.java

@@ -39,7 +39,8 @@ public Collection<AttributeDefinition> getAttributes() {
     protected List<? extends PersistentResourceDefinition> getChildren() {
         return Arrays.asList(ConfigurationChangeResourceDefinition.INSTANCE,
                 new ProcessStateListenerResourceDefinition(),
-                UnstableApiAnnotationResourceDefinition.INSTANCE
+                UnstableApiAnnotationResourceDefinition.INSTANCE,
+                VirtualThreadPinningResourceDefinition.create()
         );
     }
 }

core-management/core-management-subsystem/src/main/java/org/wildfly/extension/core/management/CoreManagementSubsystemSchema.java

@@ -17,8 +17,8 @@
 import java.util.Map;
 
 /**
- * Parser and Marshaller for core-management subsystem.
- *
+ * Parser and Marshaller for the core-management subsystem.
+ * <p/>
  * <em>All resources and attributes must be listed explicitly and not through any collections.</em>
  * This ensures that if the resource definitions change in later version (e.g. a new attribute is added),
  * this will have no impact on parsing this specific version of the subsystem.
@@ -28,8 +28,9 @@
 public enum CoreManagementSubsystemSchema implements PersistentSubsystemSchema<CoreManagementSubsystemSchema> {
 
     VERSION_1_0(1),
-    VERSION_1_0_PREVIEW(1, Stability.PREVIEW);
-    static final Map<Stability, CoreManagementSubsystemSchema> CURRENT = Feature.map(EnumSet.of(VERSION_1_0, VERSION_1_0_PREVIEW));
+    VERSION_1_0_PREVIEW(1, Stability.PREVIEW),
+    VERSION_2_0_PREVIEW(2, Stability.PREVIEW);
+    static final Map<Stability, CoreManagementSubsystemSchema> CURRENT = Feature.map(EnumSet.of(VERSION_1_0, VERSION_2_0_PREVIEW));
 
     private final VersionedNamespace<IntVersion, CoreManagementSubsystemSchema> namespace;
 
@@ -66,6 +67,13 @@ public PersistentResourceXMLDescription getXMLDescription() {
                         .addAttribute(ProcessStateListenerResourceDefinition.PROPERTIES)
                         .addAttribute(ProcessStateListenerResourceDefinition.TIMEOUT)
                         .build());
+        builder.addChild(
+                factory.builder(VirtualThreadPinningResourceDefinition.RESOURCE_REGISTRATION)
+                        .addAttribute(VirtualThreadPinningResourceDefinition.START_MODE)
+                        .addAttribute(VirtualThreadPinningResourceDefinition.LOG_LEVEL)
+                        .addAttribute(VirtualThreadPinningResourceDefinition.MAX_STACK_DEPTH)
+                        .build()
+        );
         return builder.build();
     }
 }

core-management/core-management-subsystem/src/main/java/org/wildfly/extension/core/management/VirtualThreadPinningRecorderService.java

@@ -0,0 +1,146 @@
+/*
+ * Copyright The WildFly Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.wildfly.extension.core.management;
+
+import java.time.Duration;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.util.stream.Collectors;
+
+import jdk.jfr.consumer.RecordedEvent;
+import jdk.jfr.consumer.RecordedFrame;
+import jdk.jfr.consumer.RecordedStackTrace;
+import jdk.jfr.consumer.RecordedThread;
+import jdk.jfr.consumer.RecordingStream;
+import org.jboss.as.controller.CapabilityServiceBuilder;
+import org.jboss.as.controller.CapabilityServiceTarget;
+import org.jboss.logging.Logger;
+import org.jboss.msc.Service;
+import org.jboss.msc.service.ServiceController;
+import org.jboss.msc.service.StartContext;
+import org.jboss.msc.service.StopContext;
+import org.wildfly.extension.core.management.logging.CoreManagementLogger;
+
+/**
+ * Service that initiates a JFR {@link RecordingStream} to listen for {@code jdk.VirtualThreadPinned}
+ * events. When events are received it logs them and records their duration for reporting via the
+ * subsystem metrics.
+ */
+final class VirtualThreadPinningRecorderService implements Service {
+
+    static void install(CapabilityServiceTarget target,
+                        VirtualThreadPinningResourceDefinition.StartMode mode,
+                        VirtualThreadPinningResourceDefinition.PinningRecorderData pinningRecorderData) {
+        VirtualThreadPinningRecorderService service = new VirtualThreadPinningRecorderService(pinningRecorderData);
+        CapabilityServiceBuilder<?> builder = target.addService()
+                .setInstance(service)
+                .setInitialMode(mode == VirtualThreadPinningResourceDefinition.StartMode.ON_DEMAND
+                        ? ServiceController.Mode.ON_DEMAND : ServiceController.Mode.ACTIVE);
+        builder.provides(VirtualThreadPinningResourceDefinition.RUNTIME_CAPABILITY);
+        builder.install();
+    }
+
+    private final VirtualThreadPinningResourceDefinition.PinningRecorderData pinningRecorderData;
+    private volatile RecordingStream recordingStream;
+
+    VirtualThreadPinningRecorderService(VirtualThreadPinningResourceDefinition.PinningRecorderData pinningRecorderData) {
+        this.pinningRecorderData = pinningRecorderData;
+    }
+
+    @Override
+    public void start(StartContext startContext) {
+
+        // If we are not in an SE runtime that can support jdk.VirtualThreadPinning notifications, log and return.
+        // Note: We could test this in 'install' and not install, but that would break other services that
+        // depend on this one to get it to start if it's configured to start ON_DEMAND.
+        int vmVersion = Runtime.version().feature();
+        if (vmVersion < 21) {
+            // This is at INFO as this service not doing anything is harmless in a VM that doesn't support virtual threads
+            CoreManagementLogger.ROOT_LOGGER.virtualThreadsUnsupported(vmVersion);
+            return;
+        } else {
+            // Not all JREs include jdk.jfr. Don't fail if we're in such an env; just WARN
+            try {
+                VirtualThreadPinningRecorderService.class.getClassLoader().loadClass("jdk.jfr.consumer.RecordingStream");
+            } catch (ClassNotFoundException cnfe) {
+                CoreManagementLogger.ROOT_LOGGER.virtualThreadPinningNotificationUnsupported();
+                return;
+            }
+        }
+
+        RecordingStream rs = new RecordingStream();
+        rs.enable("jdk.VirtualThreadPinned").withStackTrace();
+        rs.onEvent("jdk.VirtualThreadPinned", VirtualThreadPinningRecorderService.this::onEvent);
+        rs.setMaxAge(Duration.ofSeconds(10));
+
+        try {
+            rs.startAsync();
+            recordingStream = rs;
+        } catch (RuntimeException e) {
+            rs.close();
+            throw e;
+        }
+    }
+
+    @Override
+    public void stop(StopContext stopContext) {
+        if (recordingStream != null) {
+            recordingStream.close();
+        }
+    }
+
+    private void onEvent(RecordedEvent event) {
+        pinningRecorderData.metrics.recordPinning(event.getDuration());
+
+        Logger.Level level = pinningRecorderData.level.get();
+        if (CoreManagementLogger.VIRTUAL_THREAD_LOGGER.isEnabled(level)) {
+            CoreManagementLogger.VIRTUAL_THREAD_LOGGER.log(level, getEventLogMessage(event, pinningRecorderData.stackDepth.get()));
+        }
+
+    }
+
+    private static String getEventLogMessage(RecordedEvent event, int maxStackDepth) {
+        return CoreManagementLogger.VIRTUAL_THREAD_LOGGER.threadPinningDetected(
+                formatThreadName(event),
+                event.getDuration().toMillis(),
+                LocalDateTime.ofInstant(event.getStartTime(), ZoneId.systemDefault()),
+                getStackTrace(event, maxStackDepth)
+        );
+    }
+
+    private static String formatThreadName(RecordedEvent event) {
+        RecordedThread recordedThread = event.getThread();
+        if (recordedThread == null) {
+            return "<unknown>";
+        }
+        String javaName = recordedThread.getJavaName();
+        javaName = javaName == null || javaName.isEmpty() ? "<unnamed>" : javaName;
+        return javaName + " (javaThreadId = " + recordedThread.getJavaThreadId() + ")";
+    }
+
+    private static String getStackTrace(RecordedEvent event, int maxStackDepth) {
+        int depth = maxStackDepth < 0 ? Integer.MAX_VALUE : maxStackDepth;
+        RecordedStackTrace trace = event.getStackTrace();
+        String formatted;
+        if (trace != null && depth > 0) {
+            formatted = "\n\t" + trace.getFrames().stream()
+                    .limit(depth)
+                    .map(VirtualThreadPinningRecorderService::formatStackTraceFrame)
+                    .collect(Collectors.joining("\n\t\t at "));
+            if (depth < trace.getFrames().size()) {
+                formatted += "\n\t\t(...)";
+            }
+        } else {
+            formatted = "<unavailable>";
+        }
+        return formatted;
+    }
+
+    private static String formatStackTraceFrame(RecordedFrame frame) {
+        return frame.getMethod().getType().getName() + "#" + frame.getMethod().getName() + ": " + frame.getLineNumber();
+    }
+
+}