fix: LocalBuffer.count, first, startTime thread-safety issues

bom-testing/build.gradle.kts

@@ -16,6 +16,7 @@ dependencies {
         api("org.jmockit:jmockit-coverage:1.23")
         api("org.jmockit:jmockit:1.50")
         api("org.mockito:mockito-core:5.18.0")
+        api("org.openjdk.jcstress:jcstress-core:0.16")
         api("org.testcontainers:junit-jupiter:1.21.3")
     }
 }

boot/build.gradle.kts

@@ -3,9 +3,16 @@ plugins {
     id("build-logic.test-junit5")
     id("build-logic.test-jmockit")
     id("build-logic.kotlin")
+    kotlin("kapt")
 }
 
 dependencies {
     testImplementation("ch.qos.logback:logback-classic")
     testImplementation("io.mockk:mockk")
+    testImplementation("org.openjdk.jcstress:jcstress-core")
+    testAnnotationProcessor(platform(projects.bomTesting))
+    testAnnotationProcessor("org.openjdk.jcstress:jcstress-core")
+    testRuntimeOnly(projects.jcstressJupiterEngine)
+    kaptTest(platform(projects.bomTesting))
+    kaptTest("org.openjdk.jcstress:jcstress-core")
 }

boot/src/main/java/org/qubership/profiler/agent/LocalBuffer.java

@@ -1,17 +1,22 @@
 package org.qubership.profiler.agent;
 
 import java.util.Arrays;
+import java.util.concurrent.atomic.AtomicLongFieldUpdater;
 
 public class LocalBuffer {
     public final static int SIZE = Integer.getInteger(LocalBuffer.class.getName() + ".SIZE", 4096);
+    private final static AtomicLongFieldUpdater<LocalBuffer> START_TIME_UPDATER = AtomicLongFieldUpdater.newUpdater(LocalBuffer.class, "startTime");
     volatile public LocalState state;
     public LocalBuffer prevBuffer;
 
     public final long[] data = new long[SIZE];
     public final Object[] value = new Object[SIZE];
-    public long startTime;
-    public int count;
-    public int first;
+    // volatile since we want atomic values as it can be updated by both Dumper and mutator threads
+    public volatile long startTime;
+    // volatile as it is updated by mutator (log...) and read by Dumper thread
+    public volatile int count;
+    // volatile as it might be updated by both Dumper (stealData) and mutator (buffer.reset()) threads
+    public volatile int first;
     public boolean corrupted;
     // Contains the total amount of heap consumed by the large events stored in the buffer
     private long largeEventsVolume;
@@ -20,6 +25,10 @@ public LocalBuffer() {
         init(null);
     }
 
+    public void increaseStartTime(long value) {
+        START_TIME_UPDATER.addAndGet(this, value);
+    }
+
     public void init(LocalBuffer prevBuffer) {
         startTime = TimerCache.now;
         count = 0;

boot/src/main/java/org/qubership/profiler/agent/Profiler.java

@@ -178,12 +178,10 @@ public static void exchangeBuffer(LocalBuffer buffer,
         LocalBuffer newBuffer = state.buffer;
         long[] data = newBuffer.data;
 
-        if (newBuffer.count > 0) {
-            System.arraycopy(data, 0, data, 1, newBuffer.count);
-        }
-        data[0] = methodAndTime;
+        int count = newBuffer.count;
+        data[count] = methodAndTime;
 
-        newBuffer.count++;
+        newBuffer.count = count + 1;
     }
 
     public static MetricsConfiguration getMetricConfigByName(String callType) {

boot/src/test/kotlin/org/qubership/profiler/agent/LocalBufferEventStealTest.kt

@@ -0,0 +1,35 @@
+package org.qubership.profiler.agent
+
+import org.junit.platform.commons.annotation.Testable
+import org.openjdk.jcstress.annotations.*
+import org.openjdk.jcstress.infra.results.IIL_Result
+import org.openjdk.jcstress.infra.results.IJL_Result
+import org.openjdk.jcstress.infra.results.IJ_Result
+import org.openjdk.jcstress.infra.results.IL_Result
+
+@JCStressTest
+@Outcome(id = ["0, 0, null"], expect = Expect.ACCEPTABLE, desc = "Count field update was not visible")
+@Outcome(id = ["1, 1, value"], expect = Expect.ACCEPTABLE, desc = "Count field update and event value was visible")
+@Outcome(id = ["1, .*, null"], expect = Expect.FORBIDDEN, desc = "Event value should be visible if count is visible")
+@Outcome(id = ["1, 0, .*"], expect = Expect.FORBIDDEN, desc = "Event tag should be visible if count is visible")
+@State
+@Testable
+open class LocalBufferEventStealTest {
+
+    private val localBuffer = LocalBuffer()
+
+    @Actor
+    fun writer() {
+        localBuffer.event("value", 42)
+    }
+
+    @Actor
+    fun reader(r: IIL_Result) {
+        val count = localBuffer.count
+        r.r1 = count
+        if (count > 0) {
+            r.r2 = if (localBuffer.data[0] == 0L) 0 else 1;
+            r.r3 = localBuffer.value[0]
+        }
+    }
+}

boot/src/test/kotlin/org/qubership/profiler/agent/LocalBufferInitEnterStealTest.kt

@@ -0,0 +1,30 @@
+package org.qubership.profiler.agent
+
+import org.junit.platform.commons.annotation.Testable
+import org.openjdk.jcstress.annotations.*
+import org.openjdk.jcstress.infra.results.IJ_Result
+
+@JCStressTest
+@Outcome(id = ["0, 0"], expect = Expect.ACCEPTABLE, desc = "Count field update was not visible")
+@Outcome(id = ["1, 1311768467463790320"], expect = Expect.ACCEPTABLE, desc = "Count field update and enter event was visible")
+@Outcome(id = ["1, 0"], expect = Expect.FORBIDDEN, desc = "Method enter event should be visible if count is visible")
+@State
+@Testable
+open class LocalBufferInitEnterStealTest {
+
+    private val localBuffer = LocalBuffer()
+
+    @Actor
+    fun writer() {
+        localBuffer.initTimedEnter(0x1234_5678_9abc_def0L)
+    }
+
+    @Actor
+    fun reader(r: IJ_Result) {
+        val count = localBuffer.count
+        r.r1 = count
+        if (count > 0) {
+            r.r2 = localBuffer.data[0]
+        }
+    }
+}

boot/src/test/kotlin/org/qubership/profiler/agent/LocalBufferResetStealTest.kt

@@ -0,0 +1,31 @@
+package org.qubership.profiler.agent
+
+import org.junit.platform.commons.annotation.Testable
+import org.openjdk.jcstress.annotations.*
+import org.openjdk.jcstress.infra.results.IJ_Result
+
+@JCStressTest
+@Outcome(id = ["0, 0"], expect = Expect.ACCEPTABLE, desc = "Count field update was not visible or the buffer was reset")
+@Outcome(id = ["1, 1311768467463790320"], expect = Expect.ACCEPTABLE, desc = "Count field update and enter event was visible")
+@Outcome(id = ["1, 0"], expect = Expect.FORBIDDEN, desc = "Method enter event should be visible if count is visible")
+@State
+@Testable
+open class LocalBufferResetStealTest {
+
+    private val localBuffer = LocalBuffer()
+
+    @Actor
+    fun writer() {
+        localBuffer.initTimedEnter(0x1234_5678_9abc_def0L)
+        localBuffer.reset()
+    }
+
+    @Actor
+    fun reader(r: IJ_Result) {
+        val count = localBuffer.count
+        r.r1 = count
+        if (count > 0) {
+            r.r2 = localBuffer.data[0]
+        }
+    }
+}

build-logic/jvm/build.gradle.kts

@@ -6,6 +6,8 @@ dependencies {
     implementation(project(":basics"))
     implementation(project(":build-parameters"))
     implementation(project(":verification"))
+    api("org.jetbrains.kotlin.jvm:org.jetbrains.kotlin.jvm.gradle.plugin:2.2.0")
+    api("org.jetbrains.kotlin.kapt:org.jetbrains.kotlin.kapt.gradle.plugin:2.2.0")
     implementation("com.github.vlsi.crlf:com.github.vlsi.crlf.gradle.plugin:2.0.0")
     implementation("com.github.vlsi.gradle-extensions:com.github.vlsi.gradle-extensions.gradle.plugin:2.0.0")
     implementation("org.jetbrains.kotlin:kotlin-gradle-plugin")

build.gradle.kts

@@ -6,6 +6,7 @@ plugins {
     id("com.github.vlsi.ide")
     id("com.github.vlsi.gradle-extensions")
     id("jacoco")
+    kotlin("jvm") apply false
 }
 
 ide {

dumper/src/main/java/org/qubership/profiler/Dumper.java

@@ -800,7 +800,7 @@ private int writeBufferToFS(LocalBuffer buffer) throws IOException {
                     long startOffset = data[last - 1] >>> 32;
                     startOffset -= data[buffer.first] >>> 32;
                     buffer.first = i;
-                    buffer.startTime += startOffset;
+                    buffer.increaseStartTime(startOffset);
                     return 0;
                 }
                 // We found callInfo record and start dumping from the next record
@@ -954,7 +954,7 @@ private int writeBufferToFS(LocalBuffer buffer) throws IOException {
         traceOs.write(EVENT_FINISH_RECORD);
         long startOffset = data[last - 1] >>> 32;
         startOffset -= data[buffer.first] >>> 32;
-        buffer.startTime += startOffset;
+        buffer.increaseStartTime(startOffset);
         buffer.first = last;
         return count;
     }

jcstress-jupiter-engine/build.gradle.kts

@@ -0,0 +1,9 @@
+plugins {
+    id("build-logic.java-library")
+}
+
+dependencies {
+    api(platform(projects.bomTesting))
+    api("org.junit.jupiter:junit-jupiter-engine")
+    api("org.openjdk.jcstress:jcstress-core")
+}

jcstress-jupiter-engine/src/main/java/org/qubership/jcstress/JCStressClassDescriptor.java

@@ -0,0 +1,37 @@
+package org.qubership.jcstress;
+
+import org.junit.platform.engine.TestDescriptor;
+import org.junit.platform.engine.UniqueId;
+import org.junit.platform.engine.support.descriptor.AbstractTestDescriptor;
+import org.junit.platform.engine.support.descriptor.ClassSource;
+
+class JCStressClassDescriptor extends AbstractTestDescriptor {
+    static final String SEGMENT_TYPE = "class";
+
+    private JCStressClassDescriptor(UniqueId uniqueId, Class<?> testClass) {
+        super(uniqueId, determineDisplayName(testClass), ClassSource.from(testClass));
+        // Gradle expects Engine -> Container -> Test hierarchy, so we add a dummy "run" descriptor here.
+        // See https://github.com/junit-team/junit-framework/discussions/4825
+        // It would be great to get "run" descriptors from JCStress itself, but it does not support custom listeners yet
+        addChild(JCStressRunDescriptor.of(uniqueId));
+    }
+
+    static JCStressClassDescriptor of(TestDescriptor parent, Class<?> testClass) {
+        UniqueId uniqueId = parent.getUniqueId().append(JCStressClassDescriptor.SEGMENT_TYPE, testClass.getName());
+        return new JCStressClassDescriptor(uniqueId, testClass);
+    }
+
+    private static String determineDisplayName(Class<?> testClass) {
+        String simpleName = testClass.getSimpleName();
+        return simpleName.isEmpty() ? testClass.getName() : simpleName;
+    }
+
+    Class<?> getTestClass() {
+        return ((ClassSource) getSource().get()).getJavaClass();
+    }
+
+    @Override
+    public Type getType() {
+        return Type.CONTAINER;
+    }
+}

jcstress-jupiter-engine/src/main/java/org/qubership/jcstress/JCStressClassFilter.java

@@ -0,0 +1,15 @@
+package org.qubership.jcstress;
+
+import org.junit.platform.commons.support.AnnotationSupport;
+import org.openjdk.jcstress.annotations.JCStressTest;
+
+import java.util.function.Predicate;
+
+class JCStressClassFilter implements Predicate<Class<?>> {
+    public static final Predicate<Class<?>> INSTANCE = new JCStressClassFilter();
+
+    @Override
+    public boolean test(Class<?> candidateClass) {
+        return AnnotationSupport.findAnnotation(candidateClass, JCStressTest.class).isPresent();
+    }
+}

jcstress-jupiter-engine/src/main/java/org/qubership/jcstress/JCStressEngineDescriptor.java

@@ -0,0 +1,10 @@
+package org.qubership.jcstress;
+
+import org.junit.platform.engine.UniqueId;
+import org.junit.platform.engine.support.descriptor.EngineDescriptor;
+
+public class JCStressEngineDescriptor extends EngineDescriptor {
+    public JCStressEngineDescriptor(UniqueId uniqueId) {
+        super(uniqueId, "jcstress");
+    }
+}

jcstress-jupiter-engine/src/main/java/org/qubership/jcstress/JCStressRunDescriptor.java

@@ -0,0 +1,22 @@
+package org.qubership.jcstress;
+
+import org.junit.platform.engine.TestDescriptor;
+import org.junit.platform.engine.UniqueId;
+import org.junit.platform.engine.support.descriptor.AbstractTestDescriptor;
+
+class JCStressRunDescriptor extends AbstractTestDescriptor {
+    static final String SEGMENT_TYPE = "run";
+
+    JCStressRunDescriptor(UniqueId uniqueId) {
+        super(uniqueId, "run");
+    }
+
+    static TestDescriptor of(UniqueId uniqueId) {
+        return new JCStressRunDescriptor(uniqueId.append(SEGMENT_TYPE, "run"));
+    }
+
+    @Override
+    public Type getType() {
+        return Type.TEST;
+    }
+}

jcstress-jupiter-engine/src/main/java/org/qubership/jcstress/JCStressSelectorResolver.java

@@ -0,0 +1,51 @@
+package org.qubership.jcstress;
+
+import static java.util.Collections.singleton;
+import static org.junit.platform.engine.discovery.DiscoverySelectors.selectClass;
+
+import org.junit.platform.engine.UniqueId;
+import org.junit.platform.engine.UniqueId.Segment;
+import org.junit.platform.engine.discovery.ClassSelector;
+import org.junit.platform.engine.discovery.UniqueIdSelector;
+import org.junit.platform.engine.support.discovery.SelectorResolver;
+
+import java.util.List;
+import java.util.Optional;
+import java.util.function.Predicate;
+
+class JCStressSelectorResolver implements SelectorResolver {
+    private final Predicate<String> classNameFilter;
+
+    JCStressSelectorResolver(Predicate<String> classNameFilter) {
+        this.classNameFilter = classNameFilter;
+    }
+
+    @Override
+    public Resolution resolve(ClassSelector selector, Context context) {
+        if (!classNameFilter.test(selector.getClassName())) {
+            return Resolution.unresolved();
+        }
+        if (!JCStressClassFilter.INSTANCE.test(selector.getJavaClass())) {
+            // Gradle might supply extra classes, so we need to keep only jcstress-compatible classes here
+            return Resolution.unresolved();
+        }
+        JCStressClassDescriptor classDescriptor =
+                context.addToParent(
+                                parent -> Optional.of(JCStressClassDescriptor.of(parent, selector.getJavaClass())))
+                        .orElseThrow(IllegalStateException::new);
+        return Resolution.match(Match.exact(classDescriptor));
+    }
+
+    @Override
+    public Resolution resolve(UniqueIdSelector selector, Context context) {
+        UniqueId uniqueId = selector.getUniqueId();
+        List<Segment> segments = uniqueId.getSegments();
+        for (int i = segments.size() - 1; i >= 0; i--) {
+            Segment segment = segments.get(i);
+            if (JCStressClassDescriptor.SEGMENT_TYPE.equals(segment.getType())) {
+                return Resolution.selectors(singleton(selectClass(segment.getValue())));
+            }
+        }
+        return Resolution.unresolved();
+    }
+}