testConcurrentMessageBufferAccess() — netty Function Reference

Source Code

transport/src/test/java/io/netty/channel/local/LocalTransportThreadModelTest.java lines 237–366

    @Test
    @Timeout(value = 30000, unit = TimeUnit.MILLISECONDS)
    @Disabled
    public void testConcurrentMessageBufferAccess() throws Throwable {
        EventLoopGroup l = new MultiThreadIoEventLoopGroup(
                4, new DefaultThreadFactory("l"), LocalIoHandler.newFactory());
        EventExecutorGroup e1 = new DefaultEventExecutorGroup(4, new DefaultThreadFactory("e1"));
        EventExecutorGroup e2 = new DefaultEventExecutorGroup(4, new DefaultThreadFactory("e2"));
        EventExecutorGroup e3 = new DefaultEventExecutorGroup(4, new DefaultThreadFactory("e3"));
        EventExecutorGroup e4 = new DefaultEventExecutorGroup(4, new DefaultThreadFactory("e4"));
        EventExecutorGroup e5 = new DefaultEventExecutorGroup(4, new DefaultThreadFactory("e5"));

        try {
            final MessageForwarder1 h1 = new MessageForwarder1();
            final MessageForwarder2 h2 = new MessageForwarder2();
            final MessageForwarder3 h3 = new MessageForwarder3();
            final MessageForwarder1 h4 = new MessageForwarder1();
            final MessageForwarder2 h5 = new MessageForwarder2();
            final MessageDiscarder  h6 = new MessageDiscarder();

            final Channel ch = new LocalChannel();

            // inbound:  int -> byte[4] -> int -> int -> byte[4] -> int -> /dev/null
            // outbound: int -> int -> byte[4] -> int -> int -> byte[4] -> /dev/null
            ch.pipeline().addLast(h1)
                         .addLast(e1, h2)
                         .addLast(e2, h3)
                         .addLast(e3, h4)
                         .addLast(e4, h5)
                         .addLast(e5, h6);

            l.register(ch).sync().channel().connect(localAddr).sync();

            final int ROUNDS = 1024;
            final int ELEMS_PER_ROUNDS = 8192;
            final int TOTAL_CNT = ROUNDS * ELEMS_PER_ROUNDS;
            for (int i = 0; i < TOTAL_CNT;) {
                final int start = i;
                final int end = i + ELEMS_PER_ROUNDS;
                i = end;

                ch.eventLoop().execute(new Runnable() {
                    @Override
                    public void run() {
                        for (int j = start; j < end; j ++) {
                            ch.pipeline().fireChannelRead(Integer.valueOf(j));
                        }
                    }
                });
            }

            while (h1.inCnt < TOTAL_CNT || h2.inCnt < TOTAL_CNT || h3.inCnt < TOTAL_CNT ||
                    h4.inCnt < TOTAL_CNT || h5.inCnt < TOTAL_CNT || h6.inCnt < TOTAL_CNT) {
                if (h1.exception.get() != null) {
                    throw h1.exception.get();
                }
                if (h2.exception.get() != null) {
                    throw h2.exception.get();
                }
                if (h3.exception.get() != null) {
                    throw h3.exception.get();
                }
                if (h4.exception.get() != null) {
                    throw h4.exception.get();
                }
                if (h5.exception.get() != null) {
                    throw h5.exception.get();
                }
                if (h6.exception.get() != null) {
                    throw h6.exception.get();
                }
                Thread.sleep(10);
            }

            for (int i = 0; i < TOTAL_CNT;) {
                final int start = i;
                final int end = i + ELEMS_PER_ROUNDS;
                i = end;

                ch.pipeline().context(h6).executor().execute(new Runnable() {
                    @Override