DefaultEventLoopScheduler implements EventLoopScheduler { DefaultEventLoopScheduler(EventLoopGroup eventLoopGroup, int maxNumEventLoopsPerEndpoint, int maxNumEventLoopsPerHttp1Endpoint, List<ToIntFunction<Endpoint>> maxNumEventLoopsFunctions) { eventLoops = Streams.stream(eventLoopGroup) .map(EventLoop.class::cast) .collect(toImmutableList()); final int eventLoopSize = eventLoops.size(); acquisitionStartIndex = ThreadLocalRandom.current().nextInt(eventLoopSize); if (maxNumEventLoopsPerEndpoint <= 0) { this.maxNumEventLoopsPerEndpoint = DEFAULT_MAX_NUM_EVENT_LOOPS; } else { this.maxNumEventLoopsPerEndpoint = Math.min(maxNumEventLoopsPerEndpoint, eventLoopSize); } if (maxNumEventLoopsPerHttp1Endpoint <= 0) { this.maxNumEventLoopsPerHttp1Endpoint = this.maxNumEventLoopsPerEndpoint; } else { this.maxNumEventLoopsPerHttp1Endpoint = Math.min(maxNumEventLoopsPerHttp1Endpoint, eventLoopSize); } this.maxNumEventLoopsFunctions = ImmutableList.copyOf(maxNumEventLoopsFunctions); } DefaultEventLoopScheduler(EventLoopGroup eventLoopGroup, int maxNumEventLoopsPerEndpoint,
                              int maxNumEventLoopsPerHttp1Endpoint,
                              List<ToIntFunction<Endpoint>> maxNumEventLoopsFunctions); @Override ReleasableHolder<EventLoop> acquire(SessionProtocol sessionProtocol,
                                               EndpointGroup endpointGroup,
                                               @Nullable Endpoint endpoint);  }
@Test void acquireAndRelease() { final DefaultEventLoopScheduler s = defaultEventLoopScheduler(); final AbstractEventLoopEntry e0 = acquireEntry(s, endpoint); final EventLoop loop = e0.get(); assertThat(e0.id()).isZero(); assertThat(e0.activeRequests()).isEqualTo(1); e0.release(); assertThat(e0.activeRequests()).isZero(); for (int i = 0; i < 2; i++) { final AbstractEventLoopEntry e0again = acquireEntry(s, endpoint); assertThat(e0again).isSameAs(e0); assertThat(e0again.id()).isZero(); assertThat(e0again.activeRequests()).isEqualTo(1); assertThat(e0again.get()).isSameAs(loop); e0again.release(); } }
@Test void acquireAndReleaseWithNullEndpoint() { final DefaultEventLoopScheduler s = defaultEventLoopScheduler(); final AbstractEventLoopEntry e0 = acquireEntry(s, null); final EventLoop loop = e0.get(); assertThat(e0.id()).isZero(); assertThat(e0.activeRequests()).isEqualTo(1); e0.release(); assertThat(e0.activeRequests()).isZero(); for (int i = 0; i < 2; i++) { final AbstractEventLoopEntry e0again = acquireEntry(s, null); assertThat(e0again).isSameAs(e0); assertThat(e0again.id()).isZero(); assertThat(e0again.activeRequests()).isEqualTo(1); assertThat(e0again.get()).isSameAs(loop); e0again.release(); } }
@Test void orderedRelease() { final DefaultEventLoopScheduler s = defaultEventLoopScheduler(); final AbstractEventLoopEntry e0 = acquireEntry(s, endpoint); final EventLoop loop1 = e0.get(); assertThat(e0.id()).isZero(); assertThat(e0.activeRequests()).isEqualTo(1); final AbstractEventLoopEntry e1 = acquireEntry(s, endpoint); final EventLoop loop2 = e1.get(); assertThat(e1).isNotSameAs(e0); assertThat(loop2).isNotSameAs(loop1); assertThat(e1.id()).isEqualTo(1); assertThat(e1.activeRequests()).isEqualTo(1); final AbstractEventLoopEntry e2 = acquireEntry(s, endpoint); final EventLoop loop3 = e2.get(); assertThat(e2).isNotSameAs(e0); assertThat(e2).isNotSameAs(e1); assertThat(loop3).isNotSameAs(loop1); assertThat(loop3).isNotSameAs(loop2); assertThat(e2.id()).isEqualTo(2); assertThat(e2.activeRequests()).isEqualTo(1); e0.release(); assertThat(e0.activeRequests()).isZero(); final AbstractEventLoopEntry e0again = acquireEntry(s, endpoint); assertThat(e0again).isSameAs(e0); assertThat(e0again.activeRequests()).isEqualTo(1); e1.release(); assertThat(e1.activeRequests()).isZero(); final AbstractEventLoopEntry e1again = acquireEntry(s, endpoint); assertThat(e1again).isSameAs(e1); assertThat(e1again.activeRequests()).isEqualTo(1); e2.release(); assertThat(e2.activeRequests()).isZero(); final AbstractEventLoopEntry e2again = acquireEntry(s, endpoint); assertThat(e2again).isSameAs(e2); assertThat(e2again.activeRequests()).isEqualTo(1); }
@Test void unorderedRelease() { final DefaultEventLoopScheduler s = defaultEventLoopScheduler(); final AbstractEventLoopEntry e0 = acquireEntry(s, endpoint); final EventLoop loop1 = e0.get(); assertThat(e0.id()).isZero(); assertThat(e0.activeRequests()).isEqualTo(1); final AbstractEventLoopEntry e1 = acquireEntry(s, endpoint); final EventLoop loop2 = e1.get(); assertThat(e1).isNotSameAs(e0); assertThat(loop2).isNotSameAs(loop1); assertThat(e1.id()).isEqualTo(1); assertThat(e1.activeRequests()).isEqualTo(1); final AbstractEventLoopEntry e2 = acquireEntry(s, endpoint); final EventLoop loop3 = e2.get(); assertThat(e2).isNotSameAs(e0); assertThat(e2).isNotSameAs(e1); assertThat(loop3).isNotSameAs(loop1); assertThat(loop3).isNotSameAs(loop2); assertThat(e2.id()).isEqualTo(2); assertThat(e2.activeRequests()).isEqualTo(1); e1.release(); assertThat(e1.activeRequests()).isZero(); final AbstractEventLoopEntry e1again = acquireEntry(s, endpoint); assertThat(e1again).isSameAs(e1); assertThat(e1again.activeRequests()).isEqualTo(1); e2.release(); assertThat(e2.activeRequests()).isZero(); final AbstractEventLoopEntry e2again = acquireEntry(s, endpoint); assertThat(e2again).isSameAs(e2); assertThat(e2again.activeRequests()).isEqualTo(1); e0.release(); assertThat(e0.activeRequests()).isZero(); final AbstractEventLoopEntry e0again = acquireEntry(s, endpoint); assertThat(e0again).isSameAs(e0); assertThat(e0again.activeRequests()).isEqualTo(1); }
@Test void multipleEndpoints() { final DefaultEventLoopScheduler s = defaultEventLoopScheduler(); final Endpoint endpointA = Endpoint.of("a.com"); final Endpoint endpointB = Endpoint.of("b.com"); final Set<AbstractEventLoopEntry> entriesA = new LinkedHashSet<>(); final Set<AbstractEventLoopEntry> entriesB = new LinkedHashSet<>(); for (int i = 0; i < GROUP_SIZE; i++) { entriesA.add(acquireEntry(s, endpointA)); entriesB.add(acquireEntry(s, endpointB)); } assertThat(entriesA).hasSize(GROUP_SIZE); assertThat(entriesB).hasSize(GROUP_SIZE); entriesA.forEach(e -> assertThat(e.activeRequests()).isEqualTo(1)); entriesB.forEach(e -> assertThat(e.activeRequests()).isEqualTo(1)); for (int i = 0; i < GROUP_SIZE; i++) { entriesA.add(acquireEntry(s, endpointA)); } assertThat(entriesA).hasSize(GROUP_SIZE); entriesA.forEach(e -> assertThat(e.activeRequests()).isEqualTo(2)); entriesB.forEach(e -> assertThat(e.activeRequests()).isEqualTo(1)); }