1.x: improve ExecutorScheduler worker unsubscription

[davidmoten]

I noticed that when a worker built from Schedulers.from(Executors.newFixedThread(1)) is unsubscribed the use of a CompositeSubscription to track task subscriptions means that the tasks may be unsubscribed in any old order (CompositeSubscription holds its subscriptions in a HashSet). This means that if the worker is given task A and task B then the race can prevent A from running but allow B to run! I've included a unit test in this PR that demos it. Fails every time on my machine on the first loop.

This PR is really for discussion about the problem and possible fixes.

I have included a possible fix which is to track overall subscription using a volatile boolean and check that boolean before running any task. If this was considered the way to go some further simplification would take place in the operator (might not need to check individual task subscriptions).

I haven't checked other schedulers for this sort of problem yet.

[zsxwing]

Do you have a real case that the order does matter? I think this affects all CompositeSubscription usages.

[akarnokd]

I see this as an eager cancellation matter than a race. There is no need for the flag but just check tasks.isUnsubscribed() before and run some cleanup:

do {
    if (tasks.isUnsubscribed()) {
        queue.clear();
        return;
    }

    ScheduledAction sa = queue.poll();

    if (tasks.isUnsubscribed()) {
        queue.clear();
        return;
    }
    if (!sa.isUnsubscribed()) {
        sa.run();
    }
} while (wip.decrementAndGet() != 0);

[davidmoten]

@zsxwing I'm confusing the issue talking about CompositeSubscription, it's really about subscription management in ExecutorScheduler. I don't have a problem with CompositeSubscription in general.

@akarnokd thanks, that looks good, I'll amend the PR. It's interesting you don't classify it as a bug but I guess we haven't documented our expectations of Schedulers in this area. Do you think Scheduler.unsubscribe() should offer stronger guarantees, similar to the guarantee offered by ExecutorService.shutdownNow()?

Attempts to stop all actively executing tasks, halts the
processing of waiting tasks, and returns a list of the tasks
that were awaiting execution.

The relevant bit I suppose is halts the processing of waiting tasks.

[zsxwing]

@davidmoten Gotcha. EventLoopWorker has the same issue. You can try Schedulers.computation() or Schedulers.io() to reproduce it in your test.

[davidmoten]

@zsxwing computation() and io() both fail too, thanks.

[davidmoten]

@akarnokd Don't you think that second check of tasks.isUnsubscribed() in your snippet is overkill? Can I remove?

I'm also wondering if unsubscribe() can look like this:

    @Override
    public void unsubscribe() {
        queue.clear();
        tasks.unsubscribe();
    }

and then run() would be:

        @Override
        public void run() {
            do {
                if (tasks.isUnsubscribed()) {
                    return;
                }

                ScheduledAction sa = queue.poll();
                if (sa == null) {
                    return;
                }
                if (!sa.isUnsubscribed()) {
                    sa.run();
                }
            } while (wip.decrementAndGet() != 0);
        }

[akarnokd]

No need for stronger guarantees.

That unsubscribe you are suggesting delays the tasks.unsubscribe() and the queue could still be in use; swap the two lines. In addition, you'd still need a clear in the drain loop because it could be still scheduled with a non-empty queue.

[davidmoten]

Thanks @akarnokd . In terms of guarantees I was going to suggest that the javadoc of Scheduler at the class level be supplemented with a statement similar to that of ExecutorService.shutdownNow() once all schedulers had been enhanced like this PR.

[davidmoten]

Updated PR, squashed commits.

[akarnokd]

👍

[zsxwing]

👍

[zsxwing]

@davidmoten mind thinking about how to fix EventLoopWorker? :)

[davidmoten]

@zsxwing sure, I'll have a look at EventLoopWorker

I think there is an outstanding race with this PR:

if (!sa.isUnsubscribed()) {
    sa.run();
}

might have to be:

if (!tasks.isUnsubscribed()) {
    sa.run();
}

I'll have a look a bit later today and I'll submit another PR if needed.

[zsxwing]

if (!sa.isUnsubscribed()) { is necessary. sa may be unsubscribed by the user.

[davidmoten]

Yep I just noticed that myself, ta.

[davidmoten]

@zsxwing I think it should look like this:

        @Override
        public void run() {
            do {
                if (tasks.isUnsubscribed()) {
                    queue.clear();
                    return;
                }
                ScheduledAction sa = queue.poll();
                if (sa == null) {
                    return;
                }
                if (!sa.isUnsubscribed()) {
                    if (!tasks.isUnsubscribed()) {
                        sa.run();
                    } else {
                        queue.clear();
                        return;
                    }
                }
            } while (wip.decrementAndGet() != 0);
        }