testSetCharSequenceMultipleThreads() — netty Function Reference

Source Code

buffer/src/test/java/io/netty/buffer/AbstractByteBufTest.java lines 2953–3011

    @SuppressWarnings("unchecked")
    @ParameterizedTest
    @MethodSource("setCharSequenceCombinations")
    void testSetCharSequenceMultipleThreads(final Charset charset, CharSequenceType charSeqType) throws Exception {
        int bufSize = 32;
        ByteBuf[] bufs = new ByteBuf[16];
        for (int i = 0; i < bufs.length; i++) {
            bufs[i] = newBuffer(bufSize);
        }

        int iterations = 256;
        Semaphore start = new Semaphore(0);
        Semaphore finish = new Semaphore(0);
        char[] cs = new char[(int) (bufSize / charset.newEncoder().maxBytesPerChar())];
        Arrays.fill(cs, 'a');
        final CharSequence str = charSeqType.create(cs);
        ExecutorService executor = Executors.newFixedThreadPool(bufs.length);
        try {
            Future<Void>[] futures = new Future[bufs.length];
            for (int i = 0; i < bufs.length; i++) {
                final ByteBuf buf = bufs[i];
                futures[i] = executor.submit(() -> {
                    finish.release();
                    start.acquire();
                    for (int j = 0; j < iterations; j++) {
                        buf.setCharSequence(0, str, charset);
                    }
                    return null;
                });
            }
            finish.acquire(bufs.length);
            start.release(bufs.length);
            Exception e = null;
            for (Future<Void> future : futures) {
                try {
                    future.get();
                } catch (InterruptedException ex) {
                    if (e != null) {
                        ex.addSuppressed(e);
                    }
                    throw ex; // Propagate interrupted exceptions immediately.
                } catch (ExecutionException ex) {
                    if (e == null) {
                        e = ex;
                    } else {
                        e.addSuppressed(ex);
                    }
                }
            }
            if (e != null) {
                fail("Worker threads failed", e);
            }
        } finally {
            executor.shutdown();
            for (ByteBuf buf : bufs) {
                buf.release();
            }
        }
    }