View Javadoc
1   /*
2    * Copyright 2019 The Netty Project
3    *
4    * The Netty Project licenses this file to you under the Apache License,
5    * version 2.0 (the "License"); you may not use this file except in compliance
6    * with the License. You may obtain a copy of the License at:
7    *
8    *   https://www.apache.org/licenses/LICENSE-2.0
9    *
10   * Unless required by applicable law or agreed to in writing, software
11   * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12   * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13   * License for the specific language governing permissions and limitations
14   * under the License.
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.EventExecutor;
29  import io.netty.util.concurrent.Future;
30  import io.netty.util.concurrent.Promise;
31  import org.junit.jupiter.api.Test;
32  import org.junit.jupiter.api.Timeout;
33  import org.junit.jupiter.api.function.Executable;
34  
35  import java.util.HashSet;
36  import java.util.Iterator;
37  import java.util.NoSuchElementException;
38  import java.util.Set;
39  import java.util.concurrent.Callable;
40  import java.util.concurrent.CountDownLatch;
41  import java.util.concurrent.RejectedExecutionException;
42  import java.util.concurrent.TimeUnit;
43  
44  import static org.junit.jupiter.api.Assertions.assertEquals;
45  import static org.junit.jupiter.api.Assertions.assertFalse;
46  import static org.junit.jupiter.api.Assertions.assertNotNull;
47  import static org.junit.jupiter.api.Assertions.assertThrows;
48  import static org.junit.jupiter.api.Assertions.assertTrue;
49  import static org.junit.jupiter.api.Assertions.fail;
50  import static org.junit.jupiter.api.Assumptions.assumeTrue;
51  
52  public abstract class AbstractSingleThreadEventLoopTest {
53      @Test
54      @Timeout(value = 5000, unit = TimeUnit.MILLISECONDS)
55      public void testChannelsRegistered() throws Exception {
56          EventLoopGroup group = newEventLoopGroup();
57          final SingleThreadEventLoop loop = (SingleThreadEventLoop) group.next();
58  
59          try {
60              final Channel ch1 = newChannel();
61              final Channel ch2 = newChannel();
62  
63              int rc = registeredChannels(loop);
64              boolean channelCountSupported = rc != -1;
65  
66              if (channelCountSupported) {
67                  assertEquals(0, registeredChannels(loop));
68              }
69  
70              assertTrue(loop.register(ch1).syncUninterruptibly().isSuccess());
71              assertTrue(loop.register(ch2).syncUninterruptibly().isSuccess());
72              if (channelCountSupported) {
73                  checkNumRegisteredChannels(loop, 2);
74              }
75  
76              assertTrue(ch1.deregister().syncUninterruptibly().isSuccess());
77              if (channelCountSupported) {
78                  checkNumRegisteredChannels(loop, 1);
79              }
80          } finally {
81              group.shutdownGracefully();
82          }
83      }
84  
85      private static void checkNumRegisteredChannels(SingleThreadEventLoop loop, int numChannels) throws Exception {
86          // We need to loop as some EventLoop implementations may need some time to update the counter correctly.
87          while (registeredChannels(loop) != numChannels) {
88              Thread.sleep(50);
89          }
90      }
91  
92      // Only reliable if run from event loop
93      private static int registeredChannels(final SingleThreadEventLoop loop) throws Exception {
94          return loop.submit(new Callable<Integer>() {
95              @Override
96              public Integer call() {
97                  return loop.registeredChannels();
98              }
99          }).get(1, TimeUnit.SECONDS);
100     }
101 
102     @Test
103     @SuppressWarnings("deprecation")
104     public void shutdownBeforeStart() throws Exception {
105         EventLoopGroup group = newEventLoopGroup();
106         assertFalse(group.awaitTermination(2, TimeUnit.MILLISECONDS));
107         group.shutdown();
108         assertTrue(group.awaitTermination(200, TimeUnit.MILLISECONDS));
109     }
110 
111     @Test
112     public void shutdownGracefullyZeroQuietBeforeStart() throws Exception {
113         EventLoopGroup group = newEventLoopGroup();
114         assertTrue(group.shutdownGracefully(0L, 2L, TimeUnit.SECONDS).await(200L));
115     }
116 
117     // Copied from AbstractEventLoopTest
118     @Test
119     @Timeout(value = 5000, unit = TimeUnit.MILLISECONDS)
120     public void testShutdownGracefullyNoQuietPeriod() throws Exception {
121         EventLoopGroup loop = newEventLoopGroup();
122         ServerBootstrap b = new ServerBootstrap();
123         b.group(loop)
124         .channel(serverChannelClass())
125         .childHandler(new ChannelInboundHandlerAdapter());
126 
127         // Not close the Channel to ensure the EventLoop is still shutdown in time.
128         ChannelFuture cf = serverChannelClass() == LocalServerChannel.class
129                 ? b.bind(new LocalAddress("local")) : b.bind(0);
130         cf.sync().channel();
131 
132         Future<?> f = loop.shutdownGracefully(0, 1, TimeUnit.MINUTES);
133         assertTrue(loop.awaitTermination(600, TimeUnit.MILLISECONDS));
134         assertTrue(f.syncUninterruptibly().isSuccess());
135         assertTrue(loop.isShutdown());
136         assertTrue(loop.isTerminated());
137     }
138 
139     @Test
140     public void shutdownGracefullyBeforeStart() throws Exception {
141         EventLoopGroup group = newEventLoopGroup();
142         assertTrue(group.shutdownGracefully(200L, 1000L, TimeUnit.MILLISECONDS).await(500L));
143     }
144 
145     @Test
146     public void gracefulShutdownAfterStart() throws Exception {
147         EventLoop loop = newEventLoopGroup().next();
148         final CountDownLatch latch = new CountDownLatch(1);
149         loop.execute(new Runnable() {
150             @Override
151             public void run() {
152                 latch.countDown();
153             }
154         });
155 
156         // Wait for the event loop thread to start.
157         latch.await();
158 
159         // Request the event loop thread to stop.
160         loop.shutdownGracefully(200L, 3000L, TimeUnit.MILLISECONDS);
161 
162         // Wait until the event loop is terminated.
163         assertTrue(loop.awaitTermination(500L, TimeUnit.MILLISECONDS));
164 
165         assertRejection(loop);
166     }
167 
168     @Test
169     @Timeout(value = 3000, unit = TimeUnit.MILLISECONDS)
170     public void testChannelsIteratorEmpty() throws Exception {
171         assumeTrue(supportsChannelIteration());
172         EventLoopGroup group = newEventLoopGroup();
173         final SingleThreadEventLoop loop = (SingleThreadEventLoop) group.next();
174         try {
175             runBlockingOn(loop, new Runnable() {
176                 @Override
177                 public void run() {
178                     final Iterator<Channel> iterator = loop.registeredChannelsIterator();
179 
180                     assertFalse(iterator.hasNext());
181                     assertThrows(NoSuchElementException.class, new Executable() {
182                         @Override
183                         public void execute() {
184                             iterator.next();
185                         }
186                     });
187                 }
188             });
189         } finally {
190             group.shutdownGracefully();
191         }
192     }
193 
194     @Test
195     @Timeout(value = 3000, unit = TimeUnit.MILLISECONDS)
196     public void testChannelsIterator() throws Exception {
197         assumeTrue(supportsChannelIteration());
198         EventLoopGroup group = newEventLoopGroup();
199         final SingleThreadEventLoop loop = (SingleThreadEventLoop) group.next();
200         try {
201             final Channel ch1 = newChannel();
202             final Channel ch2 = newChannel();
203             loop.register(ch1).syncUninterruptibly();
204             loop.register(ch2).syncUninterruptibly();
205             assertEquals(2, registeredChannels(loop));
206 
207             runBlockingOn(loop, new Runnable() {
208                 @Override
209                 public void run() {
210                     final Iterator<Channel> iterator = loop.registeredChannelsIterator();
211 
212                     assertTrue(iterator.hasNext());
213                     Channel actualCh1 = iterator.next();
214                     assertNotNull(actualCh1);
215 
216                     assertTrue(iterator.hasNext());
217                     Channel actualCh2 = iterator.next();
218                     assertNotNull(actualCh2);
219 
220                     Set<Channel> expected = new HashSet<Channel>(4);
221                     expected.add(ch1);
222                     expected.add(ch2);
223                     expected.remove(actualCh1);
224                     expected.remove(actualCh2);
225                     assertTrue(expected.isEmpty());
226 
227                     assertFalse(iterator.hasNext());
228                     assertThrows(NoSuchElementException.class, new Executable() {
229                         @Override
230                         public void execute() {
231                             iterator.next();
232                         }
233                     });
234                 }
235             });
236         } finally {
237             group.shutdownGracefully();
238         }
239     }
240 
241     @Test
242     @Timeout(value = 3000, unit = TimeUnit.MILLISECONDS)
243     public void testChannelsIteratorRemoveThrows() throws Exception {
244         assumeTrue(supportsChannelIteration());
245         EventLoopGroup group = newEventLoopGroup();
246         final SingleThreadEventLoop loop = (SingleThreadEventLoop) group.next();
247 
248         try {
249             final Channel ch = newChannel();
250             loop.register(ch).syncUninterruptibly();
251             assertEquals(1, registeredChannels(loop));
252 
253             runBlockingOn(loop, new Runnable() {
254                 @Override
255                 public void run() {
256                     assertThrows(UnsupportedOperationException.class, new Executable() {
257                         @Override
258                         public void execute() {
259                             loop.registeredChannelsIterator().remove();
260                         }
261                     });
262                 }
263             });
264         } finally {
265             group.shutdownGracefully();
266         }
267     }
268 
269     private static void runBlockingOn(EventLoop eventLoop, final Runnable action) {
270         final Promise<Void> promise = eventLoop.newPromise();
271             eventLoop.execute(new Runnable() {
272                 @Override
273                 public void run() {
274                     try {
275                         action.run();
276                         promise.setSuccess(null);
277                     } catch (Throwable t) {
278                         promise.tryFailure(t);
279                     }
280                 }
281             });
282         try {
283             promise.await();
284         } catch (InterruptedException e) {
285             throw new RuntimeException(e);
286         }
287         Throwable cause = promise.cause();
288         if (cause != null) {
289             if (cause instanceof RuntimeException) {
290                 throw (RuntimeException) cause;
291             }
292             throw new RuntimeException(cause);
293         }
294     }
295 
296     private static final Runnable NOOP = new Runnable() {
297         @Override
298         public void run() { }
299     };
300 
301     private static void assertRejection(EventExecutor loop) {
302         try {
303             loop.execute(NOOP);
304             fail("A task must be rejected after shutdown() is called.");
305         } catch (RejectedExecutionException e) {
306             // Expected
307         }
308     }
309 
310     protected boolean supportsChannelIteration() {
311         return false;
312     }
313     protected abstract EventLoopGroup newEventLoopGroup();
314     protected abstract Channel newChannel();
315     protected abstract Class<? extends ServerChannel> serverChannelClass();
316 }