advanceWithWaiters() — netty Function Reference

Architecture documentation for the advanceWithWaiters() function in DefaultMockTickerTest.java from the netty codebase.

Function java Buffer Telemetry

Entity Profile

Buffer→ Telemetry→ advanceWithWaiters() — netty Function Reference

Dependency Diagram

graph TD
  3d2e9786_8639_114f_37ea_2ade1cf467ad["advanceWithWaiters()"]
  15c7b7ca_db5e_f063_6092_a6f83d0d82e5["DefaultMockTickerTest"]
  3d2e9786_8639_114f_37ea_2ade1cf467ad -->|defined in| 15c7b7ca_db5e_f063_6092_a6f83d0d82e5
  style 3d2e9786_8639_114f_37ea_2ade1cf467ad fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

common/src/test/java/io/netty/util/concurrent/DefaultMockTickerTest.java lines 69–138

    @Timeout(60)
    @Test
    void advanceWithWaiters() throws Exception {
        final List<Thread> threads = new ArrayList<>();
        final DefaultMockTicker ticker = (DefaultMockTicker) Ticker.newMockTicker();
        final int numWaiters = 4;
        final List<FutureTask<Void>> futures = new ArrayList<>();
        for (int i = 0; i < numWaiters; i++) {
            FutureTask<Void> task = new FutureTask<>(() -> {
                try {
                    ticker.sleep(1, TimeUnit.MILLISECONDS);
                } catch (InterruptedException e) {
                    throw new CompletionException(e);
                }
                return null;
            });
            Thread thread = new Thread(task);
            threads.add(thread);
            futures.add(task);
            thread.start();
        }

        try {
            // Wait for all threads to be sleeping.
            for (Thread thread : threads) {
                ticker.awaitSleepingThread(thread);
            }

            // Time did not advance at all, and thus future will not complete.
            for (int i = 0; i < numWaiters; i++) {
                final int finalCnt = i;
                assertThrows(TimeoutException.class, () -> {
                    futures.get(finalCnt).get(1, TimeUnit.MILLISECONDS);
                });
            }

            // Advance just one nanosecond before completion.
            ticker.advance(999_999, TimeUnit.NANOSECONDS);

            // All threads should still be sleeping.
            for (Thread thread : threads) {
                ticker.awaitSleepingThread(thread);
            }

            // Still needs one more nanosecond for our futures.
            for (int i = 0; i < numWaiters; i++) {
                final int finalCnt = i;
                assertThrows(TimeoutException.class, () -> {
                    futures.get(finalCnt).get(1, TimeUnit.MILLISECONDS);
                });
            }

            // Reach at the 1 millisecond mark and ensure the future is complete.
            ticker.advance(1, TimeUnit.NANOSECONDS);
            for (int i = 0; i < numWaiters; i++) {
                futures.get(i).get();
            }
        } catch (InterruptedException ie) {
            for (Thread thread : threads) {
                String name = thread.getName();
                Thread.State state = thread.getState();
                StackTraceElement[] stackTrace = thread.getStackTrace();
                thread.interrupt();
                InterruptedException threadStackTrace = new InterruptedException(name + ": " + state);
                threadStackTrace.setStackTrace(stackTrace);
                ie.addSuppressed(threadStackTrace);
            }
            throw ie;
        }
    }