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.
This commit is contained in:
多巴胺 2018-12-07 20:50:00 +08:00 committed by Norman Maurer
parent 2b651eb1a2
commit 22b2c4c3b8
2 changed files with 75 additions and 2 deletions

View File

@ -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 {

View File

@ -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;
}
};
}
});
}
}