Repository: netty/netty
File: transport-native-epoll/src/test/java/io/netty/channel/epoll/ManualEventLoopTest.java (Java)

Code before:
        group.shutdownGracefully(0, 0, TimeUnit.SECONDS).get();
    }

    @RepeatedTest(value = 100, failureThreshold = 1)
    void testTightShutodownNIO() throws InterruptedException, ExecutionException {
        ManualMultithreadedIoEventLoopGroup group = new ManualMultithreadedIoEventLoopGroup(
                NioIoHandler.newFactory());
        group.shutdownGracefully(0, 0, TimeUnit.SECONDS).get();
    }

    @Test
    void testTightSubmissionsEpoll() throws ExecutionException, InterruptedException {
        ManualMultithreadedIoEventLoopGroup group = new ManualMultithreadedIoEventLoopGroup(
                EpollIoHandler.newFactory());
        for (int i = 0; i < 1000; i++) {
            group.submit(() -> {
            }).get();
        }
        group.shutdownGracefully(0, 0, TimeUnit.SECONDS).get();
    }

    @Test
    void testTightSubmissionsNIO() throws ExecutionException, InterruptedException {
        ManualMultithreadedIoEventLoopGroup group = new ManualMultithreadedIoEventLoopGroup(
                NioIoHandler.newFactory());
        for (int i = 0; i < 1000; i++) {
            group.submit(() -> {
            }).get();
        }
        group.shutdownGracefully(0, 0, TimeUnit.SECONDS).get();
    }

    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) {
        }


Review comment (suggestion):
@normanmaurer this way to use the manual event loop looks correct/idiomatic in your opinion?

It's basically a simplified version of https://github.com/franz1981/Netty-VirtualThread-Scheduler enabling the manual event loop to be used to consume/handle other type of tasks than the task ones.

Code after:
    void testTightShutodownEpoll() throws InterruptedException, ExecutionException {
        ManualMultithreadedIoEventLoopGroup group = new ManualMultithreadedIoEventLoopGroup(
                EpollIoHandler.newFactory());
        group.shutdownGracefully(0, 0, TimeUnit.SECONDS).get();
    }

    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();