testDeadlockOnAcquire() — netty Function Reference

Source Code

transport/src/test/java/io/netty/channel/pool/FixedChannelPoolMapDeadlockTest.java lines 42–174

    @Test
    public void testDeadlockOnAcquire() throws Exception {

        final EventLoop threadA1 = new DefaultEventLoop();
        final Bootstrap bootstrapA1 = new Bootstrap()
                .channel(LocalChannel.class).group(threadA1).localAddress(new LocalAddress("A1"));
        final EventLoop threadA2 = new DefaultEventLoop();
        final Bootstrap bootstrapA2 = new Bootstrap()
                .channel(LocalChannel.class).group(threadA2).localAddress(new LocalAddress("A2"));
        final EventLoop threadB1 = new DefaultEventLoop();
        final Bootstrap bootstrapB1 = new Bootstrap()
                .channel(LocalChannel.class).group(threadB1).localAddress(new LocalAddress("B1"));
        final EventLoop threadB2 = new DefaultEventLoop();
        final Bootstrap bootstrapB2 = new Bootstrap()
                .channel(LocalChannel.class).group(threadB2).localAddress(new LocalAddress("B2"));

        final FixedChannelPool poolA1 = new FixedChannelPool(bootstrapA1, NOOP_HANDLER, 1);
        final FixedChannelPool poolA2 = new FixedChannelPool(bootstrapB2, NOOP_HANDLER, 1);
        final FixedChannelPool poolB1 = new FixedChannelPool(bootstrapB1, NOOP_HANDLER, 1);
        final FixedChannelPool poolB2 = new FixedChannelPool(bootstrapA2, NOOP_HANDLER, 1);

        // Synchronize threads on these barriers to ensure order of execution, first wait until each thread is inside
        // the newPool callback, then hold the two threads that should lose the match until the first two returns, then
        // release them to test if they deadlock when trying to release their pools on each other's threads.
        final CyclicBarrier arrivalBarrier = new CyclicBarrier(4);
        final CyclicBarrier releaseBarrier = new CyclicBarrier(3);

        final AbstractChannelPoolMap<String, FixedChannelPool> channelPoolMap =
                new AbstractChannelPoolMap<String, FixedChannelPool>() {

            @Override
            protected FixedChannelPool newPool(String key) {

                // Thread A1 gets a new pool on eventexecutor thread A1 (anywhere but A2 or B2)
                // Thread B1 gets a new pool on eventexecutor thread B1 (anywhere but A2 or B2)
                // Thread A2 gets a new pool on eventexecutor thread B2
                // Thread B2 gets a new pool on eventexecutor thread A2

                if ("A".equals(key)) {
                    if (threadA1.inEventLoop()) {
                        // Thread A1 gets pool A with thread A1
                        await(arrivalBarrier);
                        return poolA1;
                    } else if (threadA2.inEventLoop()) {
                        // Thread A2 gets pool A with thread B2, but only after A1 won
                        await(arrivalBarrier);
                        await(releaseBarrier);
                        return poolA2;
                    }
                } else if ("B".equals(key)) {
                    if (threadB1.inEventLoop()) {
                        // Thread B1 gets pool with thread B1
                        await(arrivalBarrier);
                        return poolB1;
                    } else if (threadB2.inEventLoop()) {
                        // Thread B2 gets pool with thread A2
                        await(arrivalBarrier);
                        await(releaseBarrier);
                        return poolB2;
                    }
                }
                throw new AssertionError("Unexpected key=" + key + " or thread="
                                         + Thread.currentThread().getName());
            }
        };

        // Thread A1 calls ChannelPoolMap.get(A)
        // Thread A2 calls ChannelPoolMap.get(A)
        // Thread B1 calls ChannelPoolMap.get(B)
        // Thread B2 calls ChannelPoolMap.get(B)

        Future<FixedChannelPool> futureA1 = threadA1.submit(new Callable<FixedChannelPool>() {
            @Override
            public FixedChannelPool call() throws Exception {
                return channelPoolMap.get("A");
            }
        });

        Future<FixedChannelPool> futureA2 = threadA2.submit(new Callable<FixedChannelPool>() {
            @Override
            public FixedChannelPool call() throws Exception {