You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by je...@apache.org on 2020/05/08 15:16:45 UTC
[geode] branch develop updated: GEODE-7998: Do not block Netty
threads when publishing (#5042)
This is an automated email from the ASF dual-hosted git repository.
jensdeppe pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/geode.git
The following commit(s) were added to refs/heads/develop by this push:
new 5542db5 GEODE-7998: Do not block Netty threads when publishing (#5042)
5542db5 is described below
commit 5542db5a85ff8ac683fde5b6cff12279e3910ad4
Author: Jens Deppe <jd...@pivotal.io>
AuthorDate: Fri May 8 08:16:14 2020 -0700
GEODE-7998: Do not block Netty threads when publishing (#5042)
- Now delegate PUBLISH responses to a background executor in order to
free up the Netty worker thread. This prevents possible deadlocks.
Co-authored-by: Murtuza Boxwala <mb...@vmware.com>
Co-authored-by: Ray Ingles <ri...@vmware.com>
Co-authored-by: Jens Deppe <jd...@vmware.com>
---
.../test/junit/rules/ExecutorServiceRule.java | 4 +-
.../org/apache/geode/redis/MockSubscriber.java | 16 +
.../org/apache/geode/redis/PubSubDUnitTest.java | 325 +++++++++++++++++++--
.../apache/geode/redis/PubSubIntegrationTest.java | 5 +
.../geode/redis/internal/DummySubscription.java | 5 +-
.../redis/mocks/MockSubscriberWithLatch.java} | 7 +-
.../org/apache/geode/redis/GeodeRedisServer.java | 12 +-
.../geode/redis/internal/AbstractSubscription.java | 41 +--
.../org/apache/geode/redis/internal/Client.java | 4 +
.../redis/internal/ExecutionHandlerContext.java | 14 +-
.../apache/geode/redis/internal/PubSubImpl.java | 52 ++--
.../redis/internal/PublishResultCollector.java | 60 ++++
.../apache/geode/redis/internal/Subscription.java | 3 +-
.../internal/executor/pubsub/PublishExecutor.java | 29 +-
.../executor/pubsub/UnsubscribeExecutor.java | 10 +-
.../geode/redis/internal/PubSubImplJUnitTest.java | 46 ++-
16 files changed, 534 insertions(+), 99 deletions(-)
diff --git a/geode-junit/src/main/java/org/apache/geode/test/junit/rules/ExecutorServiceRule.java b/geode-junit/src/main/java/org/apache/geode/test/junit/rules/ExecutorServiceRule.java
index bed8243..e3a6a67 100644
--- a/geode-junit/src/main/java/org/apache/geode/test/junit/rules/ExecutorServiceRule.java
+++ b/geode-junit/src/main/java/org/apache/geode/test/junit/rules/ExecutorServiceRule.java
@@ -328,7 +328,7 @@ public class ExecutorServiceRule extends SerializableExternalResource {
* a {@code Set<WeakReference<Thread>>} to track the {@code Thread}s in the factory's
* {@code ThreadGroup} excluding subgroups.
*/
- protected static class DedicatedThreadFactory implements ThreadFactory {
+ public static class DedicatedThreadFactory implements ThreadFactory {
private static final AtomicInteger POOL_NUMBER = new AtomicInteger(1);
@@ -337,7 +337,7 @@ public class ExecutorServiceRule extends SerializableExternalResource {
private final String namePrefix;
private final Set<WeakReference<Thread>> directThreads = new HashSet<>();
- protected DedicatedThreadFactory() {
+ public DedicatedThreadFactory() {
group = new ThreadGroup(ExecutorServiceRule.class.getSimpleName() + "-ThreadGroup");
namePrefix = "pool-" + POOL_NUMBER.getAndIncrement() + "-thread-";
}
diff --git a/geode-redis/src/distributedTest/java/org/apache/geode/redis/MockSubscriber.java b/geode-redis/src/distributedTest/java/org/apache/geode/redis/MockSubscriber.java
index f3d8381..5ac9c38 100644
--- a/geode-redis/src/distributedTest/java/org/apache/geode/redis/MockSubscriber.java
+++ b/geode-redis/src/distributedTest/java/org/apache/geode/redis/MockSubscriber.java
@@ -20,11 +20,18 @@ import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
+import org.apache.logging.log4j.Logger;
+import redis.clients.jedis.Client;
import redis.clients.jedis.JedisPubSub;
+import org.apache.geode.logging.internal.log4j.api.LogService;
+
public class MockSubscriber extends JedisPubSub {
private CountDownLatch latch;
private List<String> receivedMessages = new ArrayList<String>();
+ private Client client;
+
+ private static final Logger logger = LogService.getLogger();
public MockSubscriber(CountDownLatch latch) {
this.latch = latch;
@@ -36,6 +43,7 @@ public class MockSubscriber extends JedisPubSub {
@Override
public void onSubscribe(String channel, int subscribedChannels) {
+ // logger.info("--->>> Received subscription for " + client.getSocket());
latch.countDown();
}
@@ -43,4 +51,12 @@ public class MockSubscriber extends JedisPubSub {
public void onMessage(String channel, String message) {
receivedMessages.add(message);
}
+
+ @Override
+ public void proceed(Client client, String... channels) {
+ this.client = client;
+ // logger.info("--->>> Before for " + client.getSocket());
+ super.proceed(client, channels);
+ // logger.info("--->>> After for " + client.getSocket());
+ }
}
diff --git a/geode-redis/src/distributedTest/java/org/apache/geode/redis/PubSubDUnitTest.java b/geode-redis/src/distributedTest/java/org/apache/geode/redis/PubSubDUnitTest.java
index c55a96b..cd8762a 100644
--- a/geode-redis/src/distributedTest/java/org/apache/geode/redis/PubSubDUnitTest.java
+++ b/geode-redis/src/distributedTest/java/org/apache/geode/redis/PubSubDUnitTest.java
@@ -16,63 +16,170 @@
package org.apache.geode.redis;
+import static java.lang.String.valueOf;
+import static org.apache.geode.distributed.ConfigurationProperties.LOG_LEVEL;
+import static org.apache.geode.distributed.ConfigurationProperties.MAX_WAIT_TIME_RECONNECT;
+import static org.apache.geode.distributed.ConfigurationProperties.REDIS_BIND_ADDRESS;
+import static org.apache.geode.distributed.ConfigurationProperties.REDIS_PORT;
+import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
import static org.assertj.core.api.Assertions.assertThat;
+import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Properties;
+import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
+import org.junit.AfterClass;
+import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Test;
import redis.clients.jedis.Jedis;
-import org.apache.geode.distributed.ConfigurationProperties;
import org.apache.geode.internal.AvailablePortHelper;
import org.apache.geode.test.awaitility.GeodeAwaitility;
import org.apache.geode.test.dunit.rules.ClusterStartupRule;
import org.apache.geode.test.dunit.rules.MemberVM;
import org.apache.geode.test.junit.rules.ExecutorServiceRule;
+import org.apache.geode.test.junit.rules.GfshCommandRule;
public class PubSubDUnitTest {
public static final String CHANNEL_NAME = "salutations";
@ClassRule
- public static ClusterStartupRule cluster = new ClusterStartupRule();
+ public static ClusterStartupRule cluster = new ClusterStartupRule(5);
+
+ @ClassRule
+ public static GfshCommandRule gfsh = new GfshCommandRule();
@ClassRule
public static ExecutorServiceRule executor = new ExecutorServiceRule();
private static int[] ports;
- private static MemberVM server1, server2;
+ static final String LOCAL_HOST = "127.0.0.1";
+ static Jedis subscriber1;
+ static Jedis subscriber2;
+ static Jedis publisher1;
+ static Jedis publisher2;
+
+ static Properties locatorProperties;
+ static Properties serverProperties1;
+ static Properties serverProperties2;
+ static Properties serverProperties3;
+ static Properties serverProperties4;
+
+ static MemberVM locator;
+ static MemberVM server1;
+ static MemberVM server2;
+ static MemberVM server3;
+ static MemberVM server4;
@BeforeClass
- public static void beforeClass() {
- ports = AvailablePortHelper.getRandomAvailableTCPPorts(2);
- Properties redisProps = new Properties();
- redisProps.put(ConfigurationProperties.REDIS_BIND_ADDRESS, "localhost");
+ public static void beforeClass() throws Exception {
+ ports = AvailablePortHelper.getRandomAvailableTCPPorts(4);
+
+ locatorProperties = new Properties();
+ serverProperties1 = new Properties();
+ serverProperties2 = new Properties();
+ serverProperties3 = new Properties();
+ serverProperties4 = new Properties();
+
+ locatorProperties.setProperty(MAX_WAIT_TIME_RECONNECT, "15000");
+
+ serverProperties1.setProperty(REDIS_PORT, valueOf(ports[0]));
+ serverProperties1.setProperty(REDIS_BIND_ADDRESS, LOCAL_HOST);
+ serverProperties1.setProperty(LOG_LEVEL, "warn");
+
+ serverProperties2.setProperty(REDIS_PORT, valueOf(ports[1]));
+ serverProperties2.setProperty(REDIS_BIND_ADDRESS, LOCAL_HOST);
+ serverProperties2.setProperty(LOG_LEVEL, "warn");
+
+ serverProperties3.setProperty(REDIS_PORT, valueOf(ports[2]));
+ serverProperties3.setProperty(REDIS_BIND_ADDRESS, LOCAL_HOST);
+ serverProperties3.setProperty(LOG_LEVEL, "warn");
+
+ serverProperties4.setProperty(REDIS_PORT, valueOf(ports[3]));
+ serverProperties4.setProperty(REDIS_BIND_ADDRESS, LOCAL_HOST);
+ serverProperties4.setProperty(LOG_LEVEL, "warn");
+
+ locator = cluster.startLocatorVM(0, locatorProperties);
+ server1 = cluster.startServerVM(1, serverProperties1, locator.getPort());
+ server2 = cluster.startServerVM(2, serverProperties2, locator.getPort());
+ server3 = cluster.startServerVM(3, serverProperties3, locator.getPort());
+ server4 = cluster.startServerVM(4, serverProperties4, locator.getPort());
+
+ subscriber1 = new Jedis(LOCAL_HOST, ports[0], 120000);
+ subscriber2 = new Jedis(LOCAL_HOST, ports[1], 120000);
+ publisher1 = new Jedis(LOCAL_HOST, ports[2], 120000);
+ publisher2 = new Jedis(LOCAL_HOST, ports[3], 120000);
+
+
+
+ gfsh.connectAndVerify(locator);
+ }
- MemberVM locator = cluster.startLocatorVM(0);
+ @Before
+ public void testSetup() {
+ subscriber1.flushAll();
+ subscriber2.flushAll();
+ }
- redisProps.put(ConfigurationProperties.REDIS_PORT, Integer.toString(ports[0]));
- server1 = cluster.startServerVM(1, redisProps, locator.getPort());
- redisProps.put(ConfigurationProperties.REDIS_PORT, Integer.toString(ports[1]));
- server2 = cluster.startServerVM(2, redisProps, locator.getPort());
+ @AfterClass
+ public static void tearDown() {
+ subscriber1.disconnect();
+ subscriber2.disconnect();
+ publisher1.disconnect();
+ publisher2.disconnect();
+
+ server1.stop();
+ server2.stop();
+ server3.stop();
+ server4.stop();
}
@Test
- public void testSubscribePublishUsingDifferentServers() throws Exception {
- Jedis subscriber1 = new Jedis("localhost", ports[0]);
- Jedis subscriber2 = new Jedis("localhost", ports[1]);
- Jedis publisher = new Jedis("localhost", ports[1]);
+ public void shouldContinueToFunction_whenOneServerShutsDownGracefully_givenTwoSubscribersOnePublisher()
+ throws InterruptedException {
+ CountDownLatch latch = new CountDownLatch(2);
+
+ MockSubscriber mockSubscriber1 = new MockSubscriber(latch);
+ MockSubscriber mockSubscriber2 = new MockSubscriber(latch);
+
+ Future<Void> subscriber1Future = executor.submit(
+ () -> subscriber1.subscribe(mockSubscriber1, CHANNEL_NAME));
+ Future<Void> subscriber2Future = executor.submit(
+ () -> subscriber2.subscribe(mockSubscriber2, CHANNEL_NAME));
+
+ assertThat(latch.await(30, TimeUnit.SECONDS))
+ .as("channel subscription was not received")
+ .isTrue();
+ Long result = publisher1.publish(CHANNEL_NAME, "hello");
+ assertThat(result).isEqualTo(2);
+
+ server1.stop();
+ Long resultFromSecondMessage = publisher1.publish(CHANNEL_NAME, "hello again");
+ assertThat(resultFromSecondMessage).isEqualTo(1);
+
+ mockSubscriber2.unsubscribe(CHANNEL_NAME);
+ GeodeAwaitility.await().untilAsserted(subscriber2Future::get);
+
+ restartServerVM1();
+ reconnectSubscriber1();
+ }
+
+ @Test
+ public void shouldContinueToFunction_whenOneServerShutsDownAbruptly_givenTwoSubscribersOnePublisher()
+ throws InterruptedException {
CountDownLatch latch = new CountDownLatch(2);
+
MockSubscriber mockSubscriber1 = new MockSubscriber(latch);
MockSubscriber mockSubscriber2 = new MockSubscriber(latch);
@@ -85,7 +192,103 @@ public class PubSubDUnitTest {
.as("channel subscription was not received")
.isTrue();
- Long result = publisher.publish(CHANNEL_NAME, "hello");
+ Long result = publisher1.publish(CHANNEL_NAME, "hello");
+ assertThat(result).isEqualTo(2);
+
+ cluster.crashVM(2);
+
+ // Depending on the timing of this call, it may catch a function error (due to member departed)
+ // and return 0 as a result. Regardless, it should NOT hang.
+ result = publisher1.publish(CHANNEL_NAME, "hello again");
+ assertThat(result).isLessThanOrEqualTo(1);
+
+ mockSubscriber1.unsubscribe(CHANNEL_NAME);
+
+ GeodeAwaitility.await().untilAsserted(subscriber1Future::get);
+
+ restartServerVM2();
+ reconnectSubscriber2();
+ }
+
+ private void restartServerVM1() {
+ cluster.startServerVM(1, serverProperties1, locator.getPort());
+ await()
+ .untilAsserted(() -> gfsh.executeAndAssertThat("list members")
+ .statusIsSuccess()
+ .hasTableSection()
+ .hasColumn("Name")
+ .containsOnly("locator-0", "server-1", "server-2", "server-3", "server-4"));
+ }
+
+ private void restartServerVM2() {
+ cluster.startServerVM(2, serverProperties2, locator.getPort());
+ await()
+ .untilAsserted(() -> gfsh.executeAndAssertThat("list members")
+ .statusIsSuccess()
+ .hasTableSection()
+ .hasColumn("Name")
+ .containsOnly("locator-0", "server-1", "server-2", "server-3", "server-4"));
+ }
+
+ private void reconnectSubscriber1() {
+ subscriber1 = new Jedis(LOCAL_HOST, ports[0]);
+ }
+
+ private void reconnectSubscriber2() {
+ subscriber2 = new Jedis(LOCAL_HOST, ports[1]);
+ }
+
+ @Test
+ public void shouldContinueToFunction_whenOneServerShutsDownGracefully_givenTwoSubscribersTwoPublishers()
+ throws InterruptedException {
+ CountDownLatch latch = new CountDownLatch(2);
+
+ MockSubscriber mockSubscriber1 = new MockSubscriber(latch);
+ MockSubscriber mockSubscriber2 = new MockSubscriber(latch);
+
+ Future<Void> subscriber1Future = executor.submit(
+ () -> subscriber1.subscribe(mockSubscriber1, CHANNEL_NAME));
+ Future<Void> subscriber2Future = executor.submit(
+ () -> subscriber2.subscribe(mockSubscriber2, CHANNEL_NAME));
+
+ assertThat(latch.await(30, TimeUnit.SECONDS))
+ .as("channel subscription was not received")
+ .isTrue();
+
+ Long resultPublisher1 = publisher1.publish(CHANNEL_NAME, "hello");
+ Long resultPublisher2 = publisher2.publish(CHANNEL_NAME, "hello");
+ assertThat(resultPublisher1).isEqualTo(2);
+ assertThat(resultPublisher2).isEqualTo(2);
+
+ server2.stop();
+
+ publisher1.publish(CHANNEL_NAME, "hello again");
+ publisher2.publish(CHANNEL_NAME, "hello again");
+
+ mockSubscriber1.unsubscribe(CHANNEL_NAME);
+
+ GeodeAwaitility.await().untilAsserted(subscriber1Future::get);
+
+ restartServerVM2();
+ reconnectSubscriber2();
+ }
+
+ @Test
+ public void testSubscribePublishUsingDifferentServers() throws Exception {
+ CountDownLatch latch = new CountDownLatch(2);
+ MockSubscriber mockSubscriber1 = new MockSubscriber(latch);
+ MockSubscriber mockSubscriber2 = new MockSubscriber(latch);
+
+ Future<Void> subscriber1Future = executor.submit(
+ () -> subscriber1.subscribe(mockSubscriber1, CHANNEL_NAME));
+ Future<Void> subscriber2Future = executor.submit(
+ () -> subscriber2.subscribe(mockSubscriber2, CHANNEL_NAME));
+
+ assertThat(latch.await(30, TimeUnit.SECONDS))
+ .as("channel subscription was not received")
+ .isTrue();
+
+ Long result = publisher1.publish(CHANNEL_NAME, "hello");
assertThat(result).isEqualTo(2);
mockSubscriber1.unsubscribe(CHANNEL_NAME);
@@ -100,9 +303,6 @@ public class PubSubDUnitTest {
int CLIENT_COUNT = 10;
int ITERATIONS = 1000;
- Jedis subscriber1 = new Jedis("localhost", ports[0]);
- Jedis subscriber2 = new Jedis("localhost", ports[1]);
-
CountDownLatch latch = new CountDownLatch(2);
MockSubscriber mockSubscriber1 = new MockSubscriber(latch);
MockSubscriber mockSubscriber2 = new MockSubscriber(latch);
@@ -118,12 +318,13 @@ public class PubSubDUnitTest {
List<Future<Void>> futures = new LinkedList<>();
for (int i = 0; i < CLIENT_COUNT; i++) {
- Jedis client = new Jedis("localhost", ports[i % 2]);
+ Jedis publisher = new Jedis("localhost", ports[i % 2]);
Callable<Void> callable = () -> {
for (int j = 0; j < ITERATIONS; j++) {
- client.publish(CHANNEL_NAME, "hello");
+ publisher.publish(CHANNEL_NAME, "hello");
}
+ publisher.close();
return null;
};
@@ -144,4 +345,84 @@ public class PubSubDUnitTest {
assertThat(mockSubscriber2.getReceivedMessages().size()).isEqualTo(CLIENT_COUNT * ITERATIONS);
}
+ @Test
+ public void testPubSubWithMoreSubscribersThanNettyWorkerThreads() throws Exception {
+ int CLIENT_COUNT = 1000;
+ String CHANNEL_NAME = "best_channel_ever";
+
+ List<Jedis> clients = new ArrayList<>();
+
+ // Build up an initial set of subscribers
+ for (int i = 0; i < CLIENT_COUNT; i++) {
+ Jedis client = new Jedis("localhost", ports[0]);
+ clients.add(client);
+
+ CountDownLatch latch = new CountDownLatch(1);
+ MockSubscriber mockSubscriber = new MockSubscriber(latch);
+ executor.submit(() -> client.subscribe(mockSubscriber, CHANNEL_NAME));
+ latch.await();
+
+ clients.add(client);
+ }
+
+ Jedis publishingClient = new Jedis("localhost", ports[0]);
+ long result = 0;
+
+ for (int i = 0; i < 10; i++) {
+ result += publishingClient.publish(CHANNEL_NAME, "this is amazing");
+ }
+
+ assertThat(result).isEqualTo(CLIENT_COUNT * 10);
+
+ clients.forEach(Jedis::close);
+ }
+
+ @Test
+ public void testPubSubWithManyClientsDisconnecting() throws Exception {
+ int CLIENT_COUNT = 10;
+ int ITERATIONS = 400;
+
+ Random random = new Random();
+ List<Jedis> clients = new ArrayList<>();
+
+ // Build up an initial set of subscribers
+ for (int i = 0; i < CLIENT_COUNT; i++) {
+ Jedis client = new Jedis("localhost", ports[0]);
+ clients.add(client);
+
+ CountDownLatch latch = new CountDownLatch(1);
+ MockSubscriber mockSubscriber = new MockSubscriber(latch);
+ executor.submit(() -> client.subscribe(mockSubscriber, CHANNEL_NAME));
+ latch.await();
+ }
+
+ // Start actively publishing in the background
+ Jedis publishingClient = new Jedis("localhost", ports[0]);
+ Callable<Void> callable = () -> {
+ for (int j = 0; j < ITERATIONS; j++) {
+ publishingClient.publish(CHANNEL_NAME, "hello - " + j);
+ }
+ return null;
+ };
+
+ Future<Void> future = executor.submit(callable);
+
+ // Abnormally close and recreate new subscribers without unsubscribing
+ for (int i = 0; i < ITERATIONS; i++) {
+ int candy = random.nextInt(CLIENT_COUNT);
+ clients.get(candy).close();
+
+ Jedis client = new Jedis("localhost", ports[0]);
+ CountDownLatch latch = new CountDownLatch(1);
+ MockSubscriber mockSubscriber = new MockSubscriber(latch);
+ executor.submit(() -> client.subscribe(mockSubscriber, CHANNEL_NAME));
+ latch.await();
+
+ clients.set(candy, client);
+ }
+
+ GeodeAwaitility.await().untilAsserted(() -> future.get());
+
+ clients.forEach(Jedis::close);
+ }
}
diff --git a/geode-redis/src/integrationTest/java/org/apache/geode/redis/PubSubIntegrationTest.java b/geode-redis/src/integrationTest/java/org/apache/geode/redis/PubSubIntegrationTest.java
index f50d74b..58bbd38 100644
--- a/geode-redis/src/integrationTest/java/org/apache/geode/redis/PubSubIntegrationTest.java
+++ b/geode-redis/src/integrationTest/java/org/apache/geode/redis/PubSubIntegrationTest.java
@@ -27,6 +27,7 @@ import java.util.concurrent.Callable;
import org.junit.AfterClass;
import org.junit.BeforeClass;
+import org.junit.ClassRule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import redis.clients.jedis.Jedis;
@@ -38,6 +39,7 @@ import org.apache.geode.redis.mocks.MockBinarySubscriber;
import org.apache.geode.redis.mocks.MockSubscriber;
import org.apache.geode.test.awaitility.GeodeAwaitility;
import org.apache.geode.test.junit.categories.RedisTest;
+import org.apache.geode.test.junit.rules.ExecutorServiceRule;
@Category({RedisTest.class})
public class PubSubIntegrationTest {
@@ -48,6 +50,9 @@ public class PubSubIntegrationTest {
private static GemFireCache cache;
private static int port = 6379;
+ @ClassRule
+ public static ExecutorServiceRule executor = new ExecutorServiceRule();
+
@BeforeClass
public static void setUp() {
CacheFactory cf = new CacheFactory();
diff --git a/geode-redis/src/integrationTest/java/org/apache/geode/redis/internal/DummySubscription.java b/geode-redis/src/integrationTest/java/org/apache/geode/redis/internal/DummySubscription.java
index 8868ad4..f34af1b 100644
--- a/geode-redis/src/integrationTest/java/org/apache/geode/redis/internal/DummySubscription.java
+++ b/geode-redis/src/integrationTest/java/org/apache/geode/redis/internal/DummySubscription.java
@@ -26,9 +26,8 @@ public class DummySubscription implements Subscription {
}
@Override
- public PublishResult publishMessage(String channel, byte[] message) {
- return null;
- }
+ public void publishMessage(String channel, byte[] message,
+ PublishResultCollector publishResultCollector) {}
@Override
public boolean matchesClient(Client client) {
diff --git a/geode-redis/src/distributedTest/java/org/apache/geode/redis/MockSubscriber.java b/geode-redis/src/integrationTest/java/org/apache/geode/redis/mocks/MockSubscriberWithLatch.java
similarity index 89%
copy from geode-redis/src/distributedTest/java/org/apache/geode/redis/MockSubscriber.java
copy to geode-redis/src/integrationTest/java/org/apache/geode/redis/mocks/MockSubscriberWithLatch.java
index f3d8381..a3a1476 100644
--- a/geode-redis/src/distributedTest/java/org/apache/geode/redis/MockSubscriber.java
+++ b/geode-redis/src/integrationTest/java/org/apache/geode/redis/mocks/MockSubscriberWithLatch.java
@@ -14,7 +14,7 @@
*
*/
-package org.apache.geode.redis;
+package org.apache.geode.redis.mocks;
import java.util.ArrayList;
import java.util.List;
@@ -22,11 +22,12 @@ import java.util.concurrent.CountDownLatch;
import redis.clients.jedis.JedisPubSub;
-public class MockSubscriber extends JedisPubSub {
+
+public class MockSubscriberWithLatch extends JedisPubSub {
private CountDownLatch latch;
private List<String> receivedMessages = new ArrayList<String>();
- public MockSubscriber(CountDownLatch latch) {
+ public MockSubscriberWithLatch(CountDownLatch latch) {
this.latch = latch;
}
diff --git a/geode-redis/src/main/java/org/apache/geode/redis/GeodeRedisServer.java b/geode-redis/src/main/java/org/apache/geode/redis/GeodeRedisServer.java
index b703d3c..2e1a5b0 100644
--- a/geode-redis/src/main/java/org/apache/geode/redis/GeodeRedisServer.java
+++ b/geode-redis/src/main/java/org/apache/geode/redis/GeodeRedisServer.java
@@ -58,6 +58,7 @@ import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
+import io.netty.handler.timeout.WriteTimeoutHandler;
import io.netty.util.concurrent.Future;
import org.apache.geode.GemFireConfigException;
@@ -332,16 +333,15 @@ public class GeodeRedisServer {
*/
private int setNumWorkerThreads() {
String prop = System.getProperty(NUM_THREADS_SYS_PROP_NAME);
- int numCores = Runtime.getRuntime().availableProcessors();
- int def = 4 * numCores;
+ int defaultThreads = Runtime.getRuntime().availableProcessors();
if (prop == null || prop.isEmpty()) {
- return def;
+ return defaultThreads;
}
int threads;
try {
threads = Integer.parseInt(prop);
} catch (NumberFormatException e) {
- return def;
+ return defaultThreads;
}
return threads;
}
@@ -620,11 +620,13 @@ public class GeodeRedisServer {
ChannelPipeline p = ch.pipeline();
addSSLIfEnabled(ch, p);
p.addLast(ByteToCommandDecoder.class.getSimpleName(), new ByteToCommandDecoder());
+ p.addLast(new WriteTimeoutHandler(10));
p.addLast(ExecutionHandlerContext.class.getSimpleName(),
new ExecutionHandlerContext(ch, cache, regionCache, GeodeRedisServer.this, pwdB,
keyRegistrar, pubSub, hashLockService));
}
- }).option(ChannelOption.SO_REUSEADDR, true).option(ChannelOption.SO_RCVBUF, getBufferSize())
+ }).option(ChannelOption.SO_REUSEADDR, true)
+ .option(ChannelOption.SO_RCVBUF, getBufferSize())
.childOption(ChannelOption.SO_KEEPALIVE, true)
.childOption(ChannelOption.CONNECT_TIMEOUT_MILLIS, GeodeRedisServer.connectTimeoutMillis)
.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
diff --git a/geode-redis/src/main/java/org/apache/geode/redis/internal/AbstractSubscription.java b/geode-redis/src/main/java/org/apache/geode/redis/internal/AbstractSubscription.java
index 1bc3e6f..9a7ce65 100644
--- a/geode-redis/src/main/java/org/apache/geode/redis/internal/AbstractSubscription.java
+++ b/geode-redis/src/main/java/org/apache/geode/redis/internal/AbstractSubscription.java
@@ -16,11 +16,8 @@
package org.apache.geode.redis.internal;
-import java.nio.channels.ClosedChannelException;
-import java.util.concurrent.ExecutionException;
-
import io.netty.buffer.ByteBuf;
-import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
import org.apache.logging.log4j.Logger;
import org.apache.geode.logging.internal.log4j.api.LogService;
@@ -42,13 +39,10 @@ public abstract class AbstractSubscription implements Subscription {
}
@Override
- public PublishResult publishMessage(String channel, byte[] message) {
+ public void publishMessage(String channel, byte[] message,
+ PublishResultCollector publishResultCollector) {
ByteBuf messageByteBuffer = constructResponse(channel, message);
- if (messageByteBuffer == null) {
- return new PublishResult(client, false);
- }
-
- return new PublishResult(client, writeToChannelSynchronously(messageByteBuffer));
+ writeToChannel(messageByteBuffer, publishResultCollector);
}
Client getClient() {
@@ -77,23 +71,14 @@ public abstract class AbstractSubscription implements Subscription {
* to the client, resulted in an error - for example if the client has disconnected and the write
* fails. In such cases we need to be able to notify the caller.
*/
- private boolean writeToChannelSynchronously(ByteBuf messageByteBuffer) {
- ChannelFuture channelFuture = context.writeToChannel(messageByteBuffer);
-
- try {
- channelFuture.get();
- } catch (ExecutionException e) {
- if (e.getCause() instanceof ClosedChannelException) {
- logger.warn("Unable to write to channel: {}", e.getMessage());
- } else {
- logger.warn("Unable to write to channel", e);
- }
- return false;
- } catch (InterruptedException e) {
- logger.warn("Unable to write to channel", e);
- return false;
- }
-
- return channelFuture.cause() == null;
+ private void writeToChannel(ByteBuf messageByteBuffer, PublishResultCollector resultCollector) {
+ context.writeToChannel(messageByteBuffer)
+ .addListener((ChannelFutureListener) future -> {
+ if (future.cause() == null) {
+ resultCollector.success();
+ } else {
+ resultCollector.failure(client);
+ }
+ });
}
}
diff --git a/geode-redis/src/main/java/org/apache/geode/redis/internal/Client.java b/geode-redis/src/main/java/org/apache/geode/redis/internal/Client.java
index 44a31a7..a543567 100644
--- a/geode-redis/src/main/java/org/apache/geode/redis/internal/Client.java
+++ b/geode-redis/src/main/java/org/apache/geode/redis/internal/Client.java
@@ -46,4 +46,8 @@ public class Client {
public boolean isDead() {
return !this.channel.isOpen();
}
+
+ public String toString() {
+ return channel.toString();
+ }
}
diff --git a/geode-redis/src/main/java/org/apache/geode/redis/internal/ExecutionHandlerContext.java b/geode-redis/src/main/java/org/apache/geode/redis/internal/ExecutionHandlerContext.java
index 5a422d1..f8964a5 100644
--- a/geode-redis/src/main/java/org/apache/geode/redis/internal/ExecutionHandlerContext.java
+++ b/geode-redis/src/main/java/org/apache/geode/redis/internal/ExecutionHandlerContext.java
@@ -96,7 +96,7 @@ public class ExecutionHandlerContext extends ChannelInboundHandlerAdapter {
* @param cache The Geode cache instance of this vm
* @param regionProvider The region provider of this context
* @param server Instance of the server it is attached to, only used so that any execution
- * can initiate a shutdwon
+ * can initiate a shutdown
* @param password Authentication password for each context, can be null
*/
public ExecutionHandlerContext(Channel channel, Cache cache, RegionProvider regionProvider,
@@ -147,9 +147,7 @@ public class ExecutionHandlerContext extends ChannelInboundHandlerAdapter {
}
executeCommand(ctx, command);
} catch (Exception e) {
- logger.error(
- "Execution of Redis command " + command + " failed",
- e);
+ logger.warn("Execution of Redis command {} failed: {}", command, e);
throw e;
}
@@ -176,6 +174,7 @@ public class ExecutionHandlerContext extends ChannelInboundHandlerAdapter {
&& cause.getCause() instanceof RedisCommandParserException) {
response =
Coder.getErrorResponse(this.byteBufAllocator, RedisConstants.PARSING_EXCEPTION_MESSAGE);
+
} else if (cause instanceof RegionCreationException) {
this.logger.error(cause);
response =
@@ -221,8 +220,11 @@ public class ExecutionHandlerContext extends ChannelInboundHandlerAdapter {
writeToChannel(
Coder.getSimpleStringResponse(this.byteBufAllocator, RedisConstants.COMMAND_QUEUED));
} else {
- ByteBuf response = command.getResponse();
- writeToChannel(response);
+ // PUBLISH responses are always deferred
+ if (command.getCommandType() != RedisCommandType.PUBLISH) {
+ ByteBuf response = command.getResponse();
+ writeToChannel(response);
+ }
}
} else if (command.isOfType(RedisCommandType.QUIT)) {
command.execute(this);
diff --git a/geode-redis/src/main/java/org/apache/geode/redis/internal/PubSubImpl.java b/geode-redis/src/main/java/org/apache/geode/redis/internal/PubSubImpl.java
index c5e52e5..cd6082e 100644
--- a/geode-redis/src/main/java/org/apache/geode/redis/internal/PubSubImpl.java
+++ b/geode-redis/src/main/java/org/apache/geode/redis/internal/PubSubImpl.java
@@ -17,14 +17,15 @@
package org.apache.geode.redis.internal;
import java.util.List;
-import java.util.Map;
-import java.util.stream.Collectors;
+
+import org.apache.logging.log4j.Logger;
import org.apache.geode.annotations.VisibleForTesting;
import org.apache.geode.cache.execute.Function;
import org.apache.geode.cache.execute.FunctionContext;
import org.apache.geode.cache.execute.FunctionService;
import org.apache.geode.cache.execute.ResultCollector;
+import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.redis.internal.org.apache.hadoop.fs.GlobPattern;
/**
@@ -35,6 +36,8 @@ import org.apache.geode.redis.internal.org.apache.hadoop.fs.GlobPattern;
public class PubSubImpl implements PubSub {
public static final String REDIS_PUB_SUB_FUNCTION_ID = "redisPubSubFunctionID";
+ private static final Logger logger = LogService.getLogger();
+
private final Subscriptions subscriptions;
public PubSubImpl(Subscriptions subscriptions) {
@@ -51,7 +54,13 @@ public class PubSubImpl implements PubSub {
.setArguments(new Object[] {channel, message})
.execute(REDIS_PUB_SUB_FUNCTION_ID);
- List<Long> subscriberCounts = subscriberCountCollector.getResult();
+ List<Long> subscriberCounts = null;
+ try {
+ subscriberCounts = subscriberCountCollector.getResult();
+ } catch (Exception e) {
+ logger.warn("Failed to execute publish function {}", e.getMessage());
+ return 0;
+ }
return subscriberCounts.stream().mapToLong(x -> x).sum();
}
@@ -91,6 +100,17 @@ public class PubSubImpl implements PubSub {
publishMessageToSubscribers((String) publishMessage[0], (byte[]) publishMessage[1]);
context.getResultSender().lastResult(subscriberCount);
}
+
+ /**
+ * Since the publish process uses an onMembers function call, we don't want to re-publish
+ * to members if one fails.
+ * TODO: Revisit this in the event that we instead use an onMember call against individual
+ * members.
+ */
+ @Override
+ public boolean isHA() {
+ return false;
+ }
});
}
@@ -109,23 +129,19 @@ public class PubSubImpl implements PubSub {
@VisibleForTesting
long publishMessageToSubscribers(String channel, byte[] message) {
- Map<Boolean, List<PublishResult>> results = subscriptions
- .findSubscriptions(channel)
- .stream()
- .map(subscription -> subscription.publishMessage(channel, message))
- .collect(Collectors.partitioningBy(PublishResult::isSuccessful));
+ List<Subscription> foundSubscriptions = subscriptions
+ .findSubscriptions(channel);
+ if (foundSubscriptions.isEmpty()) {
+ return 0;
+ }
- prune(results.get(false));
+ PublishResultCollector publishResultCollector =
+ new PublishResultCollector(foundSubscriptions.size(), subscriptions);
- return results.get(true).size();
- }
+ foundSubscriptions.forEach(
+ subscription -> subscription.publishMessage(channel, message, publishResultCollector));
- private void prune(List<PublishResult> failedSubscriptions) {
- failedSubscriptions.forEach(publishResult -> {
- Client client = publishResult.getClient();
- if (client.isDead()) {
- subscriptions.remove(client);
- }
- });
+ return publishResultCollector.getSuccessCount();
}
+
}
diff --git a/geode-redis/src/main/java/org/apache/geode/redis/internal/PublishResultCollector.java b/geode-redis/src/main/java/org/apache/geode/redis/internal/PublishResultCollector.java
new file mode 100644
index 0000000..b05c177
--- /dev/null
+++ b/geode-redis/src/main/java/org/apache/geode/redis/internal/PublishResultCollector.java
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF 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 org.apache.geode.redis.internal;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicLong;
+
+
+public class PublishResultCollector {
+ private final CountDownLatch countDownLatch;
+ private final Subscriptions subscriptions;
+ private final AtomicLong successCount = new AtomicLong();
+
+ public PublishResultCollector(int resultsExpected, Subscriptions subscriptions) {
+ countDownLatch = new CountDownLatch(resultsExpected);
+ this.subscriptions = subscriptions;
+ }
+
+ public void success() {
+ successCount.incrementAndGet();
+ countDownLatch.countDown();
+ }
+
+ public void failure(Client result) {
+ pruneFailure(result);
+ countDownLatch.countDown();
+ }
+
+ public long getSuccessCount() {
+ waitForResults();
+ return successCount.get();
+ }
+
+ private void pruneFailure(Client c) {
+ if (c.isDead()) {
+ subscriptions.remove(c);
+ }
+ }
+
+ private void waitForResults() {
+ try {
+ countDownLatch.await();
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+}
diff --git a/geode-redis/src/main/java/org/apache/geode/redis/internal/Subscription.java b/geode-redis/src/main/java/org/apache/geode/redis/internal/Subscription.java
index c087d93..67649a7 100644
--- a/geode-redis/src/main/java/org/apache/geode/redis/internal/Subscription.java
+++ b/geode-redis/src/main/java/org/apache/geode/redis/internal/Subscription.java
@@ -32,7 +32,8 @@ public interface Subscription {
/**
* Will publish a message to the designated channel.
*/
- PublishResult publishMessage(String channel, byte[] message);
+ void publishMessage(String channel, byte[] message,
+ PublishResultCollector publishResultCollector);
/**
* Verifies that the subscription is established with the designated client.
diff --git a/geode-redis/src/main/java/org/apache/geode/redis/internal/executor/pubsub/PublishExecutor.java b/geode-redis/src/main/java/org/apache/geode/redis/internal/executor/pubsub/PublishExecutor.java
index 7512989..a3b438e 100644
--- a/geode-redis/src/main/java/org/apache/geode/redis/internal/executor/pubsub/PublishExecutor.java
+++ b/geode-redis/src/main/java/org/apache/geode/redis/internal/executor/pubsub/PublishExecutor.java
@@ -16,6 +16,8 @@
package org.apache.geode.redis.internal.executor.pubsub;
import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
import io.netty.buffer.ByteBuf;
@@ -27,6 +29,7 @@ import org.apache.geode.redis.internal.executor.AbstractExecutor;
public class PublishExecutor extends AbstractExecutor {
+ private ExecutorService executorService = Executors.newSingleThreadExecutor();
@Override
public void executeCommand(Command command, ExecutionHandlerContext context) {
@@ -37,14 +40,28 @@ public class PublishExecutor extends AbstractExecutor {
}
String channelName = new String(args.get(1));
- long publishCount = context.getPubSub().publish(channelName, args.get(2));
- writeResponse(command, context, publishCount);
+ executorService.submit(new PublishingRunnable(context, channelName, args.get(2)));
}
- private void writeResponse(Command command, ExecutionHandlerContext context,
- long publishCount) {
- ByteBuf response = Coder.getIntegerResponse(context.getByteBufAllocator(), publishCount);
- command.setResponse(response);
+ public static class PublishingRunnable implements Runnable {
+
+ private final ExecutionHandlerContext context;
+ private final String channelName;
+ private final byte[] message;
+
+ public PublishingRunnable(ExecutionHandlerContext context, String channelName, byte[] message) {
+ this.context = context;
+ this.channelName = channelName;
+ this.message = message;
+ }
+
+ @Override
+ public void run() {
+ long publishCount = context.getPubSub().publish(channelName, message);
+ ByteBuf response = Coder.getIntegerResponse(context.getByteBufAllocator(), publishCount);
+ context.writeToChannel(response);
+ }
}
+
}
diff --git a/geode-redis/src/main/java/org/apache/geode/redis/internal/executor/pubsub/UnsubscribeExecutor.java b/geode-redis/src/main/java/org/apache/geode/redis/internal/executor/pubsub/UnsubscribeExecutor.java
index 55abad4..bbdfcd9 100755
--- a/geode-redis/src/main/java/org/apache/geode/redis/internal/executor/pubsub/UnsubscribeExecutor.java
+++ b/geode-redis/src/main/java/org/apache/geode/redis/internal/executor/pubsub/UnsubscribeExecutor.java
@@ -16,6 +16,7 @@
package org.apache.geode.redis.internal.executor.pubsub;
import java.util.ArrayList;
+import java.util.List;
import io.netty.buffer.ByteBuf;
import org.apache.logging.log4j.Logger;
@@ -32,7 +33,14 @@ public class UnsubscribeExecutor extends AbstractExecutor {
@Override
public void executeCommand(Command command, ExecutionHandlerContext context) {
- byte[] channelName = command.getProcessedCommand().get(1);
+ List<byte[]> commandElems = command.getProcessedCommand();
+ if (commandElems.size() < 2) {
+ command.setResponse(
+ Coder.getErrorResponse(context.getByteBufAllocator(), "not enough arguments"));
+ return;
+ }
+
+ byte[] channelName = commandElems.get(1);
long subscriptionCount =
context.getPubSub().unsubscribe(new String(channelName), context.getClient());
diff --git a/geode-redis/src/test/java/org/apache/geode/redis/internal/PubSubImplJUnitTest.java b/geode-redis/src/test/java/org/apache/geode/redis/internal/PubSubImplJUnitTest.java
index 659c670..424b5b6 100644
--- a/geode-redis/src/test/java/org/apache/geode/redis/internal/PubSubImplJUnitTest.java
+++ b/geode-redis/src/test/java/org/apache/geode/redis/internal/PubSubImplJUnitTest.java
@@ -17,11 +17,16 @@ package org.apache.geode.redis.internal;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.when;
+import io.netty.buffer.PooledByteBufAllocator;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelPromise;
+import io.netty.channel.DefaultChannelPromise;
+import io.netty.util.concurrent.Future;
+import io.netty.util.concurrent.GenericFutureListener;
import org.junit.Test;
@@ -30,16 +35,19 @@ public class PubSubImplJUnitTest {
@Test
public void testSubscriptionWithDeadClientIsPruned() {
Subscriptions subscriptions = new Subscriptions();
- ExecutionHandlerContext mockExecutionHandlerContext = mock(ExecutionHandlerContext.class);
+ ExecutionHandlerContext mockContext = mock(ExecutionHandlerContext.class);
+
+ FailingChannelFuture mockFuture = new FailingChannelFuture();
+ when(mockContext.writeToChannel(any())).thenReturn(mockFuture);
+ when(mockContext.getByteBufAllocator()).thenReturn(new PooledByteBufAllocator());
Client deadClient = mock(Client.class);
when(deadClient.isDead()).thenReturn(true);
ChannelSubscription subscription =
spy(new ChannelSubscription(deadClient,
- "sally", mockExecutionHandlerContext));
+ "sally", mockContext));
- doReturn(new PublishResult(deadClient, false)).when(subscription).publishMessage(any(), any());
subscriptions.add(subscription);
PubSubImpl subject = new PubSubImpl(subscriptions);
@@ -49,4 +57,34 @@ public class PubSubImplJUnitTest {
assertThat(numberOfSubscriptions).isEqualTo(0);
assertThat(subscriptions.findSubscriptions(deadClient)).isEmpty();
}
+
+ @SuppressWarnings("unchecked")
+ static class FailingChannelFuture extends DefaultChannelPromise {
+ private GenericFutureListener listener;
+
+ FailingChannelFuture() {
+ super(mock(Channel.class));
+ }
+
+ @Override
+ public ChannelPromise addListener(
+ GenericFutureListener<? extends Future<? super Void>> listener) {
+ this.listener = listener;
+ fail();
+ return null;
+ }
+
+ @Override
+ public Throwable cause() {
+ return new RuntimeException("aeotunhasoen");
+ }
+
+ public void fail() {
+ try {
+ this.listener.operationComplete(this);
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+ }
}