Race with CachedThreadScheduler Eviction

[alexwen]

Although I have not been able to find a way to reproduce the issue reliably, I have seen the following stack several times now:

 ! java.util.concurrent.RejectedExecutionException: Task java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask@328cda0 rejected from java.util.concurrent.ScheduledThreadPoolExecutor@2200705d[Terminated, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 3]
 ! at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2047) ~[na:1.8.0_20]
 ! at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:823) [na:1.8.0_20]
 ! at java.util.concurrent.ScheduledThreadPoolExecutor.delayedExecute(ScheduledThreadPoolExecutor.java:326) ~[na:1.8.0_20]
 ! at java.util.concurrent.ScheduledThreadPoolExecutor.schedule(ScheduledThreadPoolExecutor.java:533) ~[na:1.8.0_20]
 ! at java.util.concurrent.ScheduledThreadPoolExecutor.submit(ScheduledThreadPoolExecutor.java:632) ~[na:1.8.0_20]
 ! at rx.internal.schedulers.NewThreadWorker.scheduleActual(NewThreadWorker.java:66) ~[sasquatch_46ada84dd260fa8b7ab66fb20b7ae79f29810947.jar:0.1-SNAPSHOT]
 ! at rx.schedulers.CachedThreadScheduler$EventLoopWorker.schedule(CachedThreadScheduler.java:149) ~[sasquatch_46ada84dd260fa8b7ab66fb20b7ae79f29810947.jar:0.1-SNAPSHOT]
 ! at rx.schedulers.CachedThreadScheduler$EventLoopWorker.schedule(CachedThreadScheduler.java:139) ~[sasquatch_46ada84dd260fa8b7ab66fb20b7ae79f29810947.jar:0.1-SNAPSHOT]

While auditing the CachedThreadScheduler, I noticed a potential race condition when evicting workers from the eviction queue, from CachedThreadScheduler:

        void evictExpiredWorkers() {
            if (!expiringWorkerQueue.isEmpty()) {
                long currentTimestamp = now();

                Iterator<ThreadWorker> threadWorkerIterator = expiringWorkerQueue.iterator();
                while (threadWorkerIterator.hasNext()) {
                    ThreadWorker threadWorker = threadWorkerIterator.next();
                    if (threadWorker.getExpirationTime() <= currentTimestamp) {
                        threadWorkerIterator.remove();
                        threadWorker.unsubscribe();
                    } else {
                        // Queue is ordered with the worker that will expire first in the beginning, so when we
                        // find a non-expired worker we can stop evicting.
                        break;
                    }
                }
            }
        }

The evictor, uses an iterator#remove in order to remove the work from the queue, but it does not check that this operation actually occurred. The iterator, though thread safe, is not atomic for .next and .remove, so, the worker could be pulled from the cache by another thread between the .next and the .remove.

If this occurs then the the NewThreadWorker will be unsubscribed, its executor shutdown, and the next action that executes on that worker would cause the exception above.

In order to fix this race I believe the evictor would need to check that it removed the worker from the queue, before attempting to unsubscribe the worker.

[benjchristensen]

Thanks for the report.