Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Demonstrates that element hangs in time buffered WorkQueueProcessor if it is submitted at the same time when other elements are consumed. Completion with exception means problem reproduction. Java 8 and Reactor 2.5.0.M4 are used. Async Log4j2 with SLF4J is used for logging. Logging can be comment out to simplify setup.
package org.ivan.experiments.reactor;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.WorkQueueProcessor;
import reactor.core.scheduler.Schedulers;
import reactor.core.scheduler.TimedScheduler;
import reactor.core.subscriber.SubmissionEmitter;
import java.io.File;
import java.time.Duration;
import java.util.List;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.LockSupport;
import static reactor.core.util.WaitStrategy.liteBlocking;
public class HangingElement {
static {
System.setProperty("log4j.configurationFile", new File("log4j.xml").getAbsolutePath());
System.setProperty("Log4jContextSelector", "org.apache.logging.log4j.core.async.AsyncLoggerContextSelector");
new File("debug.log").delete();
}
public static final Logger logger = LoggerFactory.getLogger(HangingElement.class);
public static final String e = "Element";
public static final String s = "Synchronizer";
public static void main(String[] args) throws Exception {
highRate();
}
public static void highRate() throws Exception {
WorkQueueProcessor<String> queueProcessor = WorkQueueProcessor.share("Processor", 256, liteBlocking());
TimedScheduler timer = Schedulers.newTimer("Timer", 1);
queueProcessor
.buffer(32, Duration.ofMillis(2), timer)
.subscribe(new Subscriber<List<String>>() {
int counter;
@Override
public void onSubscribe(Subscription s) {
s.request(Long.MAX_VALUE);
}
@Override
public void onNext(List<String> strings) {
int size = strings.size();
logger.debug("Consumed: " + size);
counter += size;
if (strings.contains(s)) {
synchronized (s) {
logger.debug("Synchronizer!");
s.notifyAll();
}
}
}
@Override
public void onError(Throwable t) {
}
@Override
public void onComplete() {
System.out.println("Consumed in total: " + counter);
}
});
SubmissionEmitter<String> emitter = queueProcessor.connectEmitter();
try {
submitInCurrentThread(emitter);
} finally {
emitter.finish();
timer.shutdown();
}
TimeUnit.SECONDS.sleep(1);
}
public static void submitInCurrentThread(SubmissionEmitter<String> emitter) {
Random rand = new Random();
for (int i = 0; i < 100_000; i++) {
long re = emitter.submit(e);
logger.debug("Submit element result " + re);
LockSupport.parkNanos(2_000_000 + rand.nextInt(200_000) - 100_000);
synchronized (s) {
long rd = emitter.submit(s);
logger.debug("Submit drain result " + rd);
timeoutWait(s);
}
}
}
private static void timeoutWait(Object o) {
long t0 = System.currentTimeMillis();
try {
o.wait(5_000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
if (System.currentTimeMillis() - t0 > 4_000) {
throw new RuntimeException("Timeout!");
}
}
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.