NonStickyOrderedEventExecutor Class — netty Architecture

Source Code

common/src/main/java/io/netty/util/concurrent/NonStickyEventExecutorGroup.java lines 215–347

    private static final class NonStickyOrderedEventExecutor extends AbstractEventExecutor
            implements Runnable, OrderedEventExecutor {
        private final EventExecutor executor;
        private final Queue<Runnable> tasks = PlatformDependent.newMpscQueue();

        private static final int NONE = 0;
        private static final int SUBMITTED = 1;
        private static final int RUNNING = 2;

        private final AtomicInteger state = new AtomicInteger();
        private final int maxTaskExecutePerRun;

        private final AtomicReference<Thread> executingThread = new AtomicReference<Thread>();

        NonStickyOrderedEventExecutor(EventExecutor executor, int maxTaskExecutePerRun) {
            super(executor);
            this.executor = executor;
            this.maxTaskExecutePerRun = maxTaskExecutePerRun;
        }

        @Override
        public void run() {
            if (!state.compareAndSet(SUBMITTED, RUNNING)) {
                return;
            }
            Thread current = Thread.currentThread();
            executingThread.set(current);
            for (;;) {
                int i = 0;
                try {
                    for (; i < maxTaskExecutePerRun; i++) {
                        Runnable task = tasks.poll();
                        if (task == null) {
                            break;
                        }
                        safeExecute(task);
                    }
                } finally {
                    if (i == maxTaskExecutePerRun) {
                        try {
                            state.set(SUBMITTED);
                            // Only set executingThread to null if no other thread did update it yet.
                            executingThread.compareAndSet(current, null);
                            executor.execute(this);
                            return; // done
                        } catch (Throwable ignore) {
                            // Restore executingThread since we're continuing to execute tasks.
                            executingThread.set(current);
                            // Reset the state back to running as we will keep on executing tasks.
                            state.set(RUNNING);
                            // if an error happened we should just ignore it and let the loop run again as there is not
                            // much else we can do. Most likely this was triggered by a full task queue. In this case
                            // we just will run more tasks and try again later.
                        }
                    } else {
                        state.set(NONE);
                        // After setting the state to NONE, look at the tasks queue one more time.
                        // If it is empty, then we can return from this method.
                        // Otherwise, it means the producer thread has called execute(Runnable)
                        // and enqueued a task in between the tasks.poll() above and the state.set(NONE) here.
                        // There are two possible scenarios when this happens
                        //
                        // 1. The producer thread sees state == NONE, hence the compareAndSet(NONE, SUBMITTED)
                        //    is successfully setting the state to SUBMITTED. This mean the producer
                        //    will call / has called executor.execute(this). In this case, we can just return.
                        // 2. The producer thread don't see the state change, hence the compareAndSet(NONE, SUBMITTED)
                        //    returns false. In this case, the producer thread won't call executor.execute.
                        //    In this case, we need to change the state to RUNNING and keeps running.
                        //
                        // The above cases can be distinguished by performing a
                        // compareAndSet(NONE, RUNNING). If it returns "false", it is case 1; otherwise it is case 2.
                        if (tasks.isEmpty() || !state.compareAndSet(NONE, RUNNING)) {
                            // Only set executingThread to null if no other thread did update it yet.
                            executingThread.compareAndSet(current, null);
                            return; // done
                        }
                    }
                }
            }
        }