1+ /*
2+ * Copyright 2025, Datadog, Inc.
3+ * SPDX-License-Identifier: Apache-2.0
4+ */
5+
6+ package com .datadoghq .profiler ;
7+
8+ import org .junit .jupiter .api .Test ;
9+ import org .openjdk .jmc .common .IMCStackTrace ;
10+ import org .openjdk .jmc .common .item .IItem ;
11+ import org .openjdk .jmc .common .item .IItemCollection ;
12+ import org .openjdk .jmc .common .item .IItemIterable ;
13+ import org .openjdk .jmc .common .item .ItemFilters ;
14+ import org .openjdk .jmc .flightrecorder .JfrLoaderToolkit ;
15+ import org .openjdk .jmc .flightrecorder .CouldNotLoadRecordingException ;
16+
17+ import java .io .IOException ;
18+ import java .nio .file .Files ;
19+ import java .nio .file .Path ;
20+ import java .nio .file .Paths ;
21+ import java .util .ArrayList ;
22+ import java .util .List ;
23+ import java .util .concurrent .CountDownLatch ;
24+ import java .util .concurrent .CyclicBarrier ;
25+ import java .util .concurrent .atomic .AtomicLong ;
26+
27+ import static org .junit .jupiter .api .Assertions .*;
28+
29+ /**
30+ * Test to validate that CallTraceStorage::put() contention is low
31+ * when exclusive operations (processTraces) are running concurrently.
32+ *
33+ * This test exercises contention between:
34+ * - Multiple threads calling put() operations (shared lock)
35+ * - JFR dump operations calling processTraces() (exclusive lock)
36+ */
37+ public class ContendedStorageTest extends AbstractProfilerTest {
38+
39+ @ Override
40+ protected String getProfilerCommand () {
41+ // Generate a lot of CPU samples
42+ return "cpu=1ms" ;
43+ }
44+
45+ @ Override
46+ protected boolean isPlatformSupported () {
47+ return !Platform .isJ9 (); // Avoid J9-specific issues
48+ }
49+
50+ @ Test
51+ public void shouldShowImprovedContentionWithRetries () throws Exception {
52+ List <ContentionResult > currentResults = measureContention ();
53+
54+ // The test validates that the measurement infrastructure works
55+ // In practice, you would modify CallTraceStorage::put to accept retry count
56+ // and test with higher values like tryLockShared(100)
57+
58+ for (ContentionResult currentResult : currentResults ) {
59+ // For this test, we verify that contention measurement works
60+ assertTrue (currentResult .droppedSamples == 0 , "Should measure dropped samples" );
61+ assertTrue (currentResult .totalAttempts > 0 , "Should measure total attempts" );
62+
63+ System .out .printf ("Contention measurement successful: %d/%d samples dropped (%.2f%%)%n" ,
64+ currentResult .droppedSamples , currentResult .totalAttempts ,
65+ (double ) currentResult .droppedSamples / currentResult .totalAttempts * 100 );
66+ }
67+
68+ // The key insight: this test framework can be used to validate
69+ // that increasing retry counts reduces dropped samples
70+ }
71+
72+ private List <ContentionResult > measureContention () throws Exception {
73+ Path jfrFile = Paths .get ("contention-test.jfr" );
74+ List <Path > recordings = new ArrayList <>();
75+ recordings .add (jfrFile );
76+
77+ try {
78+ // Create high contention scenario
79+ int numThreads = Runtime .getRuntime ().availableProcessors () * 2 ;
80+ CyclicBarrier startBarrier = new CyclicBarrier (numThreads + 1 );
81+ CountDownLatch finishLatch = new CountDownLatch (numThreads );
82+
83+ // Start concurrent allocation threads
84+ for (int i = 0 ; i < numThreads ; i ++) {
85+ final int threadId = i ;
86+ Thread worker = new Thread (() -> {
87+ try {
88+ startBarrier .await (); // Synchronize start
89+
90+ // Generate CPU load for 5 seconds to ensure samples
91+ long endTime = System .currentTimeMillis () + 5000 ;
92+ while (System .currentTimeMillis () < endTime ) {
93+ performCpuIntensiveWork (threadId );
94+ }
95+ } catch (Exception e ) {
96+ throw new RuntimeException (e );
97+ } finally {
98+ finishLatch .countDown ();
99+ }
100+ });
101+ worker .start ();
102+ }
103+
104+ // Wait for all threads to be ready
105+ startBarrier .await ();
106+
107+ // Let allocation threads run for a bit, then trigger contention with dumps
108+ Thread .sleep (500 );
109+
110+ // Trigger contention by calling dump during heavy allocation
111+ // This forces processTraces() to acquire exclusive lock while put() operations are active
112+ for (int i = 0 ; i < 3 ; i ++) {
113+ Path tempDump = Paths .get ("temp-contention-" + i + ".jfr" );
114+ dump (tempDump ); // This will cause contention in CallTraceStorage
115+ recordings .add (tempDump );
116+ Thread .sleep (500 );
117+ }
118+
119+ // Wait for all allocation threads to finish
120+ finishLatch .await ();
121+
122+ // Final dump to get all data
123+ dump (jfrFile );
124+
125+ // Analyze contention from JFR data
126+ return analyzeContentionFromJFR (recordings );
127+
128+ } finally {
129+ recordings .forEach (f -> {
130+ try {
131+ Files .deleteIfExists (f );
132+ } catch (IOException e ) {
133+ // ignore
134+ }
135+ });
136+ }
137+ }
138+
139+ private List <ContentionResult > analyzeContentionFromJFR (List <Path > recordings ) throws IOException , CouldNotLoadRecordingException {
140+ List <ContentionResult > results = new ArrayList <>();
141+ for (Path jfrFile : recordings ) {
142+ IItemCollection events = JfrLoaderToolkit .loadEvents (Files .newInputStream (jfrFile ));
143+
144+ // Count profiling events - represents successful put() operations
145+ IItemCollection cpuEvents = events .apply (ItemFilters .type ("datadog.ExecutionSample" ));
146+ IItemCollection allocationEvents = events .apply (ItemFilters .type ("jdk.ObjectAllocationInNewTLAB" ));
147+
148+ // Count events with and without stack traces
149+ long cpuWithStack = countEventsWithStackTrace (cpuEvents );
150+ long cpuWithoutStack = countEventsWithoutStackTrace (cpuEvents );
151+ long allocWithStack = countEventsWithStackTrace (allocationEvents );
152+ long allocWithoutStack = countEventsWithoutStackTrace (allocationEvents );
153+
154+ // Events without stack traces indicate contention - CallTraceStorage::put() returned 0
155+ long contentionDrops = cpuWithoutStack + allocWithoutStack ;
156+ long totalEvents = cpuWithStack + cpuWithoutStack + allocWithStack + allocWithoutStack ;
157+
158+ System .out .printf ("JFR Contention Analysis:%n" );
159+ System .out .printf (" CPU: %d with stack, %d without stack%n" , cpuWithStack , cpuWithoutStack );
160+ System .out .printf (" Alloc: %d with stack, %d without stack%n" , allocWithStack , allocWithoutStack );
161+ System .out .printf (" Contention drops: %d/%d (%.2f%%)%n" ,
162+ contentionDrops , totalEvents ,
163+ totalEvents > 0 ? (double ) contentionDrops / totalEvents * 100 : 0 );
164+ results .add (new ContentionResult (contentionDrops , totalEvents ));
165+ }
166+
167+ return results ;
168+ }
169+
170+ private long countEventsWithStackTrace (IItemCollection events ) {
171+ if (!events .hasItems ()) return 0 ;
172+
173+ long count = 0 ;
174+ for (IItemIterable iterable : events ) {
175+ for (IItem item : iterable ) {
176+ IMCStackTrace stackTrace = STACK_TRACE .getAccessor (iterable .getType ()).getMember (item );
177+ if (stackTrace != null && !stackTrace .getFrames ().isEmpty ()) {
178+ count ++;
179+ }
180+ }
181+ }
182+ return count ;
183+ }
184+
185+ private long countEventsWithoutStackTrace (IItemCollection events ) {
186+ if (!events .hasItems ()) return 0 ;
187+
188+ long count = 0 ;
189+ for (IItemIterable iterable : events ) {
190+ for (IItem item : iterable ) {
191+ IMCStackTrace stackTrace = STACK_TRACE .getAccessor (iterable .getType ()).getMember (item );
192+ if (stackTrace == null || stackTrace .getFrames ().isEmpty ()) {
193+ count ++;
194+ }
195+ }
196+ }
197+ return count ;
198+ }
199+
200+ private void performCpuIntensiveWork (int threadId ) {
201+ // Simple CPU-intensive loop similar to ProfiledCode.burnCycles()
202+ burnCycles (threadId );
203+ }
204+
205+ private void burnCycles (int threadId ) {
206+ // CPU burning pattern that ensures we get profiling samples
207+ long sink = 0 ;
208+ for (int i = 0 ; i < 100000 ; i ++) {
209+ sink += i * threadId ;
210+ sink ^= threadId ;
211+ if (i % 1000 == 0 ) {
212+ // Add some method calls to create interesting stack traces
213+ sink += computeHash (sink , threadId );
214+ }
215+ }
216+ // Store in volatile to prevent optimization
217+ volatileResult = sink ;
218+ }
219+
220+ private long computeHash (long value , int threadId ) {
221+ // Another method in the stack trace
222+ long result = value ;
223+ for (int i = 0 ; i < 100 ; i ++) {
224+ result = Long .rotateLeft (result , 1 );
225+ result ^= (threadId + i );
226+ }
227+ return result ;
228+ }
229+
230+ private volatile long volatileResult ; // Prevent optimization
231+
232+ private static class ContentionResult {
233+ final long droppedSamples ;
234+ final long totalAttempts ;
235+
236+ ContentionResult (long droppedSamples , long totalAttempts ) {
237+ this .droppedSamples = droppedSamples ;
238+ this .totalAttempts = totalAttempts ;
239+ }
240+
241+ double getDropRate () {
242+ return totalAttempts > 0 ? (double ) droppedSamples / totalAttempts : 0.0 ;
243+ }
244+ }
245+ }
0 commit comments