ManualEventLoopTest Class — netty Architecture

Source Code

transport-native-epoll/src/test/java/io/netty/channel/epoll/ManualEventLoopTest.java lines 39–224

@Timeout(5)
public class ManualEventLoopTest {

    @Test
    void firstRacySubmissionMissWakeupEpoll() throws Exception {
        racySubmissionMissWakeup(EpollIoHandler.newFactory(), 1);
    }

    @Test
    void secondRacySubmissionMissWakeupEpoll() throws Exception {
        racySubmissionMissWakeup(EpollIoHandler.newFactory(), 2);
    }

    @Test
    void firstRacyOtherSubmissionMissWakeupEpoll() throws Exception {
        racyOtherSubmissionMissWakeup(EpollIoHandler.newFactory(), 1);
    }

    @Test
    void secondRacyOtherSubmissionMissWakeupEpoll() throws Exception {
        racyOtherSubmissionMissWakeup(EpollIoHandler.newFactory(), 2);
    }

    private void racySubmissionMissWakeup(IoHandlerFactory handlerFactory, long canBlockAttempt)
            throws Exception {
        CyclicBarrier waitBeforeSubmittingTask = new CyclicBarrier(2);
        CountDownLatch taskSubmitted = new CountDownLatch(1);
        AtomicLong canBlock = new AtomicLong(0);
        ManualMultithreadedIoEventLoopGroup group = new ManualMultithreadedIoEventLoopGroup(handlerFactory) {
            @Override
            protected void beforeCanBlock(Executor executor) {
                if (canBlock.incrementAndGet() == canBlockAttempt) {
                    try {
                        waitBeforeSubmittingTask.await();
                    } catch (Throwable ignore) {
                        //
                    }
                    try {
                        taskSubmitted.await();
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }
            }
        };
        waitBeforeSubmittingTask.await();
        // depending on canBlockAttempt this submission could observe an AWAKE event loop
        // or with a setup NONE deadline (e.g. Long.MAX_VALUE).
        // In the latter case, it can be already asleep or ready to do it.
        Future<?> submitted = group.submit(() -> {
        });
        // unblock canBlock
        taskSubmitted.countDown();
        submitted.get();
        group.shutdownGracefully(0, 0, TimeUnit.SECONDS).get();
    }

    private void racyOtherSubmissionMissWakeup(IoHandlerFactory handlerFactory, long canBlockAttempt)
            throws Exception {
        CyclicBarrier waitBeforeSubmittingTask = new CyclicBarrier(2);
        CountDownLatch taskSubmitted = new CountDownLatch(1);
        AtomicLong canBlock = new AtomicLong(0);
        ManualMultithreadedIoEventLoopGroup group = new ManualMultithreadedIoEventLoopGroup(handlerFactory) {
            @Override
            protected void beforeCanBlock(Executor executor) {
                // this should be called when canBlock is called after setting the wakeup flag!
                if (canBlock.incrementAndGet() == canBlockAttempt) {
                    try {
                        waitBeforeSubmittingTask.await();
                    } catch (Throwable ignore) {
                        //
                    }
                    try {
                        taskSubmitted.await();
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }
            }
        };
        waitBeforeSubmittingTask.await();