ManualMultithreadedIoEventLoopGroup Class — netty Architecture

Source Code

transport-native-epoll/src/test/java/io/netty/channel/epoll/ManualEventLoopTest.java lines 137–223

    public static class ManualMultithreadedIoEventLoopGroup extends MultiThreadIoEventLoopGroup {

        private ManualIoEventLoopRunner ioEventLoopRunner;
        private Consumer<ManualIoEventLoopRunner> beforeCanBlock;

        public ManualMultithreadedIoEventLoopGroup(IoHandlerFactory ioHandlerFactory) {
            super(1, ioHandlerFactory);
        }

        @Override
        protected IoEventLoop newChild(Executor executor, IoHandlerFactory ioHandlerFactory, Object... args) {
            this.ioEventLoopRunner = new ManualIoEventLoopRunner(this, ioHandlerFactory,
                    executor, this::beforeCanBlock);
            return ioEventLoopRunner.ioEventLoop;
        }

        protected void beforeCanBlock(Executor executor) {
        }

        private static class ManualIoEventLoopRunner implements Executor {

            private final ManualIoEventLoop ioEventLoop;
            private final Queue<Runnable> otherTasks = new ConcurrentLinkedQueue<>();

            ManualIoEventLoopRunner(IoEventLoopGroup parent, IoHandlerFactory factory,
                                    Executor executor, Consumer<Executor> beforeCanBlock) {
                this.ioEventLoop = new ManualIoEventLoop(parent, null, factory) {
                    @Override
                    protected boolean canBlock() {
                        if (beforeCanBlock != null) {
                            beforeCanBlock.accept(ManualIoEventLoopRunner.this);
                        }
                        return otherTasks.isEmpty();
                    }
                };
                CountDownLatch started = new CountDownLatch(1);
                executor.execute(() -> {
                    ioEventLoop.setOwningThread(Thread.currentThread());
                    // it would force a first init
                    ioEventLoop.runNow();
                    started.countDown();
                    mainLoop();
                });
                try {
                    started.await();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    throw new RuntimeException(e);
                }
            }

            private void mainLoop() {
                while (!ioEventLoop.isShuttingDown()) {
                    ioEventLoop.run(0);
                    Runnable otherTask = otherTasks.poll();
                    if (otherTask != null) {
                        safeExecute(otherTask);
                    }
                }
                while (!ioEventLoop.isTerminated() || !otherTasks.isEmpty()) {
                    ioEventLoop.runNow();
                    Runnable otherTask = otherTasks.poll();
                    if (otherTask != null) {
                        safeExecute(otherTask);
                    }
                }
            }

            private void safeExecute(Runnable task) {
                try {
                    task.run();
                } catch (Throwable ignore) {
                    //
                }
            }

            public void execute(Runnable otherTask) {
                otherTasks.add(otherTask);
                if (ioEventLoop.isShutdown()) {
                    if (otherTasks.remove(otherTask)) {
                        throw new RejectedExecutionException("Event loop shut down");