BoundedElasticScheduler does not invoke any tasks after interruption.

[jacekmg]

I've stumbled upon a problem with BoundedElasticScheduler when there's high traffic creating many queued task on it. Invoking dispose() on scheduled tasks seems to break scheduler, to the point where it is not invoking any new tasks.

Expected Behavior

Scheduler should correctly invoke tasks, after disposing all of its other tasks.

Actual Behavior

Scheduler doesn't do anything after disposing all of its tasks, and scheduling a new one.

Steps to Reproduce

In the following example, behavior is as follows:

• ~520 tasks are scheduled
• ~300 scheduled tasks are processed correctly
• When dispose() is called all of scheduled tasks are cancelled
• On most scheduled tasks InterruptedException occurs, except ~15 last ones (505-520)
• Scheduled task after dispose() call is never invoked.

Logger logger = LoggerFactory.getLogger("Test");

    @Test
    public void test() throws InterruptedException {
        Scheduler scheduler =
                Schedulers.newBoundedElastic(30, Schedulers.DEFAULT_BOUNDED_ELASTIC_QUEUESIZE, "test-scheduler");
        Disposable subscription = Flux.range(0, 100000000)
                .flatMap(i -> {
                    logger.info("Scheduling {}", i);
                    return Mono.fromSupplier(() -> invokeLongOperation(i))
                            .subscribeOn(scheduler)
                            .publishOn(Schedulers.parallel())
                            .doOnError(throwable -> logger.error("Error occurred in thread {} ", i, throwable))
                            .doOnCancel(() -> logger.info("Thread {} cancelled", i));
                })
                .subscribe(i -> logger.info("Thread {} finished", i),
                           throwable -> logger.error("Error ", throwable));
        Thread.sleep(10000);
        logger.info("Invoking dispose");
        subscription.dispose();
        Thread.sleep(1000);
        Mono.fromSupplier(() -> invokeLongOperation(999999999))
                .subscribeOn(scheduler)
                .publishOn(Schedulers.parallel())
                .subscribe((i) -> logger.info("Consumed value {} ", i),
                           throwable -> logger.error("Exception"));
        logger.info("Sleeping");
        Thread.sleep(100000);
    }

    private int invokeLongOperation(Integer i) {
        logger.info("Invoking long operation {}", i);
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            logger.error("Thread {} interrupted", i);
        }
        return i;
    }

##Environment

• Reactor version used: 3.3.0
• JVM version: 1.8.0

[simonbasle]

Hi @jacekmg and thanks for the report. I was unable to reproduce the issue, that last task with i=999999999 seems to always correctly end up executing on my machine.

Note that since the disposal of tasks inside the flatMap is sequential, and that each disposal of an active Worker in the BoundedElasticScheduler leads to it attempting to reuse the underlying thread for the next queued-due-to-thread-cap task, a number of tasks that have been scheduled but were enqueued by the time subscription.dispose() is invoked will still briefly start running then be cancelled (ie. interrupted).

(As a side note, the unit test was not entirely useful because it relies on logging rather than capturing state (eg. in AtomicInteger or the like) and asserting said state at the end. The test as written is always green, which didn't help me check whether or not things were behaving as expected)

[LSwiatek]

I've experienced the same issue

on first glance it looks like race condition when disposing ActiveWroker/DefferedWorker, which results in worker threads not beeing released to idle pool when cancel is called on scheduled tasks

my test case :

import org.junit.Assert;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.Disposable;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoSink;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

import java.time.Duration;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Elastic {

    private static final Logger LOG = LoggerFactory.getLogger(Elastic.class);
    private Scheduler scheduler;
    private LinkedList<MonoSink<String>> listeners;
    private ExecutorService producer;

    @BeforeEach
    void setUp() {
        scheduler = Schedulers.newBoundedElastic(3, 100000, "subscriberElastic", 600, true);
        listeners = new LinkedList<>();
        startProducer();
    }

    @AfterEach
    void tearDown() {
        producer.shutdown();
    }

    @Test
    void stuck() throws InterruptedException {
        List<Disposable> scheduled = new LinkedList<>();

        for (int i = 0; i < 50; i++) {
            scheduled.add(
                    Mono.create(this::addListener)
                            .subscribeOn(scheduler)
                            .subscribe(LOG::info)
            );
        }

        Thread.sleep(1000);
        scheduled.forEach(Disposable::dispose);
        Thread.sleep(1000);

        String result = Mono.create(this::addListener)
                .subscribeOn(scheduler)
                .map(res -> res + " the end")
                .block(Duration.ofSeconds(5));
        LOG.info(result);
        Assert.assertNotNull(result);

    }

    private void startProducer() {
        producer = Executors.newSingleThreadExecutor();
        producer.submit(() -> {
            int i = 0;
            while (true) {
                MonoSink<String> sink = listeners.poll();
                if (sink != null) {
                    sink.success(Integer.toString(i++));
                }

                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    LOG.info("Producer stopping");
                    return;
                }
            }
        });
    }

    public void addListener(MonoSink<String> sink) {
        try {
            Thread.sleep(200);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        listeners.add(sink);
        sink.onDispose(() -> removeListener(sink));
    }

    public void removeListener(MonoSink<String> sink) {
        listeners.remove(sink);
    }
}

on my machine it always ends with
java.lang.IllegalStateException: Timeout on blocking read for 5000 MILLISECONDS
you may tweek number of threads and tasks if needed

[simonbasle]

also verified to be fixed by the new implementation in #2040