From 22b2c4c3b8be75ad3ea97f115d1bc56fa46069d2 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E5=A4=9A=E5=B7=B4=E8=83=BA?= <1330456228zjp@gmail.com>
Date: Fri, 7 Dec 2018 20:50:00 +0800
Subject: [PATCH] Fix concurrency problem in UniqueIpFilter (#8635)

Motivation:

If two requests from the same IP are reached at the same time, `connected.contains(remoteIp)` may return false in both threads.

Modifications:

Check if there is already a connection with the same IP using return values.

Result:

Become thread safe.
---
 .../handler/ipfilter/UniqueIpFilter.java      |  3 +-
 .../handler/ipfilter/UniqueIpFilterTest.java  | 74 +++++++++++++++++++
 2 files changed, 75 insertions(+), 2 deletions(-)
 create mode 100644 handler/src/test/java/io/netty/handler/ipfilter/UniqueIpFilterTest.java

diff --git a/handler/src/main/java/io/netty/handler/ipfilter/UniqueIpFilter.java b/handler/src/main/java/io/netty/handler/ipfilter/UniqueIpFilter.java
index ac346481d7..15099bb2a3 100644
--- a/handler/src/main/java/io/netty/handler/ipfilter/UniqueIpFilter.java
+++ b/handler/src/main/java/io/netty/handler/ipfilter/UniqueIpFilter.java
@@ -38,10 +38,9 @@ public class UniqueIpFilter extends AbstractRemoteAddressFilter<InetSocketAddres
     @Override
     protected boolean accept(ChannelHandlerContext ctx, InetSocketAddress remoteAddress) throws Exception {
         final InetAddress remoteIp = remoteAddress.getAddress();
-        if (connected.contains(remoteIp)) {
+        if (!connected.add(remoteIp)) {
             return false;
         } else {
-            connected.add(remoteIp);
             ctx.channel().closeFuture().addListener(new ChannelFutureListener() {
                 @Override
                 public void operationComplete(ChannelFuture future) throws Exception {
diff --git a/handler/src/test/java/io/netty/handler/ipfilter/UniqueIpFilterTest.java b/handler/src/test/java/io/netty/handler/ipfilter/UniqueIpFilterTest.java
new file mode 100644
index 0000000000..6e9eb66f3e
--- /dev/null
+++ b/handler/src/test/java/io/netty/handler/ipfilter/UniqueIpFilterTest.java
@@ -0,0 +1,74 @@
+/*
+ * Copyright 2018 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.handler.ipfilter;
+
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.embedded.EmbeddedChannel;
+import io.netty.util.internal.SocketUtils;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.net.SocketAddress;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+public class UniqueIpFilterTest {
+
+    @Test
+    public void testUniqueIpFilterHandler() throws ExecutionException, InterruptedException {
+        final CyclicBarrier barrier = new CyclicBarrier(2);
+        ExecutorService executorService = Executors.newFixedThreadPool(2);
+        try {
+            for (int round = 0; round < 10000; round++) {
+                final UniqueIpFilter ipFilter = new UniqueIpFilter();
+                Future<EmbeddedChannel> future1 = newChannelAsync(barrier, executorService, ipFilter);
+                Future<EmbeddedChannel> future2 = newChannelAsync(barrier, executorService, ipFilter);
+                EmbeddedChannel ch1 = future1.get();
+                EmbeddedChannel ch2 = future2.get();
+                Assert.assertTrue(ch1.isActive() || ch2.isActive());
+                Assert.assertFalse(ch1.isActive() && ch2.isActive());
+
+                barrier.reset();
+                ch1.close().await();
+                ch2.close().await();
+            }
+        } finally {
+            executorService.shutdown();
+        }
+    }
+
+    private static Future<EmbeddedChannel> newChannelAsync(final CyclicBarrier barrier,
+            ExecutorService executorService,
+            final ChannelHandler... handler) {
+        return executorService.submit(new Callable<EmbeddedChannel>() {
+            @Override
+            public EmbeddedChannel call() throws Exception {
+                barrier.await();
+                return new EmbeddedChannel(handler) {
+                    @Override
+                    protected SocketAddress remoteAddress0() {
+                        return isActive() ? SocketUtils.socketAddress("91.92.93.1", 5421) : null;
+                    }
+                };
+            }
+        });
+    }
+
+}