1.x: improve ExecutorScheduler worker unsubscription some more

[davidmoten]

As per discussion in #3842, there was an outstanding possibility that unsubscription of a Worker would not cancel all tasks waiting in the queue. This PR addresses that possibility. I attempted to provoke the condition in a unit test but didn't manage it. Nethertheless I think this change completes the protection desired in #3842.

I do have mixed feelings about the possible double calling of queue.clear() (once in the run() method and once in the unsubscribe() method. Any preferences?

[akarnokd]

there was an outstanding possibility that unsubscription of a Worker would not cancel all tasks waiting in the queue.

Tasks are tracked in a separate structure for cancellation, which always happens but maybe not that eagerly and not the same order they were submitted.

You should also keep the original up-front clear() part.

[davidmoten]

Thanks @akarnokd, I've updated the PR and squashed commits.

[akarnokd]

👍

[zsxwing]

there was an outstanding possibility that unsubscription of a Worker would not cancel all tasks waiting in the queue

@davidmoten Could you explain how this will happen and why adding if (!tasks.isUnsubscribed()) { could help? Even if tasks.isUnsubscribed returns false, it may become true after you just check it.

[davidmoten]

@zsxwing sure, you're right that that is possible but that scenario doesn't worry me in that I just consider that as not being able to stop work in progress as opposed to work that is queued. So I guess that's all that is happening here because the check of tasks.isUnsubscribed() at the start of of the loop is all that's required to prevent the issue that I was worried about in #3842. This additional check just firms up not running queued work that has been cancelled.

[zsxwing]

This additional check just firms up not running queued work that has been cancelled.

Sounds like adding isUnsubscribed in an operator?

[davidmoten]

Yep, analagous to that

[zsxwing]

👍