1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package io.netty.testsuite.transport;
17
18 import io.netty.bootstrap.ServerBootstrap;
19 import io.netty.channel.Channel;
20 import io.netty.channel.ChannelFuture;
21 import io.netty.channel.ChannelInboundHandlerAdapter;
22 import io.netty.channel.EventLoop;
23 import io.netty.channel.EventLoopGroup;
24 import io.netty.channel.ServerChannel;
25 import io.netty.channel.SingleThreadEventLoop;
26 import io.netty.channel.local.LocalAddress;
27 import io.netty.channel.local.LocalServerChannel;
28 import io.netty.util.concurrent.AutoScalingEventExecutorChooserFactory;
29 import io.netty.util.concurrent.EventExecutor;
30 import io.netty.util.concurrent.EventExecutorChooserFactory;
31 import io.netty.util.concurrent.Future;
32 import io.netty.util.concurrent.Promise;
33 import io.netty.util.concurrent.ScheduledFuture;
34 import io.netty.util.internal.PlatformDependent;
35 import org.junit.jupiter.api.Test;
36 import org.junit.jupiter.api.Timeout;
37 import org.junit.jupiter.api.function.Executable;
38
39 import java.util.ArrayList;
40 import java.util.HashSet;
41 import java.util.Iterator;
42 import java.util.List;
43 import java.util.NoSuchElementException;
44 import java.util.Set;
45 import java.util.concurrent.Callable;
46 import java.util.concurrent.CountDownLatch;
47 import java.util.concurrent.RejectedExecutionException;
48 import java.util.concurrent.TimeUnit;
49
50 import static java.util.concurrent.TimeUnit.MILLISECONDS;
51 import static org.junit.jupiter.api.Assertions.assertEquals;
52 import static org.junit.jupiter.api.Assertions.assertFalse;
53 import static org.junit.jupiter.api.Assertions.assertNotNull;
54 import static org.junit.jupiter.api.Assertions.assertThrows;
55 import static org.junit.jupiter.api.Assertions.assertTrue;
56 import static org.junit.jupiter.api.Assumptions.assumeTrue;
57
58 public abstract class AbstractSingleThreadEventLoopTest {
59 protected static final int SCALING_MIN_THREADS = 1;
60 protected static final int SCALING_MAX_THREADS = 2;
61 protected static final long SCALING_WINDOW_SECONDS = 100;
62 protected static final TimeUnit SCALING_WINDOW_UNIT = MILLISECONDS;
63 protected static final double SCALEDOWN_THRESHOLD = 0.2;
64 protected static final double SCALEUP_THRESHOLD = 0.9;
65 protected static final int SCALING_PATIENCE_CYCLES = 1;
66
67 protected static final EventExecutorChooserFactory AUTO_SCALING_CHOOSER_FACTORY =
68 new AutoScalingEventExecutorChooserFactory(SCALING_MIN_THREADS, SCALING_MAX_THREADS, SCALING_WINDOW_SECONDS,
69 SCALING_WINDOW_UNIT, SCALEDOWN_THRESHOLD, SCALEUP_THRESHOLD,
70 SCALING_MAX_THREADS, SCALING_MAX_THREADS, SCALING_PATIENCE_CYCLES
71 );
72
73 @Test
74 @Timeout(value = 5000, unit = TimeUnit.MILLISECONDS)
75 public void testChannelsRegistered() throws Exception {
76 EventLoopGroup group = newEventLoopGroup();
77 final SingleThreadEventLoop loop = (SingleThreadEventLoop) group.next();
78
79 try {
80 final Channel ch1 = newChannel();
81 final Channel ch2 = newChannel();
82
83 int rc = registeredChannels(loop);
84 boolean channelCountSupported = rc != -1;
85
86 if (channelCountSupported) {
87 assertEquals(0, registeredChannels(loop));
88 }
89
90 assertTrue(loop.register(ch1).syncUninterruptibly().isSuccess());
91 assertTrue(loop.register(ch2).syncUninterruptibly().isSuccess());
92 if (channelCountSupported) {
93 checkNumRegisteredChannels(loop, 2);
94 }
95
96 assertTrue(ch1.deregister().syncUninterruptibly().isSuccess());
97 if (channelCountSupported) {
98 checkNumRegisteredChannels(loop, 1);
99 }
100 } finally {
101 group.shutdownGracefully();
102 }
103 }
104
105 private static void checkNumRegisteredChannels(SingleThreadEventLoop loop, int numChannels) throws Exception {
106
107 while (registeredChannels(loop) != numChannels) {
108 Thread.sleep(50);
109 }
110 }
111
112
113 private static int registeredChannels(final SingleThreadEventLoop loop) throws Exception {
114 return loop.submit(new Callable<Integer>() {
115 @Override
116 public Integer call() {
117 return loop.registeredChannels();
118 }
119 }).get(1, TimeUnit.SECONDS);
120 }
121
122 @Test
123 @SuppressWarnings("deprecation")
124 public void shutdownBeforeStart() throws Exception {
125 EventLoopGroup group = newEventLoopGroup();
126 assertFalse(group.awaitTermination(2, TimeUnit.MILLISECONDS));
127 group.shutdown();
128 assertTrue(group.awaitTermination(200, TimeUnit.MILLISECONDS));
129 }
130
131 @Test
132 public void shutdownGracefullyZeroQuietBeforeStart() throws Exception {
133 EventLoopGroup group = newEventLoopGroup();
134 assertTrue(group.shutdownGracefully(0L, 2L, TimeUnit.SECONDS).await(200L));
135 }
136
137
138 @Test
139 @Timeout(value = 5000, unit = TimeUnit.MILLISECONDS)
140 public void testShutdownGracefullyNoQuietPeriod() throws Exception {
141 EventLoopGroup loop = newEventLoopGroup();
142 ServerBootstrap b = new ServerBootstrap();
143 b.group(loop)
144 .channel(serverChannelClass())
145 .childHandler(new ChannelInboundHandlerAdapter());
146
147
148 ChannelFuture cf = serverChannelClass() == LocalServerChannel.class
149 ? b.bind(new LocalAddress(AbstractSingleThreadEventLoopTest.class)) : b.bind(0);
150 cf.sync().channel();
151
152 Future<?> f = loop.shutdownGracefully(0, 1, TimeUnit.MINUTES);
153 assertTrue(loop.awaitTermination(600, TimeUnit.MILLISECONDS));
154 assertTrue(f.syncUninterruptibly().isSuccess());
155 assertTrue(loop.isShutdown());
156 assertTrue(loop.isTerminated());
157 }
158
159 @Test
160 public void shutdownGracefullyBeforeStart() throws Exception {
161 EventLoopGroup group = newEventLoopGroup();
162 assertTrue(group.shutdownGracefully(200L, 1000L, TimeUnit.MILLISECONDS).await(500L));
163 }
164
165 @Test
166 public void gracefulShutdownAfterStart() throws Exception {
167 EventLoop loop = newEventLoopGroup().next();
168 final CountDownLatch latch = new CountDownLatch(1);
169 loop.execute(new Runnable() {
170 @Override
171 public void run() {
172 latch.countDown();
173 }
174 });
175
176
177 latch.await();
178
179
180 loop.shutdownGracefully(200L, 3000L, TimeUnit.MILLISECONDS);
181
182
183 assertTrue(loop.awaitTermination(500L, TimeUnit.MILLISECONDS));
184
185 assertRejection(loop);
186 }
187
188 @Test
189 @Timeout(value = 3000, unit = TimeUnit.MILLISECONDS)
190 public void testChannelsIteratorEmpty() throws Exception {
191 assumeTrue(supportsChannelIteration());
192 EventLoopGroup group = newEventLoopGroup();
193 final SingleThreadEventLoop loop = (SingleThreadEventLoop) group.next();
194 try {
195 runBlockingOn(loop, new Runnable() {
196 @Override
197 public void run() {
198 final Iterator<Channel> iterator = loop.registeredChannelsIterator();
199
200 assertFalse(iterator.hasNext());
201 assertThrows(NoSuchElementException.class, new Executable() {
202 @Override
203 public void execute() {
204 iterator.next();
205 }
206 });
207 }
208 });
209 } finally {
210 group.shutdownGracefully();
211 }
212 }
213
214 @Test
215 @Timeout(value = 3000, unit = TimeUnit.MILLISECONDS)
216 public void testChannelsIterator() throws Exception {
217 assumeTrue(supportsChannelIteration());
218 EventLoopGroup group = newEventLoopGroup();
219 final SingleThreadEventLoop loop = (SingleThreadEventLoop) group.next();
220 try {
221 final Channel ch1 = newChannel();
222 final Channel ch2 = newChannel();
223 loop.register(ch1).syncUninterruptibly();
224 loop.register(ch2).syncUninterruptibly();
225 assertEquals(2, registeredChannels(loop));
226
227 runBlockingOn(loop, new Runnable() {
228 @Override
229 public void run() {
230 final Iterator<Channel> iterator = loop.registeredChannelsIterator();
231
232 assertTrue(iterator.hasNext());
233 Channel actualCh1 = iterator.next();
234 assertNotNull(actualCh1);
235
236 assertTrue(iterator.hasNext());
237 Channel actualCh2 = iterator.next();
238 assertNotNull(actualCh2);
239
240 Set<Channel> expected = new HashSet<Channel>(4);
241 expected.add(ch1);
242 expected.add(ch2);
243 expected.remove(actualCh1);
244 expected.remove(actualCh2);
245 assertTrue(expected.isEmpty());
246
247 assertFalse(iterator.hasNext());
248 assertThrows(NoSuchElementException.class, new Executable() {
249 @Override
250 public void execute() {
251 iterator.next();
252 }
253 });
254 }
255 });
256 } finally {
257 group.shutdownGracefully();
258 }
259 }
260
261 @Test
262 @Timeout(value = 3000, unit = TimeUnit.MILLISECONDS)
263 public void testChannelsIteratorRemoveThrows() throws Exception {
264 assumeTrue(supportsChannelIteration());
265 EventLoopGroup group = newEventLoopGroup();
266 final SingleThreadEventLoop loop = (SingleThreadEventLoop) group.next();
267
268 try {
269 final Channel ch = newChannel();
270 loop.register(ch).syncUninterruptibly();
271 assertEquals(1, registeredChannels(loop));
272
273 runBlockingOn(loop, new Runnable() {
274 @Override
275 public void run() {
276 assertThrows(UnsupportedOperationException.class, new Executable() {
277 @Override
278 public void execute() {
279 loop.registeredChannelsIterator().remove();
280 }
281 });
282 }
283 });
284 } finally {
285 group.shutdownGracefully();
286 }
287 }
288
289 @Test
290 void schedulingAndCancellingTasks() throws Exception {
291 Runnable runnable = new Runnable() {
292 @Override
293 public void run() {
294 }
295 };
296 List<ScheduledFuture<?>> tasks = new ArrayList<ScheduledFuture<?>>();
297 EventLoopGroup group = newEventLoopGroup();
298 try {
299 EventLoop eventLoop = group.next();
300 for (int i = 0; i < 5000; i++) {
301 tasks.add(eventLoop.scheduleAtFixedRate(runnable, 1, 1, MILLISECONDS));
302 if (tasks.size() > 500) {
303 tasks.get(PlatformDependent.threadLocalRandom().nextInt(tasks.size())).cancel(false);
304 }
305 }
306 for (ScheduledFuture<?> task : tasks) {
307 task.cancel(false);
308 }
309 for (ScheduledFuture<?> task : tasks) {
310 task.await();
311 }
312 for (ScheduledFuture<?> task : tasks) {
313 if (!task.isCancelled()) {
314 task.sync();
315 }
316 }
317 } finally {
318 group.shutdownGracefully();
319 }
320 }
321
322 @Test
323 @Timeout(30)
324 public void testAutoScalingEventLoopGroupCanScaleDownAndBeUsed() throws Exception {
325 EventLoopGroup group = newAutoScalingEventLoopGroup();
326 if (group == null) {
327 return;
328 }
329 try {
330 startAllExecutors(group);
331 assertEquals(SCALING_MAX_THREADS, countActiveExecutors(group),
332 "Group should start with max threads active.");
333
334 long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
335 while (countActiveExecutors(group) > SCALING_MIN_THREADS && System.nanoTime() < deadline) {
336 Thread.sleep(100);
337 }
338
339 assertEquals(SCALING_MIN_THREADS, countActiveExecutors(group),
340 "Group did not scale down to min threads in time.");
341 } finally {
342 group.shutdownGracefully().syncUninterruptibly();
343 }
344 }
345
346 @Test
347 @Timeout(30)
348 public void testSubmittingTaskWakesUpSuspendedExecutor() throws Exception {
349 EventLoopGroup group = newAutoScalingEventLoopGroup();
350 if (group == null) {
351 return;
352 }
353 try {
354 startAllExecutors(group);
355 assertEquals(SCALING_MAX_THREADS, countActiveExecutors(group),
356 "Group should start with max threads.");
357
358 long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
359 while (countActiveExecutors(group) > SCALING_MIN_THREADS && System.nanoTime() < deadline) {
360 Thread.sleep(100);
361 }
362 assertEquals(SCALING_MIN_THREADS, countActiveExecutors(group),
363 "Group did not scale down to min threads in time.");
364
365 EventLoop suspendedLoop = null;
366 for (EventExecutor exec : group) {
367 if (exec.isSuspended()) {
368 suspendedLoop = (EventLoop) exec;
369 break;
370 }
371 }
372 assertNotNull(suspendedLoop, "Could not find a suspended executor to test.");
373
374
375 int i = 0;
376 boolean isSuspended;
377 do {
378 Future<?> future = suspendedLoop.submit(() -> { });
379 future.sync();
380 } while ((isSuspended = suspendedLoop.isSuspended()) && i++ < 10);
381
382 assertFalse(isSuspended, "Executor should wake up after task submission.");
383 assertEquals(SCALING_MAX_THREADS, countActiveExecutors(group),
384 "Active executor count should increase after wake-up.");
385 } finally {
386 group.shutdownGracefully().sync();
387 }
388 }
389
390 private static int countActiveExecutors(EventLoopGroup group) {
391 int activeCount = 0;
392 for (EventExecutor executor : group) {
393 if (!executor.isSuspended()) {
394 activeCount++;
395 }
396 }
397 return activeCount;
398 }
399
400 private static void startAllExecutors(EventLoopGroup group) throws InterruptedException {
401 CountDownLatch startLatch = new CountDownLatch(SCALING_MAX_THREADS);
402 for (EventExecutor executor : group) {
403 executor.execute(startLatch::countDown);
404 }
405 startLatch.await();
406 }
407
408 private static void runBlockingOn(EventLoop eventLoop, final Runnable action) {
409 final Promise<Void> promise = eventLoop.newPromise();
410 eventLoop.execute(new Runnable() {
411 @Override
412 public void run() {
413 try {
414 action.run();
415 promise.setSuccess(null);
416 } catch (Throwable t) {
417 promise.tryFailure(t);
418 }
419 }
420 });
421 try {
422 promise.await();
423 } catch (InterruptedException e) {
424 throw new RuntimeException(e);
425 }
426 Throwable cause = promise.cause();
427 if (cause != null) {
428 if (cause instanceof RuntimeException) {
429 throw (RuntimeException) cause;
430 }
431 throw new RuntimeException(cause);
432 }
433 }
434
435 private static final Runnable NOOP = new Runnable() {
436 @Override
437 public void run() { }
438 };
439
440 private static void assertRejection(EventExecutor loop) {
441 assertThrows(RejectedExecutionException.class, () -> loop.execute(NOOP));
442 }
443
444 protected boolean supportsChannelIteration() {
445 return false;
446 }
447 protected abstract EventLoopGroup newEventLoopGroup();
448 protected abstract EventLoopGroup newAutoScalingEventLoopGroup();
449 protected abstract Channel newChannel();
450 protected abstract Class<? extends ServerChannel> serverChannelClass();
451 }