@ Override public void execute ( final Runnable task ) { checkNotNull ( task ) ; final Runnable submittedTask ; final long oldRunCount ; synchronized ( queue ) { // If the worker is already running (or execute() on the delegate returned successfully, and // the worker has yet to start) then we don't need to start the worker. if ( workerRunningState == RUNNING || workerRunningState == QUEUED ) { queue . add ( task ) ; return ; } oldRunCount = workerRunCount ; // If the worker is not yet running, the delegate Executor might reject our attempt to start // it. To preserve FIFO order and failure atomicity of rejected execution when the same // Runnable is executed more than once, allocate a wrapper that we know is safe to remove by // object identity. // A data structure that returned a removal handle from add() would allow eliminating this // allocation. submittedTask = new Runnable ( ) { @ Override public void run ( ) { task . run ( ) ; } } ; queue . add ( submittedTask ) ; workerRunningState = QUEUING ; } try { executor . execute ( worker ) ; } catch ( RuntimeException | Error t ) { synchronized ( queue ) { boolean removed = ( workerRunningState == IDLE || workerRunningState == QUEUING ) && queue . removeLastOccurrence ( submittedTask ) ; // If the delegate is directExecutor(), the submitted runnable could have thrown a REE. But // that's handled by the log check that catches RuntimeExceptions in the queue worker. if ( ! ( t instanceof RejectedExecutionException ) || removed ) { throw t ; } } return ; } /* * This is an unsynchronized read! After the read, the function returns immediately or acquires * the lock to check again. Since an IDLE state was observed inside the preceding synchronized * block, and reference field assignment is atomic, this may save reacquiring the lock when * another thread or the worker task has cleared the count and set the state. * * <p>When {@link #executor} is a directExecutor(), the value written to * {@code workerRunningState} will be available synchronously, and behaviour will be * deterministic. */ @ SuppressWarnings ( "GuardedBy" ) boolean alreadyMarkedQueued = workerRunningState != QUEUING ; if ( alreadyMarkedQueued ) { return ; } synchronized ( queue ) { if ( workerRunCount == oldRunCount && workerRunningState == QUEUING ) { workerRunningState = QUEUED ; } } }
Adds a task to the queue and makes sure a worker thread is running .