001/*
002 * Copyright © 2025 CUI-OpenSource-Software (info@cuioss.de)
003 *
004 * Licensed under the Apache License, Version 2.0 (the "License");
005 * you may not use this file except in compliance with the License.
006 * You may obtain a copy of the License at
007 *
008 *     http://www.apache.org/licenses/LICENSE-2.0
009 *
010 * Unless required by applicable law or agreed to in writing, software
011 * distributed under the License is distributed on an "AS IS" BASIS,
012 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
013 * See the License for the specific language governing permissions and
014 * limitations under the License.
015 */
016package de.cuioss.benchmarking.common.jfr;
017
018import org.HdrHistogram.Histogram;
019import org.HdrHistogram.Recorder;
020
021import java.util.Map;
022import java.util.concurrent.ConcurrentHashMap;
023import java.util.concurrent.Executors;
024import java.util.concurrent.ScheduledExecutorService;
025import java.util.concurrent.TimeUnit;
026import java.util.concurrent.atomic.AtomicInteger;
027import java.util.concurrent.atomic.AtomicLong;
028
029/**
030 * Central management for JFR instrumentation in benchmarks.
031 * Provides utilities for recording events and computing statistics.
032 */
033public class JfrInstrumentation {
034
035    private static final long HIGHEST_TRACKABLE_VALUE = TimeUnit.SECONDS.toNanos(10);
036    private static final int NUMBER_OF_SIGNIFICANT_VALUE_DIGITS = 3;
037
038    private final Map<String, OperationStats> operationStats = new ConcurrentHashMap<>();
039    private final AtomicInteger concurrentOperations = new AtomicInteger(0);
040    private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(r -> {
041        Thread t = new Thread(r, "JFR-Statistics-Reporter");
042        t.setDaemon(true);
043        return t;
044    });
045
046    public JfrInstrumentation() {
047        // Schedule periodic statistics reporting
048        scheduler.scheduleAtFixedRate(this::reportStatistics, 1, 1, TimeUnit.SECONDS);
049    }
050
051    /**
052     * Records the start of a benchmark operation.
053     * @return an AutoCloseable that should be used in try-with-resources to ensure proper event completion
054     */
055    public OperationRecorder recordOperation(String benchmarkName, String operationType) {
056        return new OperationRecorder(benchmarkName, operationType);
057    }
058
059    /**
060     * Records a benchmark phase transition.
061     */
062    public void recordPhase(String benchmarkName, String phase, int iteration, int totalIterations, int fork, int threadCount) {
063        BenchmarkPhaseEvent event = new BenchmarkPhaseEvent();
064        event.benchmarkName = benchmarkName;
065        event.phase = phase;
066        event.iteration = iteration;
067        event.totalIterations = totalIterations;
068        event.fork = fork;
069        event.threadCount = threadCount;
070        event.commit();
071    }
072
073    /**
074     * Shuts down the instrumentation and releases resources.
075     */
076    public void shutdown() {
077        scheduler.shutdown();
078        try {
079            if (!scheduler.awaitTermination(5, TimeUnit.SECONDS)) {
080                scheduler.shutdownNow();
081            }
082        } catch (InterruptedException e) {
083            scheduler.shutdownNow();
084            Thread.currentThread().interrupt();
085        }
086    }
087
088    private void reportStatistics() {
089        operationStats.forEach((key, stats) -> {
090            Histogram snapshot = stats.recorder.getIntervalHistogram();
091            if (snapshot.getTotalCount() > 0) {
092                OperationStatisticsEvent event = new OperationStatisticsEvent();
093                String[] parts = key.split(":");
094                event.benchmarkName = parts[0];
095                event.operationType = parts[1];
096                event.sampleCount = snapshot.getTotalCount();
097                event.successCount = stats.successCount.getAndSet(0);
098                event.errorCount = stats.errorCount.getAndSet(0);
099                event.meanLatency = (long) snapshot.getMean();
100                event.p50Latency = snapshot.getValueAtPercentile(50.0);
101                event.p95Latency = snapshot.getValueAtPercentile(95.0);
102                event.p99Latency = snapshot.getValueAtPercentile(99.0);
103                event.maxLatency = snapshot.getMaxValue();
104                event.standardDeviation = (long) snapshot.getStdDeviation();
105                event.variance = Math.pow(snapshot.getStdDeviation(), 2);
106                event.coefficientOfVariation = snapshot.getMean() > 0 ?
107                        (snapshot.getStdDeviation() / snapshot.getMean() * 100) : 0;
108                event.concurrentThreads = stats.maxConcurrentThreads.getAndSet(0);
109                event.cacheHitRate = stats.cacheHits.getAndSet(0) * 100.0 / snapshot.getTotalCount();
110                event.commit();
111            }
112        });
113    }
114
115    /**
116     * Helper class for recording individual operations.
117     */
118    public class OperationRecorder implements AutoCloseable {
119        private final OperationEvent event;
120        private final String statsKey;
121        private final long startTime;
122
123        private OperationRecorder(String benchmarkName, String operationType) {
124            this.event = new OperationEvent();
125            this.event.benchmarkName = benchmarkName;
126            this.event.operationType = operationType;
127            this.event.threadName = Thread.currentThread().getName();
128            this.statsKey = benchmarkName + ":" + operationType;
129            this.startTime = System.nanoTime();
130
131            event.begin();
132            int concurrent = concurrentOperations.incrementAndGet();
133            event.concurrentOperations = concurrent;
134
135            // Update max concurrent threads
136            OperationStats stats = operationStats.get(statsKey);
137            if (stats != null) {
138                stats.maxConcurrentThreads.updateAndGet(current -> Math.max(current, concurrent));
139            }
140        }
141
142        public OperationRecorder withPayloadSize(long size) {
143            event.payloadSize = size;
144            return this;
145        }
146
147        public OperationRecorder withMetadata(String key, String value) {
148            event.metadataKey = key;
149            event.metadataValue = value;
150            return this;
151        }
152
153        public OperationRecorder withSuccess(boolean success) {
154            event.success = success;
155            return this;
156        }
157
158        public OperationRecorder withError(String errorType) {
159            event.success = false;
160            event.errorType = errorType;
161            return this;
162        }
163
164        public OperationRecorder withCached(boolean cached) {
165            event.cached = cached;
166            return this;
167        }
168
169        @Override
170        public void close() {
171            concurrentOperations.decrementAndGet();
172            event.end();
173
174            if (event.shouldCommit()) {
175                event.commit();
176            }
177
178            // Record statistics
179            long duration = System.nanoTime() - startTime;
180            OperationStats stats = operationStats.computeIfAbsent(statsKey, k -> new OperationStats());
181            stats.recorder.recordValue(duration);
182
183            if (event.success) {
184                stats.successCount.incrementAndGet();
185            } else {
186                stats.errorCount.incrementAndGet();
187            }
188
189            if (event.cached) {
190                stats.cacheHits.incrementAndGet();
191            }
192        }
193    }
194
195    private static class OperationStats {
196        final Recorder recorder = new Recorder(HIGHEST_TRACKABLE_VALUE, NUMBER_OF_SIGNIFICANT_VALUE_DIGITS);
197        final AtomicLong successCount = new AtomicLong();
198        final AtomicLong errorCount = new AtomicLong();
199        final AtomicLong cacheHits = new AtomicLong();
200        final AtomicInteger maxConcurrentThreads = new AtomicInteger();
201    }
202}