From 80d7c3d3325cb8936ee3749b385ac337a4e8bdb7 Mon Sep 17 00:00:00 2001 From: Mateusz Nalepa Date: Sat, 29 Aug 2026 22:00:43 +0200 Subject: [PATCH] add reproducible example --- docker-compose.yml | 8 + .../TwoConcurrentRequestsSpecification.groovy | 290 ++++++++++++++++++ gradle.properties | 1 + 3 files changed, 299 insertions(+) create mode 100644 docker-compose.yml create mode 100644 driver-reactive-streams/src/test/functional/com/mongodb/reactivestreams/client/TwoConcurrentRequestsSpecification.groovy diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000000..0f05020040 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,8 @@ +version: "3.9" + +services: + mongo: + image: mongo:latest + ports: + - "27017:27017" + command: ["--setParameter", "enableTestCommands=1"] diff --git a/driver-reactive-streams/src/test/functional/com/mongodb/reactivestreams/client/TwoConcurrentRequestsSpecification.groovy b/driver-reactive-streams/src/test/functional/com/mongodb/reactivestreams/client/TwoConcurrentRequestsSpecification.groovy new file mode 100644 index 0000000000..b6a2d4185a --- /dev/null +++ b/driver-reactive-streams/src/test/functional/com/mongodb/reactivestreams/client/TwoConcurrentRequestsSpecification.groovy @@ -0,0 +1,290 @@ +package com.mongodb.reactivestreams.client + + +import com.mongodb.MongoClientSettings +import com.mongodb.connection.TransportSettings +import com.mongodb.observability.micrometer.MicrometerObservabilitySettings +import io.micrometer.observation.Observation +import io.micrometer.observation.ObservationHandler +import io.micrometer.observation.ObservationRegistry +import org.bson.Document +import reactor.core.publisher.Flux +import spock.lang.IgnoreIf + +import java.time.Duration +import java.time.LocalTime +import java.util.concurrent.Callable +import java.util.concurrent.ExecutionException +import java.util.concurrent.ExecutorService +import java.util.concurrent.Executors +import java.util.concurrent.Future +import java.util.concurrent.ThreadFactory +import java.util.concurrent.TimeUnit +import java.util.concurrent.TimeoutException +import java.util.concurrent.atomic.AtomicInteger +import java.util.stream.Collectors + +import static com.mongodb.ClusterFixture.getServerApi + +@IgnoreIf({ getServerApi() != null }) +class TwoConcurrentRequestsSpecification extends FunctionalSpecification { + MongoClient client + MongoDatabase database + MongoCollection collection + String someCollectionName = 'testCollection' + + def setup() { + + def observationRegistry = ObservationRegistry.create() + + observationRegistry.observationConfig().observationHandler(new ObservationHandler() { + + @Override + void onStart(Observation.Context context) { + context.put("startNanos", System.nanoTime()) + logMessage("started: ${context.name}") + } + + @Override + void onStop(Observation.Context context) { + def startNanos = context.get("startNanos") as Long + def elapsedMillis = startNanos == null ? -1 : (System.nanoTime() - startNanos) / 1_000_000 + logMessage("stopped: ${context.name} in ${elapsedMillis} ms") + } + + @Override + boolean supportsContext(Observation.Context context) { + return true + } + }) + def micrometerSettings = MicrometerObservabilitySettings.builder() + .observationRegistry(observationRegistry) + .build() + + def threadCounter = new AtomicInteger(0) + def threadFactory = { Runnable r -> + def number = threadCounter.incrementAndGet() + new Thread(r, "mongo-thread-" + number) + } as ThreadFactory + + +// def transportSettings = +// TransportSettings +// .nettyBuilder() +// .eventLoopGroup(new MultiThreadIoEventLoopGroup(1, threadFactory, NioIoHandler.newFactory())) +// .build() + + def transportSettings = + TransportSettings + .asyncBuilder() + .executorService( + new MonitoredExecutorService( + Executors.newFixedThreadPool(1, threadFactory)) + ) + .build() + + def settings = + MongoClientSettings + .builder() + .transportSettings(transportSettings) + .observabilitySettings(micrometerSettings) + .build() + + client = MongoClients.create(settings) + database = client.getDatabase(getDatabaseName()) + + Flux.from(database.createCollection(someCollectionName)).blockLast() + collection = database.getCollection(someCollectionName) + + def document = new Document('name', 'test').append('value', 42) + Flux.from(collection.insertOne(document)).blockLast() + println "#### setup done" + println "" + + } + + def cleanup() { + println "" + println "#### cleanup started" + client?.close() + } + + def 'should execute two commands'() { + given: + def filter = Document.parse('{ $where: "sleep(5000) || true" }') + + def runnable = { + Flux + .from(collection.find(filter)) + .collectList() + .map { docs -> + simulateCpuCodeForOperationIndex(1) + docs + } + .block() + } + + when: + def threadUser1 = new Thread(runnable, "user-thread-1") + threadUser1.start() + + def threadUser2 = new Thread(runnable, "user-thread-2") + threadUser2.start() + + then: + threadUser1.join() + threadUser2.join() + true + } + + static void logMessage(String msg) { + def now = LocalTime.now() + def thread = Thread.currentThread().name + println "[${thread}] | ${now.hour}:${now.minute}:${now.second} | ${msg}" + } + + static long simulateCpuCodeForOperationIndex(int index) { + long startTime = System.nanoTime() + long iteration = 0L + + while (Duration.ofNanos(System.nanoTime() - startTime).getSeconds() < 5) { + iteration++; + Math.sqrt((double) iteration); // busy cpu wait + } + + return iteration + } + + class MonitoredExecutorService implements ExecutorService { + + private final ExecutorService delegate; + + MonitoredExecutorService(ExecutorService delegate) { + this.delegate = delegate; + } + + @Override + void shutdown() { + delegate.shutdown(); + } + + @Override + List shutdownNow() { + return delegate.shutdownNow(); + } + + @Override + boolean isShutdown() { + return delegate.isShutdown(); + } + + @Override + boolean isTerminated() { + return delegate.isTerminated(); + } + + @Override + boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException { + return delegate.awaitTermination(timeout, unit); + } + + @Override + Future submit(Callable task) { + return delegate.submit(new MonitoredCallable<>(task)); + } + + @Override + Future submit(Runnable task, T result) { + return delegate.submit(new MonitoredRunnable(task), result); + } + + @Override + Future submit(Runnable task) { + return delegate.submit(new MonitoredRunnable(task)); + } + + @Override + List> invokeAll(Collection> tasks) throws InterruptedException { + return delegate.invokeAll(tasks.stream() + .map(MonitoredCallable::new) + .collect(Collectors.toList())); + } + + @Override + List> invokeAll(Collection> tasks, long timeout, TimeUnit unit) throws InterruptedException { + return delegate.invokeAll(tasks.stream() + .map(MonitoredCallable::new) + .collect(Collectors.toList()), timeout, unit); + } + + @Override + T invokeAny(Collection> tasks) throws InterruptedException, ExecutionException { + return delegate.invokeAny(tasks.stream() + .map(MonitoredCallable::new) + .collect(Collectors.toList())); + } + + @Override + T invokeAny(Collection> tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { + return delegate.invokeAny(tasks.stream() + .map(MonitoredCallable::new) + .collect(Collectors.toList()), timeout, unit); + } + + @Override + void execute(Runnable command) { + delegate.execute(new MonitoredRunnable(command)); + } + } + + static class MonitoredRunnable implements Runnable { + + private final Runnable delegate; + + private final long createdAtNanos = System.nanoTime(); + + MonitoredRunnable(Runnable delegate) { + this.delegate = delegate; + } + + @Override + public void run() { + Duration waitTime = Duration.ofNanos(System.nanoTime() - createdAtNanos); + logMessage("Runnable waited: " + waitTime.toMillis() + " ms"); + + long startExecutionNanos = System.nanoTime(); + try { + delegate.run(); + } finally { + Duration executionTime = Duration.ofNanos(System.nanoTime() - startExecutionNanos); + logMessage("Runnable executed in: " + executionTime.toMillis() + " ms"); + } + } + + } + + class MonitoredCallable implements Callable { + + private final Callable delegate; + + private final long createdAtNanos = System.nanoTime(); + + MonitoredCallable(Callable delegate) { + this.delegate = delegate; + } + + @Override + public V call() throws Exception { + Duration waitTime = Duration.ofNanos(System.nanoTime() - createdAtNanos); + logMessage("Callable waited: " + waitTime.toMillis() + " ms"); + + long startExecutionNanos = System.nanoTime(); + try { + return delegate.call(); + } finally { + Duration executionTime = Duration.ofNanos(System.nanoTime() - startExecutionNanos); + logMessage("Callable executed in: " + executionTime.toMillis() + " ms"); + } + } + } +} diff --git a/gradle.properties b/gradle.properties index 0c02147d7f..eab2cb54e7 100644 --- a/gradle.properties +++ b/gradle.properties @@ -24,3 +24,4 @@ defaultScalaVersion=2.13 runOnceTasks=clean,release org.gradle.java.installations.auto-download=false org.gradle.java.installations.fromEnv=JDK8,JDK11,JDK17,JDK21,JDK21_GRAALVM +skipCryptVerify=true