testDeadlockOnRemove() — netty Function Reference

Source Code

transport/src/test/java/io/netty/channel/pool/FixedChannelPoolMapDeadlockTest.java lines 176–242

    @Test
    public void testDeadlockOnRemove() throws Exception {

        final EventLoop thread1 = new DefaultEventLoop();
        final Bootstrap bootstrap1 = new Bootstrap()
                .channel(LocalChannel.class).group(thread1).localAddress(new LocalAddress("#1"));
        final EventLoop thread2 = new DefaultEventLoop();
        final Bootstrap bootstrap2 = new Bootstrap()
                .channel(LocalChannel.class).group(thread2).localAddress(new LocalAddress("#2"));

        // pool1 runs on thread2, pool2 runs on thread1
        final FixedChannelPool pool1 = new FixedChannelPool(bootstrap2, NOOP_HANDLER, 1);
        final FixedChannelPool pool2 = new FixedChannelPool(bootstrap1, NOOP_HANDLER, 1);

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

            @Override
            protected FixedChannelPool newPool(String key) {
                if ("#1".equals(key)) {
                    return pool1;
                } else if ("#2".equals(key)) {
                    return pool2;
                } else {
                    throw new AssertionError("Unexpected key=" + key);
                }
            }
        };

        assertSame(pool1, channelPoolMap.get("#1"));
        assertSame(pool2, channelPoolMap.get("#2"));

        // thread1 tries to remove pool1 which is running on thread2
        // thread2 tries to remove pool2 which is running on thread1

        final CyclicBarrier barrier = new CyclicBarrier(2);

        Future<?> future1 = thread1.submit(new Runnable() {
            @Override
            public void run() {
                await(barrier);
                channelPoolMap.remove("#1");
            }
        });

        Future<?> future2 = thread2.submit(new Runnable() {
            @Override
            public void run() {
                await(barrier);
                channelPoolMap.remove("#2");
            }
        });

        // A blocking close on remove will cause a deadlock here and the test will time out
        try {
            future1.get(1, TimeUnit.SECONDS);
            future2.get(1, TimeUnit.SECONDS);
        } catch (TimeoutException e) {
            // Fail the test on timeout to distinguish from other errors
            throw new AssertionError(e);
        } finally {
            pool1.close();
            pool2.close();
            channelPoolMap.close();
            shutdown(thread1, thread2);
        }
    }