You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@geode.apache.org by GitBox <gi...@apache.org> on 2021/10/01 01:01:39 UTC
[GitHub] [geode] dschneider-pivotal opened a new pull request #6926: GEODE-9607 take2
dschneider-pivotal opened a new pull request #6926:
URL: https://github.com/apache/geode/pull/6926
still a WIP
<!-- Thank you for submitting a contribution to Apache Geode. -->
<!-- In order to streamline the review of the contribution we ask you
to ensure the following steps have been taken:
-->
### For all changes:
- [ ] Is there a JIRA ticket associated with this PR? Is it referenced in the commit message?
- [ ] Has your PR been rebased against the latest commit within the target branch (typically `develop`)?
- [ ] Is your initial contribution a single, squashed commit?
- [ ] Does `gradlew build` run cleanly?
- [ ] Have you written or updated unit tests to verify your changes?
- [ ] If adding new dependencies to the code, are these dependencies licensed in a way that is compatible for inclusion under [ASF 2.0](http://www.apache.org/legal/resolved.html#category-a)?
<!-- Note:
Please ensure that once the PR is submitted, check Concourse for build issues and
submit an update to your PR as soon as possible. If you need help, please send an
email to dev@geode.apache.org.
-->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] kirklund commented on a change in pull request #6926: GEODE-9607 take2
Posted by GitBox <gi...@apache.org>.
kirklund commented on a change in pull request #6926:
URL: https://github.com/apache/geode/pull/6926#discussion_r725207167
##########
File path: geode-for-redis/src/distributedTest/java/org/apache/geode/redis/OutOfMemoryDUnitTest.java
##########
@@ -103,7 +104,51 @@ public void shouldReturnOOMError_forWriteOperations_whenThresholdReached()
fillMemory(jedis, false);
assertThatThrownBy(
- () -> jedis.set("{key}oneMoreKey", makeLongStringValue(2 * LARGE_VALUE_SIZE)))
+ () -> jedis.set("{key}oneMoreKey", "value"))
+ .hasMessageContaining("OOM");
+
+ memoryPressureThread.interrupt();
+ memoryPressureThread.join();
Review comment:
Should this `join` use a timeout like `join(GeodeAwaitility.getTimeout().getSeconds(), TimeUnit.SECONDS)`?
##########
File path: geode-for-redis/src/distributedTest/java/org/apache/geode/redis/OutOfMemoryDUnitTest.java
##########
@@ -103,7 +104,51 @@ public void shouldReturnOOMError_forWriteOperations_whenThresholdReached()
fillMemory(jedis, false);
assertThatThrownBy(
- () -> jedis.set("{key}oneMoreKey", makeLongStringValue(2 * LARGE_VALUE_SIZE)))
+ () -> jedis.set("{key}oneMoreKey", "value"))
+ .hasMessageContaining("OOM");
+
+ memoryPressureThread.interrupt();
+ memoryPressureThread.join();
+ }
+
+ @Test
+ public void shouldReturnOOMError_forSubscribe_whenThresholdReached()
+ throws InterruptedException {
+ IgnoredException.addIgnoredException(expectedEx);
+ IgnoredException.addIgnoredException("LowMemoryException");
+ MockSubscriber mockSubscriber = new MockSubscriber();
+ int redisServerPort1 = clusterStartUp.getRedisPort(1);
+ JedisCluster subJedis =
+ new JedisCluster(new HostAndPort(BIND_ADDRESS, redisServerPort1), REDIS_CLIENT_TIMEOUT);
+
+ memoryPressureThread = new Thread(makeMemoryPressureRunnable());
Review comment:
`ExecutorServiceRule` will work well within the controller JVM. If you need a version of the rule that works within any JVM in a dunit test, then use `DistributedExecutorServiceRule`
##########
File path: geode-for-redis/src/distributedTest/java/org/apache/geode/redis/OutOfMemoryDUnitTest.java
##########
@@ -103,7 +104,51 @@ public void shouldReturnOOMError_forWriteOperations_whenThresholdReached()
fillMemory(jedis, false);
assertThatThrownBy(
- () -> jedis.set("{key}oneMoreKey", makeLongStringValue(2 * LARGE_VALUE_SIZE)))
+ () -> jedis.set("{key}oneMoreKey", "value"))
+ .hasMessageContaining("OOM");
+
+ memoryPressureThread.interrupt();
+ memoryPressureThread.join();
+ }
+
+ @Test
+ public void shouldReturnOOMError_forSubscribe_whenThresholdReached()
+ throws InterruptedException {
+ IgnoredException.addIgnoredException(expectedEx);
+ IgnoredException.addIgnoredException("LowMemoryException");
+ MockSubscriber mockSubscriber = new MockSubscriber();
+ int redisServerPort1 = clusterStartUp.getRedisPort(1);
+ JedisCluster subJedis =
+ new JedisCluster(new HostAndPort(BIND_ADDRESS, redisServerPort1), REDIS_CLIENT_TIMEOUT);
+
+ memoryPressureThread = new Thread(makeMemoryPressureRunnable());
Review comment:
You might want to consider using `ExecutorServiceRule` or `DistributedExecutorServiceRule` with a `CompletableFuture` instead of directly using a thread:
```
@Rule
public ExecutorServiceRule executorServiceRule = new ExecutorServiceRule();
```
```
CompletableFuture<Void> memoryPressure = executorServiceRule.runAsync(() -> {
try {
while (true) {
Thread.sleep(1000);
}
} catch (InterruptedException e) {
// done
}
});
// do something
memoryPressure.cancel(true);
```
The `cancel(true)` will interrupt the thread. If the test throws before that, `ExecutorServiceRule` will clean up any threads during tear down by invoking shutdownNow which will also interrupt any running threads. If the thread resists interrupt and hangs, the rule should print a stack trace for it as a test failure which would then make it easy to debug.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] jdeppe-pivotal commented on a change in pull request #6926: GEODE-9607 take2
Posted by GitBox <gi...@apache.org>.
jdeppe-pivotal commented on a change in pull request #6926:
URL: https://github.com/apache/geode/pull/6926#discussion_r725220888
##########
File path: geode-for-redis/src/main/java/org/apache/geode/redis/internal/pubsub/Publisher.java
##########
@@ -0,0 +1,480 @@
+/*
+ * 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.pubsub;
+
+import static org.apache.geode.internal.lang.utils.JavaWorkarounds.computeIfAbsent;
+import static org.apache.geode.logging.internal.executors.LoggingExecutors.newCachedThreadPool;
+import static org.apache.geode.redis.internal.netty.Coder.bytesToString;
+import static org.apache.geode.redis.internal.netty.StringBytesGlossary.bMESSAGE;
+import static org.apache.geode.redis.internal.netty.StringBytesGlossary.bPMESSAGE;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.util.ReferenceCountUtil;
+import org.apache.logging.log4j.Logger;
+
+import org.apache.geode.DataSerializer;
+import org.apache.geode.annotations.VisibleForTesting;
+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.distributed.DistributedMember;
+import org.apache.geode.internal.InternalDataSerializer;
+import org.apache.geode.internal.cache.execute.InternalFunction;
+import org.apache.geode.internal.serialization.DataSerializableFixedID;
+import org.apache.geode.internal.serialization.DeserializationContext;
+import org.apache.geode.internal.serialization.KnownVersion;
+import org.apache.geode.internal.serialization.SerializationContext;
+import org.apache.geode.logging.internal.log4j.api.LogService;
+import org.apache.geode.redis.internal.RegionProvider;
+import org.apache.geode.redis.internal.netty.Client;
+import org.apache.geode.redis.internal.netty.Coder;
+import org.apache.geode.redis.internal.pubsub.Subscriptions.PatternSubscriptions;
+
+/**
+ * This class deals with doing a redis pubsub publish operation.
+ * Since publish requests from the same client must have order
+ * preserved, and since publish requests myst be done in a thread
+ * other than the one the request arrived from, this class has an
+ * instance of ClientPublisher for any Client that does a publish.
+ * The ClientPublisher maintains an ordered queue of requests
+ * and is careful to process them in the order they arrived.
+ * It supports batching them which can be a significant win as
+ * soon as a second geode server is added to the cluster.
+ * WARNING: The queue on the ClientPublisher is unbounded
+ * which can cause the server to run out of memory if the
+ * client keeps sending publish requests faster than the server
+ * can process them. It would be nice for this queue to have
+ * a bound but that could cause a hang because of the way
+ * we use Netty.
+ */
+public class Publisher {
+ private static final Logger logger = LogService.getLogger();
+
+ private final ExecutorService executor;
+ private final RegionProvider regionProvider;
+ private final Subscriptions subscriptions;
+ private final Map<Client, ClientPublisher> clientPublishers = new ConcurrentHashMap<>();
+
+ public Publisher(RegionProvider regionProvider, Subscriptions subscriptions) {
+ this.executor = createExecutorService();
+ this.regionProvider = regionProvider;
+ this.subscriptions = subscriptions;
+ registerPublishFunction();
+ }
+
+ @VisibleForTesting
+ Publisher(RegionProvider regionProvider, Subscriptions subscriptions, ExecutorService executor) {
+ this.executor = executor;
+ this.regionProvider = regionProvider;
+ this.subscriptions = subscriptions;
+ // no need to register function in unit tests
+ }
+
+ public void publish(Client client, byte[] channel, byte[] message) {
+ ClientPublisher clientPublisher =
+ computeIfAbsent(clientPublishers, client, key -> new ClientPublisher());
+ clientPublisher.publish(channel, message);
+ }
+
+ public void disconnect(Client client) {
+ clientPublishers.remove(client);
+ }
+
+ public void close() {
+ executor.shutdown();
+ try {
+ if (!executor.awaitTermination(1, TimeUnit.MINUTES)) {
+ logger.warn("Timed out waiting for queued publish requests to be sent.");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ @VisibleForTesting
+ int getClientCount() {
+ return clientPublishers.size();
+ }
+
+
+ private static ExecutorService createExecutorService() {
+ // return newFixedThreadPool(2, "GeodeRedisServer-Publish-", true);
Review comment:
This can be removed
##########
File path: geode-for-redis/src/main/java/org/apache/geode/redis/internal/pubsub/Publisher.java
##########
@@ -0,0 +1,480 @@
+/*
+ * 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.pubsub;
+
+import static org.apache.geode.internal.lang.utils.JavaWorkarounds.computeIfAbsent;
+import static org.apache.geode.logging.internal.executors.LoggingExecutors.newCachedThreadPool;
+import static org.apache.geode.redis.internal.netty.Coder.bytesToString;
+import static org.apache.geode.redis.internal.netty.StringBytesGlossary.bMESSAGE;
+import static org.apache.geode.redis.internal.netty.StringBytesGlossary.bPMESSAGE;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.util.ReferenceCountUtil;
+import org.apache.logging.log4j.Logger;
+
+import org.apache.geode.DataSerializer;
+import org.apache.geode.annotations.VisibleForTesting;
+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.distributed.DistributedMember;
+import org.apache.geode.internal.InternalDataSerializer;
+import org.apache.geode.internal.cache.execute.InternalFunction;
+import org.apache.geode.internal.serialization.DataSerializableFixedID;
+import org.apache.geode.internal.serialization.DeserializationContext;
+import org.apache.geode.internal.serialization.KnownVersion;
+import org.apache.geode.internal.serialization.SerializationContext;
+import org.apache.geode.logging.internal.log4j.api.LogService;
+import org.apache.geode.redis.internal.RegionProvider;
+import org.apache.geode.redis.internal.netty.Client;
+import org.apache.geode.redis.internal.netty.Coder;
+import org.apache.geode.redis.internal.pubsub.Subscriptions.PatternSubscriptions;
+
+/**
+ * This class deals with doing a redis pubsub publish operation.
+ * Since publish requests from the same client must have order
+ * preserved, and since publish requests myst be done in a thread
+ * other than the one the request arrived from, this class has an
+ * instance of ClientPublisher for any Client that does a publish.
+ * The ClientPublisher maintains an ordered queue of requests
+ * and is careful to process them in the order they arrived.
+ * It supports batching them which can be a significant win as
+ * soon as a second geode server is added to the cluster.
+ * WARNING: The queue on the ClientPublisher is unbounded
+ * which can cause the server to run out of memory if the
+ * client keeps sending publish requests faster than the server
+ * can process them. It would be nice for this queue to have
+ * a bound but that could cause a hang because of the way
+ * we use Netty.
+ */
+public class Publisher {
+ private static final Logger logger = LogService.getLogger();
+
+ private final ExecutorService executor;
+ private final RegionProvider regionProvider;
+ private final Subscriptions subscriptions;
+ private final Map<Client, ClientPublisher> clientPublishers = new ConcurrentHashMap<>();
+
+ public Publisher(RegionProvider regionProvider, Subscriptions subscriptions) {
+ this.executor = createExecutorService();
+ this.regionProvider = regionProvider;
+ this.subscriptions = subscriptions;
+ registerPublishFunction();
+ }
+
+ @VisibleForTesting
+ Publisher(RegionProvider regionProvider, Subscriptions subscriptions, ExecutorService executor) {
+ this.executor = executor;
+ this.regionProvider = regionProvider;
+ this.subscriptions = subscriptions;
+ // no need to register function in unit tests
+ }
+
+ public void publish(Client client, byte[] channel, byte[] message) {
+ ClientPublisher clientPublisher =
+ computeIfAbsent(clientPublishers, client, key -> new ClientPublisher());
+ clientPublisher.publish(channel, message);
+ }
+
+ public void disconnect(Client client) {
+ clientPublishers.remove(client);
+ }
+
+ public void close() {
+ executor.shutdown();
+ try {
+ if (!executor.awaitTermination(1, TimeUnit.MINUTES)) {
+ logger.warn("Timed out waiting for queued publish requests to be sent.");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ @VisibleForTesting
+ int getClientCount() {
+ return clientPublishers.size();
+ }
+
+
+ private static ExecutorService createExecutorService() {
+ // return newFixedThreadPool(2, "GeodeRedisServer-Publish-", true);
+ return newCachedThreadPool("GeodeRedisServer-Publish-", true);
+ }
+
+ private void registerPublishFunction() {
+ FunctionService.registerFunction(new PublishFunction(this));
+ }
+
+ @SuppressWarnings("unchecked")
+ private void internalPublish(PublishRequestBatch batch) {
+ Set<DistributedMember> remoteMembers = regionProvider.getRemoteRegionMembers();
+ List<PublishRequest> optimizedBatch = batch.optimize();
+ try {
+ ResultCollector<?, ?> resultCollector = null;
+ try {
+ if (!remoteMembers.isEmpty()) {
+ // send function to remotes
+ resultCollector = FunctionService
+ .onMembers(remoteMembers)
+ .setArguments(optimizedBatch)
+ .execute(PublishFunction.ID);
+ }
+ } finally {
+ // execute it locally
+ publishBatchToLocalSubscribers(optimizedBatch);
+ if (resultCollector != null) {
+ // block until remote execute completes
+ resultCollector.getResult();
+ }
+ }
+ } catch (Exception e) {
+ // the onMembers contract is for execute to throw an exception
+ // if one of the members goes down.
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "Exception executing publish function on batch {}. If a server departed during the publish then an exception is expected.",
+ batch, e);
+ }
+ }
+ }
+
+ private void publishBatchToLocalSubscribers(List<PublishRequest> batch) {
+ batch.forEach(request -> {
+ publishRequestToLocalChannelSubscribers(request);
+ publishRequestToLocalPatternSubscribers(request);
+ });
+ }
+
+ private void publishRequestToLocalChannelSubscribers(PublishRequest request) {
+ Collection<Subscription> channelSubscriptions =
+ subscriptions.getChannelSubscriptions(request.getChannel());
+ if (channelSubscriptions.isEmpty()) {
+ return;
+ }
+ ByteBuf writeBuf = channelSubscriptions.iterator().next().getChannelWriteBuffer();
+ for (byte[] message : request.getMessages()) {
+ Coder.writeArrayResponse(writeBuf, bMESSAGE, request.getChannel(), message);
+ }
+ if (channelSubscriptions.size() == 1) {
+ Subscription singleSubscription = channelSubscriptions.iterator().next();
+ singleSubscription.waitUntilReadyToPublish();
+ singleSubscription.writeBufferToChannel(writeBuf);
Review comment:
Seems like `waitUntilReadyToPublish` is always being called before `writeBufferToChannel` - should these two methods be collapsed?
##########
File path: geode-for-redis/src/main/java/org/apache/geode/redis/internal/pubsub/Publisher.java
##########
@@ -0,0 +1,480 @@
+/*
+ * 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.pubsub;
+
+import static org.apache.geode.internal.lang.utils.JavaWorkarounds.computeIfAbsent;
+import static org.apache.geode.logging.internal.executors.LoggingExecutors.newCachedThreadPool;
+import static org.apache.geode.redis.internal.netty.Coder.bytesToString;
+import static org.apache.geode.redis.internal.netty.StringBytesGlossary.bMESSAGE;
+import static org.apache.geode.redis.internal.netty.StringBytesGlossary.bPMESSAGE;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.util.ReferenceCountUtil;
+import org.apache.logging.log4j.Logger;
+
+import org.apache.geode.DataSerializer;
+import org.apache.geode.annotations.VisibleForTesting;
+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.distributed.DistributedMember;
+import org.apache.geode.internal.InternalDataSerializer;
+import org.apache.geode.internal.cache.execute.InternalFunction;
+import org.apache.geode.internal.serialization.DataSerializableFixedID;
+import org.apache.geode.internal.serialization.DeserializationContext;
+import org.apache.geode.internal.serialization.KnownVersion;
+import org.apache.geode.internal.serialization.SerializationContext;
+import org.apache.geode.logging.internal.log4j.api.LogService;
+import org.apache.geode.redis.internal.RegionProvider;
+import org.apache.geode.redis.internal.netty.Client;
+import org.apache.geode.redis.internal.netty.Coder;
+import org.apache.geode.redis.internal.pubsub.Subscriptions.PatternSubscriptions;
+
+/**
+ * This class deals with doing a redis pubsub publish operation.
+ * Since publish requests from the same client must have order
+ * preserved, and since publish requests myst be done in a thread
Review comment:
Minor typo: myst -> must
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] DonalEvans commented on a change in pull request #6926: GEODE-9607 take2
Posted by GitBox <gi...@apache.org>.
DonalEvans commented on a change in pull request #6926:
URL: https://github.com/apache/geode/pull/6926#discussion_r725307656
##########
File path: geode-for-redis/src/main/java/org/apache/geode/redis/internal/netty/Coder.java
##########
@@ -141,6 +141,21 @@ public static ByteBuf getArrayResponse(ByteBuf buffer, Collection<?> items,
return buffer;
}
+ public static void writeArrayResponse(ByteBuf buffer, Object... items) {
+ buffer.markWriterIndex();
+ try {
+ buffer.writeByte(ARRAY_ID);
+ appendAsciiDigitsToByteBuf(items.length, buffer);
+ buffer.writeBytes(bCRLF);
+ for (Object next : items) {
+ writeCollectionOrString(buffer, next, true);
+ }
+ } catch (CoderException e) {
+ buffer.resetWriterIndex();
+ getErrorResponse(buffer, "Internal server error: " + e.getMessage());
+ }
Review comment:
This try/catch behaviour is only present in this method. For consistency, should it also be added to other Coder methods that write to buffers?
##########
File path: geode-for-redis/src/distributedTest/java/org/apache/geode/redis/internal/executor/pubsub/PubSubConcurrentDUnitTest.java
##########
@@ -0,0 +1,300 @@
+/*
+ * 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.executor.pubsub;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.Logger;
+import org.apache.logging.log4j.core.config.Configurator;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import redis.clients.jedis.Jedis;
+import redis.clients.jedis.JedisPubSub;
+
+import org.apache.geode.logging.internal.executors.LoggingExecutors;
+import org.apache.geode.logging.internal.log4j.api.FastLogger;
+import org.apache.geode.logging.internal.log4j.api.LogService;
+import org.apache.geode.test.awaitility.GeodeAwaitility;
+import org.apache.geode.test.dunit.Host;
+import org.apache.geode.test.dunit.SerializableRunnableIF;
+import org.apache.geode.test.dunit.VM;
+import org.apache.geode.test.dunit.rules.MemberVM;
+import org.apache.geode.test.dunit.rules.RedisClusterStartupRule;
+import org.apache.geode.test.dunit.rules.SerializableFunction;
+import org.apache.geode.test.junit.rules.ExecutorServiceRule;
+import org.apache.geode.test.junit.rules.GfshCommandRule;
+import org.apache.geode.test.junit.rules.ServerStarterRule;
+
+
+public class PubSubConcurrentDUnitTest {
+
+ public static final String CHANNEL_NAME = "salutations";
+ public static final int JEDIS_TIMEOUT = Math.toIntExact(GeodeAwaitility.getTimeout().toMillis());
+ public static final int PUBLISHER_COUNT = 5;
+ public static final int PUBLISH_ITERATIONS = 10000;
+ public static final int SUBSCRIBER_COUNT = 5;
+ public static final int EXPECTED_MESSAGES = PUBLISHER_COUNT * PUBLISH_ITERATIONS;
+
+
+ @ClassRule
+ public static RedisClusterStartupRule cluster = new RedisClusterStartupRule(4);
+
+ @ClassRule
+ public static GfshCommandRule gfsh = new GfshCommandRule();
+
+ @ClassRule
+ public static ExecutorServiceRule executor = new ExecutorServiceRule();
+
+ public static ExecutorService subscribeExecutor;
+
+ private static final List<Jedis> subscribers = new ArrayList<>(SUBSCRIBER_COUNT);
+ private static final List<LatencySubscriber> mockSubscribers = new ArrayList<>(SUBSCRIBER_COUNT);
+ private static final List<Future<Void>> subscriberFutures = new ArrayList<>(SUBSCRIBER_COUNT);
+
+ private static final String LOCAL_HOST = "127.0.0.1";
+
+ private static MemberVM locator;
+ private MemberVM server1;
+ private MemberVM server2;
+ private VM subscriberVM;
+
+ private int redisServerPort1;
+ private int redisServerPort2;
+
+ @BeforeClass
+ public static void beforeClass() {
+ locator = cluster.startLocatorVM(0);
+ }
+
+ @Before
+ public void before() throws Exception {
+ int locatorPort = locator.getPort();
+ SerializableFunction<ServerStarterRule> operator = x -> x
+ .withSystemProperty("io.netty.eventLoopThreads", "10")
+ .withConnectionToLocator(locatorPort);
+
+ server1 = cluster.startRedisVM(1, operator);
+ server2 = cluster.startRedisVM(2, operator);
+ subscriberVM = cluster.getVM(3);
+
+ for (VM v : Host.getHost(0).getAllVMs()) {
+ v.invoke(() -> {
+ Logger logger = LogService.getLogger("org.apache.geode.redis");
+ Configurator.setAllLevels(logger.getName(), Level.getLevel("INFO"));
+ FastLogger.setDelegating(true);
+ });
+ }
+
+ redisServerPort1 = cluster.getRedisPort(1);
+ redisServerPort2 = cluster.getRedisPort(2);
+
+ gfsh.connectAndVerify(locator);
+ }
+
+ @After
+ public void tearDown() {
+ server1.stop();
+ server2.stop();
+ subscriberVM.invoke(() -> {
+ subscribers.clear();
+ mockSubscribers.clear();
+ subscriberFutures.clear();
+ subscribeExecutor.shutdownNow();
+ });
+ }
+
+ @Test
+ public void testConcurrentPubSubWithChannels() {
+ concurrentPubSub(false);
+ }
+
+ @Test
+ public void testConcurrentPubSubWithPatterns() {
+ concurrentPubSub(true);
+ }
+
+ private void concurrentPubSub(boolean usePattern) {
+ int[] ports = new int[] {redisServerPort1, redisServerPort2};
+
+ subscriberVM.invoke(new CreateSubscribers(ports, usePattern));
+
+ List<Future<Void>> futures = new LinkedList<>();
+ for (int i = 0; i < PUBLISHER_COUNT; i++) {
+ Jedis publisher = new Jedis(LOCAL_HOST, ports[i % ports.length]);
+ int localI = i;
+ Callable<Void> callable = () -> {
+ long start = System.nanoTime();
+ for (int j = 0; j < PUBLISH_ITERATIONS; j++) {
+ publisher.publish(CHANNEL_NAME,
+ String.format("hello-%d-%d-%d", localI, j, System.nanoTime()));
+ }
+ long end = System.nanoTime();
+ System.out.println("Did " + PUBLISH_ITERATIONS + " publishes in "
+ + TimeUnit.SECONDS.convert(end - start, TimeUnit.NANOSECONDS) + " seconds to "
+ + SUBSCRIBER_COUNT + " subscribers. Nanos/op=" + (end - start) / PUBLISH_ITERATIONS);
+ publisher.close();
+ return null;
+ };
+
+ futures.add(executor.submit(callable));
+ }
+
+ for (Future<Void> future : futures) {
+ GeodeAwaitility.await().untilAsserted(future::get);
+ }
+
+ subscriberVM.invoke(new WaitForSubscribers(usePattern));
+ }
+
+ private static class LatencySubscriber extends JedisPubSub {
+ private long totalLatency = 0;
+ private int messageCount = 0;
+ private final CountDownLatch subscriptionLatch;
+ // key is the publisherId and value is the messageId
+ private final Map<Integer, Integer> receivedMessages = new HashMap<>();
+
+ public LatencySubscriber(CountDownLatch subscriptionLatch) {
+ this.subscriptionLatch = subscriptionLatch;
+ }
+
+ @Override
+ public void onMessage(String channel, String message) {
+ long end = System.nanoTime();
+ long start = Long.parseLong(message.substring(message.lastIndexOf('-') + 1));
+ totalLatency += end - start;
+ messageCount++;
+ // if (messageCount % 1000 == 0) {
+ // System.out.println(getLatencyReport());
+ // }
Review comment:
This should be removed.
##########
File path: geode-for-redis/src/main/java/org/apache/geode/redis/internal/pubsub/PatternSubscriptionManager.java
##########
@@ -54,4 +67,47 @@ public void remove(Client client) {
client.getPatternSubscriptions().forEach(
channel -> remove(channel, client));
}
+
+ public List<PatternSubscriptions> getPatternSubscriptions(byte[] channel) {
+ SubscriptionId channelId = new SubscriptionId(channel);
+ return computeIfAbsent(patternSubscriptionCache, channelId, this::createPatternSubscriptions);
+ }
+
+ private List<PatternSubscriptions> createPatternSubscriptions(SubscriptionId channelId) {
+ List<PatternSubscriptions> result = null;
+ for (Map.Entry<SubscriptionId, ClientSubscriptionManager> entry : clientManagers.entrySet()) {
+ if (matches(entry.getKey(), channelId.getSubscriptionIdBytes())) {
+ if (result == null) {
+ result = new ArrayList<>();
+ }
+ result.add(new PatternSubscriptions(entry.getKey().getSubscriptionIdBytes(),
+ entry.getValue().getSubscriptions()));
+ }
+ }
+ // since we are going to cache result make its size perfect
+ if (result == null) {
+ result = emptyList();
+ } else if (result.size() == 1) {
+ result = singletonList(result.get(0));
+ } else {
+ // noinspection ToArrayCallWithZeroLengthArrayArgument
+ result = asList(result.toArray(new PatternSubscriptions[result.size()]));
Review comment:
The explanation for this inspection (see below) states that since Java 6, using a pre-sized array in `toArray()` calls has the same or possibly worse performance than just using an empty array. Rather than suppressing the inspection, it should be okay to just use `new PatternSubscriptions[0]` here.
>In older Java versions using pre-sized array was recommended, as the reflection call which is necessary to create an array of proper size was quite slow. However since late updates of OpenJDK 6 this call was intrinsified, making the performance of the empty array version the same and sometimes even better, compared to the pre-sized version.
##########
File path: geode-for-redis/src/distributedTest/java/org/apache/geode/redis/internal/executor/pubsub/PubSubConcurrentDUnitTest.java
##########
@@ -0,0 +1,300 @@
+/*
+ * 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.executor.pubsub;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.Logger;
+import org.apache.logging.log4j.core.config.Configurator;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import redis.clients.jedis.Jedis;
+import redis.clients.jedis.JedisPubSub;
+
+import org.apache.geode.logging.internal.executors.LoggingExecutors;
+import org.apache.geode.logging.internal.log4j.api.FastLogger;
+import org.apache.geode.logging.internal.log4j.api.LogService;
+import org.apache.geode.test.awaitility.GeodeAwaitility;
+import org.apache.geode.test.dunit.Host;
+import org.apache.geode.test.dunit.SerializableRunnableIF;
+import org.apache.geode.test.dunit.VM;
+import org.apache.geode.test.dunit.rules.MemberVM;
+import org.apache.geode.test.dunit.rules.RedisClusterStartupRule;
+import org.apache.geode.test.dunit.rules.SerializableFunction;
+import org.apache.geode.test.junit.rules.ExecutorServiceRule;
+import org.apache.geode.test.junit.rules.GfshCommandRule;
+import org.apache.geode.test.junit.rules.ServerStarterRule;
+
+
+public class PubSubConcurrentDUnitTest {
+
+ public static final String CHANNEL_NAME = "salutations";
+ public static final int JEDIS_TIMEOUT = Math.toIntExact(GeodeAwaitility.getTimeout().toMillis());
+ public static final int PUBLISHER_COUNT = 5;
+ public static final int PUBLISH_ITERATIONS = 10000;
+ public static final int SUBSCRIBER_COUNT = 5;
+ public static final int EXPECTED_MESSAGES = PUBLISHER_COUNT * PUBLISH_ITERATIONS;
+
+
+ @ClassRule
+ public static RedisClusterStartupRule cluster = new RedisClusterStartupRule(4);
+
+ @ClassRule
+ public static GfshCommandRule gfsh = new GfshCommandRule();
+
+ @ClassRule
+ public static ExecutorServiceRule executor = new ExecutorServiceRule();
+
+ public static ExecutorService subscribeExecutor;
+
+ private static final List<Jedis> subscribers = new ArrayList<>(SUBSCRIBER_COUNT);
+ private static final List<LatencySubscriber> mockSubscribers = new ArrayList<>(SUBSCRIBER_COUNT);
+ private static final List<Future<Void>> subscriberFutures = new ArrayList<>(SUBSCRIBER_COUNT);
+
+ private static final String LOCAL_HOST = "127.0.0.1";
+
+ private static MemberVM locator;
+ private MemberVM server1;
+ private MemberVM server2;
+ private VM subscriberVM;
+
+ private int redisServerPort1;
+ private int redisServerPort2;
+
+ @BeforeClass
+ public static void beforeClass() {
+ locator = cluster.startLocatorVM(0);
+ }
+
+ @Before
+ public void before() throws Exception {
+ int locatorPort = locator.getPort();
+ SerializableFunction<ServerStarterRule> operator = x -> x
+ .withSystemProperty("io.netty.eventLoopThreads", "10")
+ .withConnectionToLocator(locatorPort);
+
+ server1 = cluster.startRedisVM(1, operator);
+ server2 = cluster.startRedisVM(2, operator);
+ subscriberVM = cluster.getVM(3);
+
+ for (VM v : Host.getHost(0).getAllVMs()) {
+ v.invoke(() -> {
+ Logger logger = LogService.getLogger("org.apache.geode.redis");
+ Configurator.setAllLevels(logger.getName(), Level.getLevel("INFO"));
+ FastLogger.setDelegating(true);
+ });
+ }
+
+ redisServerPort1 = cluster.getRedisPort(1);
+ redisServerPort2 = cluster.getRedisPort(2);
+
+ gfsh.connectAndVerify(locator);
+ }
+
+ @After
+ public void tearDown() {
+ server1.stop();
+ server2.stop();
+ subscriberVM.invoke(() -> {
+ subscribers.clear();
+ mockSubscribers.clear();
+ subscriberFutures.clear();
+ subscribeExecutor.shutdownNow();
+ });
+ }
+
+ @Test
+ public void testConcurrentPubSubWithChannels() {
+ concurrentPubSub(false);
+ }
+
+ @Test
+ public void testConcurrentPubSubWithPatterns() {
+ concurrentPubSub(true);
+ }
+
+ private void concurrentPubSub(boolean usePattern) {
+ int[] ports = new int[] {redisServerPort1, redisServerPort2};
+
+ subscriberVM.invoke(new CreateSubscribers(ports, usePattern));
+
+ List<Future<Void>> futures = new LinkedList<>();
+ for (int i = 0; i < PUBLISHER_COUNT; i++) {
+ Jedis publisher = new Jedis(LOCAL_HOST, ports[i % ports.length]);
+ int localI = i;
+ Callable<Void> callable = () -> {
+ long start = System.nanoTime();
+ for (int j = 0; j < PUBLISH_ITERATIONS; j++) {
+ publisher.publish(CHANNEL_NAME,
+ String.format("hello-%d-%d-%d", localI, j, System.nanoTime()));
+ }
+ long end = System.nanoTime();
+ System.out.println("Did " + PUBLISH_ITERATIONS + " publishes in "
+ + TimeUnit.SECONDS.convert(end - start, TimeUnit.NANOSECONDS) + " seconds to "
+ + SUBSCRIBER_COUNT + " subscribers. Nanos/op=" + (end - start) / PUBLISH_ITERATIONS);
+ publisher.close();
+ return null;
+ };
+
+ futures.add(executor.submit(callable));
+ }
+
+ for (Future<Void> future : futures) {
+ GeodeAwaitility.await().untilAsserted(future::get);
+ }
+
+ subscriberVM.invoke(new WaitForSubscribers(usePattern));
+ }
+
+ private static class LatencySubscriber extends JedisPubSub {
+ private long totalLatency = 0;
+ private int messageCount = 0;
+ private final CountDownLatch subscriptionLatch;
+ // key is the publisherId and value is the messageId
+ private final Map<Integer, Integer> receivedMessages = new HashMap<>();
+
+ public LatencySubscriber(CountDownLatch subscriptionLatch) {
+ this.subscriptionLatch = subscriptionLatch;
+ }
+
+ @Override
+ public void onMessage(String channel, String message) {
+ long end = System.nanoTime();
+ long start = Long.parseLong(message.substring(message.lastIndexOf('-') + 1));
+ totalLatency += end - start;
+ messageCount++;
+ // if (messageCount % 1000 == 0) {
+ // System.out.println(getLatencyReport());
+ // }
+ int publisherIdStart = message.indexOf('-') + 1;
+ int publisherIdEnd = message.indexOf('-', publisherIdStart);
+ int publisherId = Integer.parseInt(message.substring(publisherIdStart, publisherIdEnd));
+ int messageIdStart = publisherIdEnd + 1;
+ int messageIdEnd = message.indexOf('-', messageIdStart);
+ int messageId = Integer.parseInt(message.substring(messageIdStart, messageIdEnd));
+ if (receivedMessages.containsKey(publisherId)) {
+ assertThat(messageId).isEqualTo(receivedMessages.get(publisherId) + 1);
+ } else {
+ assertThat(messageId).isZero();
+ }
+ receivedMessages.put(publisherId, messageId);
+ }
+
+ public void onPMessage(String pattern, String channel, String message) {
+ onMessage(channel, message);
+ }
+
+ @Override
+ public void onSubscribe(String channel, int subscribedChannels) {
+ subscriptionLatch.countDown();
+ }
+
+ @Override
+ public void onPSubscribe(String pattern, int subscribedChannels) {
+ subscriptionLatch.countDown();
+ }
+
+ public int getMessageCount() {
+ return messageCount;
+ }
+
+ public String getLatencyReport() {
+ return "Received " + messageCount + " messages in " + totalLatency + " nanos. nanos/op="
+ + (totalLatency / messageCount);
+ }
+ }
+
+ private static class CreateSubscribers implements SerializableRunnableIF {
+ private final int[] ports;
+ private final boolean usePattern;
+
+ public CreateSubscribers(int[] ports, boolean usePattern) {
+ this.ports = ports;
+ this.usePattern = usePattern;
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public void run() throws Exception {
+ subscribeExecutor = LoggingExecutors.newCachedThreadPool("subscribers", true);
+ CountDownLatch latch = new CountDownLatch(SUBSCRIBER_COUNT);
+ for (int i = 0; i < SUBSCRIBER_COUNT; i++) {
+ Jedis subscriber = new Jedis(LOCAL_HOST, ports[i % ports.length], 120000);
+ subscribers.add(subscriber);
+ LatencySubscriber mockSubscriber = new LatencySubscriber(latch);
+ mockSubscribers.add(mockSubscriber);
+ if (usePattern) {
+ subscriberFutures.add((Future<Void>) subscribeExecutor
+ .submit(() -> subscriber.psubscribe(mockSubscriber, "*")));
+ } else {
+ subscriberFutures.add((Future<Void>) subscribeExecutor
+ .submit(() -> subscriber.subscribe(mockSubscriber, CHANNEL_NAME)));
+ }
+ }
+
+ assertThat(latch.await(30, TimeUnit.SECONDS)).as("channel subscription was not received")
+ .isTrue();
+ }
+ }
+ private static class WaitForSubscribers implements SerializableRunnableIF {
+ private final boolean usePattern;
+
+ public WaitForSubscribers(boolean usePattern) {
+ this.usePattern = usePattern;
+ }
+
+ @Override
+ public void run() throws Exception {
+ for (LatencySubscriber mockSubscriber : mockSubscribers) {
+ GeodeAwaitility.await()
+ .untilAsserted(() -> assertThat(mockSubscriber.getMessageCount())
+ .isEqualTo(EXPECTED_MESSAGES));
+ }
+
+ for (LatencySubscriber mockSubscriber : mockSubscribers) {
+ System.out.println(mockSubscriber.getLatencyReport());
Review comment:
This print statement, and the latency tracking as a whole, seems like it might be okay to remove from the test. If we want to test latency, it's probably best to do so in a dedicated test/benchmark, rather than as a side effect in a test like this, where we don't actually make any assertions on its value.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] dschneider-pivotal commented on a change in pull request #6926: GEODE-9607 take2
Posted by GitBox <gi...@apache.org>.
dschneider-pivotal commented on a change in pull request #6926:
URL: https://github.com/apache/geode/pull/6926#discussion_r725389495
##########
File path: geode-for-redis/src/distributedTest/java/org/apache/geode/redis/OutOfMemoryDUnitTest.java
##########
@@ -103,7 +104,51 @@ public void shouldReturnOOMError_forWriteOperations_whenThresholdReached()
fillMemory(jedis, false);
assertThatThrownBy(
- () -> jedis.set("{key}oneMoreKey", makeLongStringValue(2 * LARGE_VALUE_SIZE)))
+ () -> jedis.set("{key}oneMoreKey", "value"))
+ .hasMessageContaining("OOM");
+
+ memoryPressureThread.interrupt();
+ memoryPressureThread.join();
+ }
+
+ @Test
+ public void shouldReturnOOMError_forSubscribe_whenThresholdReached()
+ throws InterruptedException {
+ IgnoredException.addIgnoredException(expectedEx);
+ IgnoredException.addIgnoredException("LowMemoryException");
+ MockSubscriber mockSubscriber = new MockSubscriber();
+ int redisServerPort1 = clusterStartUp.getRedisPort(1);
+ JedisCluster subJedis =
+ new JedisCluster(new HostAndPort(BIND_ADDRESS, redisServerPort1), REDIS_CLIENT_TIMEOUT);
+
+ memoryPressureThread = new Thread(makeMemoryPressureRunnable());
Review comment:
Thanks for the feedback. Given that OutOfMemoryDUnitTest was already written to create its own threads and join on them and I just followed that existing pattern in adding publish and subscribe coverage; I'm going to file a new ticket with your suggested improvements to this test instead of doing it in this rather large PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] DonalEvans commented on a change in pull request #6926: GEODE-9607 take2
Posted by GitBox <gi...@apache.org>.
DonalEvans commented on a change in pull request #6926:
URL: https://github.com/apache/geode/pull/6926#discussion_r725373504
##########
File path: geode-for-redis/src/distributedTest/java/org/apache/geode/redis/internal/executor/pubsub/PubSubConcurrentDUnitTest.java
##########
@@ -0,0 +1,300 @@
+/*
+ * 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.executor.pubsub;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.Logger;
+import org.apache.logging.log4j.core.config.Configurator;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import redis.clients.jedis.Jedis;
+import redis.clients.jedis.JedisPubSub;
+
+import org.apache.geode.logging.internal.executors.LoggingExecutors;
+import org.apache.geode.logging.internal.log4j.api.FastLogger;
+import org.apache.geode.logging.internal.log4j.api.LogService;
+import org.apache.geode.test.awaitility.GeodeAwaitility;
+import org.apache.geode.test.dunit.Host;
+import org.apache.geode.test.dunit.SerializableRunnableIF;
+import org.apache.geode.test.dunit.VM;
+import org.apache.geode.test.dunit.rules.MemberVM;
+import org.apache.geode.test.dunit.rules.RedisClusterStartupRule;
+import org.apache.geode.test.dunit.rules.SerializableFunction;
+import org.apache.geode.test.junit.rules.ExecutorServiceRule;
+import org.apache.geode.test.junit.rules.GfshCommandRule;
+import org.apache.geode.test.junit.rules.ServerStarterRule;
+
+
+public class PubSubConcurrentDUnitTest {
+
+ public static final String CHANNEL_NAME = "salutations";
+ public static final int JEDIS_TIMEOUT = Math.toIntExact(GeodeAwaitility.getTimeout().toMillis());
+ public static final int PUBLISHER_COUNT = 5;
+ public static final int PUBLISH_ITERATIONS = 10000;
+ public static final int SUBSCRIBER_COUNT = 5;
+ public static final int EXPECTED_MESSAGES = PUBLISHER_COUNT * PUBLISH_ITERATIONS;
+
+
+ @ClassRule
+ public static RedisClusterStartupRule cluster = new RedisClusterStartupRule(4);
+
+ @ClassRule
+ public static GfshCommandRule gfsh = new GfshCommandRule();
+
+ @ClassRule
+ public static ExecutorServiceRule executor = new ExecutorServiceRule();
+
+ public static ExecutorService subscribeExecutor;
+
+ private static final List<Jedis> subscribers = new ArrayList<>(SUBSCRIBER_COUNT);
+ private static final List<LatencySubscriber> mockSubscribers = new ArrayList<>(SUBSCRIBER_COUNT);
+ private static final List<Future<Void>> subscriberFutures = new ArrayList<>(SUBSCRIBER_COUNT);
+
+ private static final String LOCAL_HOST = "127.0.0.1";
+
+ private static MemberVM locator;
+ private MemberVM server1;
+ private MemberVM server2;
+ private VM subscriberVM;
+
+ private int redisServerPort1;
+ private int redisServerPort2;
+
+ @BeforeClass
+ public static void beforeClass() {
+ locator = cluster.startLocatorVM(0);
+ }
+
+ @Before
+ public void before() throws Exception {
+ int locatorPort = locator.getPort();
+ SerializableFunction<ServerStarterRule> operator = x -> x
+ .withSystemProperty("io.netty.eventLoopThreads", "10")
+ .withConnectionToLocator(locatorPort);
+
+ server1 = cluster.startRedisVM(1, operator);
+ server2 = cluster.startRedisVM(2, operator);
+ subscriberVM = cluster.getVM(3);
+
+ for (VM v : Host.getHost(0).getAllVMs()) {
+ v.invoke(() -> {
+ Logger logger = LogService.getLogger("org.apache.geode.redis");
+ Configurator.setAllLevels(logger.getName(), Level.getLevel("INFO"));
+ FastLogger.setDelegating(true);
+ });
+ }
+
+ redisServerPort1 = cluster.getRedisPort(1);
+ redisServerPort2 = cluster.getRedisPort(2);
+
+ gfsh.connectAndVerify(locator);
+ }
+
+ @After
+ public void tearDown() {
+ server1.stop();
+ server2.stop();
+ subscriberVM.invoke(() -> {
+ subscribers.clear();
+ mockSubscribers.clear();
+ subscriberFutures.clear();
+ subscribeExecutor.shutdownNow();
+ });
+ }
+
+ @Test
+ public void testConcurrentPubSubWithChannels() {
+ concurrentPubSub(false);
+ }
+
+ @Test
+ public void testConcurrentPubSubWithPatterns() {
+ concurrentPubSub(true);
+ }
+
+ private void concurrentPubSub(boolean usePattern) {
+ int[] ports = new int[] {redisServerPort1, redisServerPort2};
+
+ subscriberVM.invoke(new CreateSubscribers(ports, usePattern));
+
+ List<Future<Void>> futures = new LinkedList<>();
+ for (int i = 0; i < PUBLISHER_COUNT; i++) {
+ Jedis publisher = new Jedis(LOCAL_HOST, ports[i % ports.length]);
+ int localI = i;
+ Callable<Void> callable = () -> {
+ long start = System.nanoTime();
+ for (int j = 0; j < PUBLISH_ITERATIONS; j++) {
+ publisher.publish(CHANNEL_NAME,
+ String.format("hello-%d-%d-%d", localI, j, System.nanoTime()));
+ }
+ long end = System.nanoTime();
+ System.out.println("Did " + PUBLISH_ITERATIONS + " publishes in "
+ + TimeUnit.SECONDS.convert(end - start, TimeUnit.NANOSECONDS) + " seconds to "
+ + SUBSCRIBER_COUNT + " subscribers. Nanos/op=" + (end - start) / PUBLISH_ITERATIONS);
+ publisher.close();
+ return null;
+ };
+
+ futures.add(executor.submit(callable));
+ }
+
+ for (Future<Void> future : futures) {
+ GeodeAwaitility.await().untilAsserted(future::get);
+ }
+
+ subscriberVM.invoke(new WaitForSubscribers(usePattern));
+ }
+
+ private static class LatencySubscriber extends JedisPubSub {
+ private long totalLatency = 0;
+ private int messageCount = 0;
+ private final CountDownLatch subscriptionLatch;
+ // key is the publisherId and value is the messageId
+ private final Map<Integer, Integer> receivedMessages = new HashMap<>();
+
+ public LatencySubscriber(CountDownLatch subscriptionLatch) {
+ this.subscriptionLatch = subscriptionLatch;
+ }
+
+ @Override
+ public void onMessage(String channel, String message) {
+ long end = System.nanoTime();
+ long start = Long.parseLong(message.substring(message.lastIndexOf('-') + 1));
+ totalLatency += end - start;
+ messageCount++;
+ // if (messageCount % 1000 == 0) {
+ // System.out.println(getLatencyReport());
+ // }
+ int publisherIdStart = message.indexOf('-') + 1;
+ int publisherIdEnd = message.indexOf('-', publisherIdStart);
+ int publisherId = Integer.parseInt(message.substring(publisherIdStart, publisherIdEnd));
+ int messageIdStart = publisherIdEnd + 1;
+ int messageIdEnd = message.indexOf('-', messageIdStart);
+ int messageId = Integer.parseInt(message.substring(messageIdStart, messageIdEnd));
+ if (receivedMessages.containsKey(publisherId)) {
+ assertThat(messageId).isEqualTo(receivedMessages.get(publisherId) + 1);
+ } else {
+ assertThat(messageId).isZero();
+ }
+ receivedMessages.put(publisherId, messageId);
+ }
+
+ public void onPMessage(String pattern, String channel, String message) {
+ onMessage(channel, message);
+ }
+
+ @Override
+ public void onSubscribe(String channel, int subscribedChannels) {
+ subscriptionLatch.countDown();
+ }
+
+ @Override
+ public void onPSubscribe(String pattern, int subscribedChannels) {
+ subscriptionLatch.countDown();
+ }
+
+ public int getMessageCount() {
+ return messageCount;
+ }
+
+ public String getLatencyReport() {
+ return "Received " + messageCount + " messages in " + totalLatency + " nanos. nanos/op="
+ + (totalLatency / messageCount);
+ }
+ }
+
+ private static class CreateSubscribers implements SerializableRunnableIF {
+ private final int[] ports;
+ private final boolean usePattern;
+
+ public CreateSubscribers(int[] ports, boolean usePattern) {
+ this.ports = ports;
+ this.usePattern = usePattern;
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public void run() throws Exception {
+ subscribeExecutor = LoggingExecutors.newCachedThreadPool("subscribers", true);
+ CountDownLatch latch = new CountDownLatch(SUBSCRIBER_COUNT);
+ for (int i = 0; i < SUBSCRIBER_COUNT; i++) {
+ Jedis subscriber = new Jedis(LOCAL_HOST, ports[i % ports.length], 120000);
+ subscribers.add(subscriber);
+ LatencySubscriber mockSubscriber = new LatencySubscriber(latch);
+ mockSubscribers.add(mockSubscriber);
+ if (usePattern) {
+ subscriberFutures.add((Future<Void>) subscribeExecutor
+ .submit(() -> subscriber.psubscribe(mockSubscriber, "*")));
+ } else {
+ subscriberFutures.add((Future<Void>) subscribeExecutor
+ .submit(() -> subscriber.subscribe(mockSubscriber, CHANNEL_NAME)));
+ }
+ }
+
+ assertThat(latch.await(30, TimeUnit.SECONDS)).as("channel subscription was not received")
+ .isTrue();
+ }
+ }
+ private static class WaitForSubscribers implements SerializableRunnableIF {
+ private final boolean usePattern;
+
+ public WaitForSubscribers(boolean usePattern) {
+ this.usePattern = usePattern;
+ }
+
+ @Override
+ public void run() throws Exception {
+ for (LatencySubscriber mockSubscriber : mockSubscribers) {
+ GeodeAwaitility.await()
+ .untilAsserted(() -> assertThat(mockSubscriber.getMessageCount())
+ .isEqualTo(EXPECTED_MESSAGES));
+ }
+
+ for (LatencySubscriber mockSubscriber : mockSubscribers) {
+ System.out.println(mockSubscriber.getLatencyReport());
Review comment:
Fair enough.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] dschneider-pivotal commented on a change in pull request #6926: GEODE-9607 take2
Posted by GitBox <gi...@apache.org>.
dschneider-pivotal commented on a change in pull request #6926:
URL: https://github.com/apache/geode/pull/6926#discussion_r725363571
##########
File path: geode-for-redis/src/main/java/org/apache/geode/redis/internal/netty/Coder.java
##########
@@ -141,6 +141,21 @@ public static ByteBuf getArrayResponse(ByteBuf buffer, Collection<?> items,
return buffer;
}
+ public static void writeArrayResponse(ByteBuf buffer, Object... items) {
+ buffer.markWriterIndex();
+ try {
+ buffer.writeByte(ARRAY_ID);
+ appendAsciiDigitsToByteBuf(items.length, buffer);
+ buffer.writeBytes(bCRLF);
+ for (Object next : items) {
+ writeCollectionOrString(buffer, next, true);
+ }
+ } catch (CoderException e) {
+ buffer.resetWriterIndex();
+ getErrorResponse(buffer, "Internal server error: " + e.getMessage());
+ }
Review comment:
Good point. The CoderException/getErrorResponse is actually something RedisResponse does. Publish is no longer using a RedisResponse but instead directly writing to the buffer. I will change this new Coder method to be like the others and put the exception handling in Publisher for now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] kirklund commented on a change in pull request #6926: GEODE-9607 take2
Posted by GitBox <gi...@apache.org>.
kirklund commented on a change in pull request #6926:
URL: https://github.com/apache/geode/pull/6926#discussion_r725210793
##########
File path: geode-for-redis/src/distributedTest/java/org/apache/geode/redis/OutOfMemoryDUnitTest.java
##########
@@ -103,7 +104,51 @@ public void shouldReturnOOMError_forWriteOperations_whenThresholdReached()
fillMemory(jedis, false);
assertThatThrownBy(
- () -> jedis.set("{key}oneMoreKey", makeLongStringValue(2 * LARGE_VALUE_SIZE)))
+ () -> jedis.set("{key}oneMoreKey", "value"))
+ .hasMessageContaining("OOM");
+
+ memoryPressureThread.interrupt();
+ memoryPressureThread.join();
+ }
+
+ @Test
+ public void shouldReturnOOMError_forSubscribe_whenThresholdReached()
+ throws InterruptedException {
+ IgnoredException.addIgnoredException(expectedEx);
+ IgnoredException.addIgnoredException("LowMemoryException");
+ MockSubscriber mockSubscriber = new MockSubscriber();
+ int redisServerPort1 = clusterStartUp.getRedisPort(1);
+ JedisCluster subJedis =
+ new JedisCluster(new HostAndPort(BIND_ADDRESS, redisServerPort1), REDIS_CLIENT_TIMEOUT);
+
+ memoryPressureThread = new Thread(makeMemoryPressureRunnable());
Review comment:
You might want to consider using `ExecutorServiceRule` or `DistributedExecutorServiceRule` with a `CompletableFuture` instead of directly using a thread:
```
@Rule
public ExecutorServiceRule executorServiceRule = new ExecutorServiceRule();
```
```
CompletableFuture<Void> memoryPressure = executorServiceRule.runAsync(() -> {
try {
while (true) {
Thread.sleep(1000);
}
} catch (InterruptedException e) {
// done
}
});
// do something
memoryPressure.cancel(true);
```
The `cancel(true)` will interrupt the thread. If the test throws before that, `ExecutorServiceRule` will clean up any threads during tear down by invoking `shutdownNow` which will also interrupt any running threads. If the thread resists interrupt and hangs, the rule should print a stack trace for it as a test failure which would then make it easy to debug.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] dschneider-pivotal commented on a change in pull request #6926: GEODE-9607 take2
Posted by GitBox <gi...@apache.org>.
dschneider-pivotal commented on a change in pull request #6926:
URL: https://github.com/apache/geode/pull/6926#discussion_r725391748
##########
File path: geode-for-redis/src/distributedTest/java/org/apache/geode/redis/OutOfMemoryDUnitTest.java
##########
@@ -103,7 +104,51 @@ public void shouldReturnOOMError_forWriteOperations_whenThresholdReached()
fillMemory(jedis, false);
assertThatThrownBy(
- () -> jedis.set("{key}oneMoreKey", makeLongStringValue(2 * LARGE_VALUE_SIZE)))
+ () -> jedis.set("{key}oneMoreKey", "value"))
+ .hasMessageContaining("OOM");
+
+ memoryPressureThread.interrupt();
+ memoryPressureThread.join();
Review comment:
added to: https://issues.apache.org/jira/browse/GEODE-9706
##########
File path: geode-for-redis/src/distributedTest/java/org/apache/geode/redis/OutOfMemoryDUnitTest.java
##########
@@ -103,7 +104,51 @@ public void shouldReturnOOMError_forWriteOperations_whenThresholdReached()
fillMemory(jedis, false);
assertThatThrownBy(
- () -> jedis.set("{key}oneMoreKey", makeLongStringValue(2 * LARGE_VALUE_SIZE)))
+ () -> jedis.set("{key}oneMoreKey", "value"))
+ .hasMessageContaining("OOM");
+
+ memoryPressureThread.interrupt();
+ memoryPressureThread.join();
+ }
+
+ @Test
+ public void shouldReturnOOMError_forSubscribe_whenThresholdReached()
+ throws InterruptedException {
+ IgnoredException.addIgnoredException(expectedEx);
+ IgnoredException.addIgnoredException("LowMemoryException");
+ MockSubscriber mockSubscriber = new MockSubscriber();
+ int redisServerPort1 = clusterStartUp.getRedisPort(1);
+ JedisCluster subJedis =
+ new JedisCluster(new HostAndPort(BIND_ADDRESS, redisServerPort1), REDIS_CLIENT_TIMEOUT);
+
+ memoryPressureThread = new Thread(makeMemoryPressureRunnable());
Review comment:
added to: https://issues.apache.org/jira/browse/GEODE-9706
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] dschneider-pivotal merged pull request #6926: GEODE-9607 take2
Posted by GitBox <gi...@apache.org>.
dschneider-pivotal merged pull request #6926:
URL: https://github.com/apache/geode/pull/6926
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] dschneider-pivotal commented on a change in pull request #6926: GEODE-9607 take2
Posted by GitBox <gi...@apache.org>.
dschneider-pivotal commented on a change in pull request #6926:
URL: https://github.com/apache/geode/pull/6926#discussion_r725352459
##########
File path: geode-for-redis/src/distributedTest/java/org/apache/geode/redis/internal/executor/pubsub/PubSubConcurrentDUnitTest.java
##########
@@ -0,0 +1,300 @@
+/*
+ * 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.executor.pubsub;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.Logger;
+import org.apache.logging.log4j.core.config.Configurator;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import redis.clients.jedis.Jedis;
+import redis.clients.jedis.JedisPubSub;
+
+import org.apache.geode.logging.internal.executors.LoggingExecutors;
+import org.apache.geode.logging.internal.log4j.api.FastLogger;
+import org.apache.geode.logging.internal.log4j.api.LogService;
+import org.apache.geode.test.awaitility.GeodeAwaitility;
+import org.apache.geode.test.dunit.Host;
+import org.apache.geode.test.dunit.SerializableRunnableIF;
+import org.apache.geode.test.dunit.VM;
+import org.apache.geode.test.dunit.rules.MemberVM;
+import org.apache.geode.test.dunit.rules.RedisClusterStartupRule;
+import org.apache.geode.test.dunit.rules.SerializableFunction;
+import org.apache.geode.test.junit.rules.ExecutorServiceRule;
+import org.apache.geode.test.junit.rules.GfshCommandRule;
+import org.apache.geode.test.junit.rules.ServerStarterRule;
+
+
+public class PubSubConcurrentDUnitTest {
+
+ public static final String CHANNEL_NAME = "salutations";
+ public static final int JEDIS_TIMEOUT = Math.toIntExact(GeodeAwaitility.getTimeout().toMillis());
+ public static final int PUBLISHER_COUNT = 5;
+ public static final int PUBLISH_ITERATIONS = 10000;
+ public static final int SUBSCRIBER_COUNT = 5;
+ public static final int EXPECTED_MESSAGES = PUBLISHER_COUNT * PUBLISH_ITERATIONS;
+
+
+ @ClassRule
+ public static RedisClusterStartupRule cluster = new RedisClusterStartupRule(4);
+
+ @ClassRule
+ public static GfshCommandRule gfsh = new GfshCommandRule();
+
+ @ClassRule
+ public static ExecutorServiceRule executor = new ExecutorServiceRule();
+
+ public static ExecutorService subscribeExecutor;
+
+ private static final List<Jedis> subscribers = new ArrayList<>(SUBSCRIBER_COUNT);
+ private static final List<LatencySubscriber> mockSubscribers = new ArrayList<>(SUBSCRIBER_COUNT);
+ private static final List<Future<Void>> subscriberFutures = new ArrayList<>(SUBSCRIBER_COUNT);
+
+ private static final String LOCAL_HOST = "127.0.0.1";
+
+ private static MemberVM locator;
+ private MemberVM server1;
+ private MemberVM server2;
+ private VM subscriberVM;
+
+ private int redisServerPort1;
+ private int redisServerPort2;
+
+ @BeforeClass
+ public static void beforeClass() {
+ locator = cluster.startLocatorVM(0);
+ }
+
+ @Before
+ public void before() throws Exception {
+ int locatorPort = locator.getPort();
+ SerializableFunction<ServerStarterRule> operator = x -> x
+ .withSystemProperty("io.netty.eventLoopThreads", "10")
+ .withConnectionToLocator(locatorPort);
+
+ server1 = cluster.startRedisVM(1, operator);
+ server2 = cluster.startRedisVM(2, operator);
+ subscriberVM = cluster.getVM(3);
+
+ for (VM v : Host.getHost(0).getAllVMs()) {
+ v.invoke(() -> {
+ Logger logger = LogService.getLogger("org.apache.geode.redis");
+ Configurator.setAllLevels(logger.getName(), Level.getLevel("INFO"));
+ FastLogger.setDelegating(true);
+ });
+ }
+
+ redisServerPort1 = cluster.getRedisPort(1);
+ redisServerPort2 = cluster.getRedisPort(2);
+
+ gfsh.connectAndVerify(locator);
+ }
+
+ @After
+ public void tearDown() {
+ server1.stop();
+ server2.stop();
+ subscriberVM.invoke(() -> {
+ subscribers.clear();
+ mockSubscribers.clear();
+ subscriberFutures.clear();
+ subscribeExecutor.shutdownNow();
+ });
+ }
+
+ @Test
+ public void testConcurrentPubSubWithChannels() {
+ concurrentPubSub(false);
+ }
+
+ @Test
+ public void testConcurrentPubSubWithPatterns() {
+ concurrentPubSub(true);
+ }
+
+ private void concurrentPubSub(boolean usePattern) {
+ int[] ports = new int[] {redisServerPort1, redisServerPort2};
+
+ subscriberVM.invoke(new CreateSubscribers(ports, usePattern));
+
+ List<Future<Void>> futures = new LinkedList<>();
+ for (int i = 0; i < PUBLISHER_COUNT; i++) {
+ Jedis publisher = new Jedis(LOCAL_HOST, ports[i % ports.length]);
+ int localI = i;
+ Callable<Void> callable = () -> {
+ long start = System.nanoTime();
+ for (int j = 0; j < PUBLISH_ITERATIONS; j++) {
+ publisher.publish(CHANNEL_NAME,
+ String.format("hello-%d-%d-%d", localI, j, System.nanoTime()));
+ }
+ long end = System.nanoTime();
+ System.out.println("Did " + PUBLISH_ITERATIONS + " publishes in "
+ + TimeUnit.SECONDS.convert(end - start, TimeUnit.NANOSECONDS) + " seconds to "
+ + SUBSCRIBER_COUNT + " subscribers. Nanos/op=" + (end - start) / PUBLISH_ITERATIONS);
+ publisher.close();
+ return null;
+ };
+
+ futures.add(executor.submit(callable));
+ }
+
+ for (Future<Void> future : futures) {
+ GeodeAwaitility.await().untilAsserted(future::get);
+ }
+
+ subscriberVM.invoke(new WaitForSubscribers(usePattern));
+ }
+
+ private static class LatencySubscriber extends JedisPubSub {
+ private long totalLatency = 0;
+ private int messageCount = 0;
+ private final CountDownLatch subscriptionLatch;
+ // key is the publisherId and value is the messageId
+ private final Map<Integer, Integer> receivedMessages = new HashMap<>();
+
+ public LatencySubscriber(CountDownLatch subscriptionLatch) {
+ this.subscriptionLatch = subscriptionLatch;
+ }
+
+ @Override
+ public void onMessage(String channel, String message) {
+ long end = System.nanoTime();
+ long start = Long.parseLong(message.substring(message.lastIndexOf('-') + 1));
+ totalLatency += end - start;
+ messageCount++;
+ // if (messageCount % 1000 == 0) {
+ // System.out.println(getLatencyReport());
+ // }
+ int publisherIdStart = message.indexOf('-') + 1;
+ int publisherIdEnd = message.indexOf('-', publisherIdStart);
+ int publisherId = Integer.parseInt(message.substring(publisherIdStart, publisherIdEnd));
+ int messageIdStart = publisherIdEnd + 1;
+ int messageIdEnd = message.indexOf('-', messageIdStart);
+ int messageId = Integer.parseInt(message.substring(messageIdStart, messageIdEnd));
+ if (receivedMessages.containsKey(publisherId)) {
+ assertThat(messageId).isEqualTo(receivedMessages.get(publisherId) + 1);
+ } else {
+ assertThat(messageId).isZero();
+ }
+ receivedMessages.put(publisherId, messageId);
+ }
+
+ public void onPMessage(String pattern, String channel, String message) {
+ onMessage(channel, message);
+ }
+
+ @Override
+ public void onSubscribe(String channel, int subscribedChannels) {
+ subscriptionLatch.countDown();
+ }
+
+ @Override
+ public void onPSubscribe(String pattern, int subscribedChannels) {
+ subscriptionLatch.countDown();
+ }
+
+ public int getMessageCount() {
+ return messageCount;
+ }
+
+ public String getLatencyReport() {
+ return "Received " + messageCount + " messages in " + totalLatency + " nanos. nanos/op="
+ + (totalLatency / messageCount);
+ }
+ }
+
+ private static class CreateSubscribers implements SerializableRunnableIF {
+ private final int[] ports;
+ private final boolean usePattern;
+
+ public CreateSubscribers(int[] ports, boolean usePattern) {
+ this.ports = ports;
+ this.usePattern = usePattern;
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public void run() throws Exception {
+ subscribeExecutor = LoggingExecutors.newCachedThreadPool("subscribers", true);
+ CountDownLatch latch = new CountDownLatch(SUBSCRIBER_COUNT);
+ for (int i = 0; i < SUBSCRIBER_COUNT; i++) {
+ Jedis subscriber = new Jedis(LOCAL_HOST, ports[i % ports.length], 120000);
+ subscribers.add(subscriber);
+ LatencySubscriber mockSubscriber = new LatencySubscriber(latch);
+ mockSubscribers.add(mockSubscriber);
+ if (usePattern) {
+ subscriberFutures.add((Future<Void>) subscribeExecutor
+ .submit(() -> subscriber.psubscribe(mockSubscriber, "*")));
+ } else {
+ subscriberFutures.add((Future<Void>) subscribeExecutor
+ .submit(() -> subscriber.subscribe(mockSubscriber, CHANNEL_NAME)));
+ }
+ }
+
+ assertThat(latch.await(30, TimeUnit.SECONDS)).as("channel subscription was not received")
+ .isTrue();
+ }
+ }
+ private static class WaitForSubscribers implements SerializableRunnableIF {
+ private final boolean usePattern;
+
+ public WaitForSubscribers(boolean usePattern) {
+ this.usePattern = usePattern;
+ }
+
+ @Override
+ public void run() throws Exception {
+ for (LatencySubscriber mockSubscriber : mockSubscribers) {
+ GeodeAwaitility.await()
+ .untilAsserted(() -> assertThat(mockSubscriber.getMessageCount())
+ .isEqualTo(EXPECTED_MESSAGES));
+ }
+
+ for (LatencySubscriber mockSubscriber : mockSubscribers) {
+ System.out.println(mockSubscriber.getLatencyReport());
Review comment:
It could be removed but I'd prefer not do that. It does not hurt this test which also does validation and I may find it helpful in the future to have this to run in my ide.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] dschneider-pivotal commented on a change in pull request #6926: GEODE-9607 take2
Posted by GitBox <gi...@apache.org>.
dschneider-pivotal commented on a change in pull request #6926:
URL: https://github.com/apache/geode/pull/6926#discussion_r725349261
##########
File path: geode-for-redis/src/main/java/org/apache/geode/redis/internal/pubsub/Publisher.java
##########
@@ -0,0 +1,480 @@
+/*
+ * 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.pubsub;
+
+import static org.apache.geode.internal.lang.utils.JavaWorkarounds.computeIfAbsent;
+import static org.apache.geode.logging.internal.executors.LoggingExecutors.newCachedThreadPool;
+import static org.apache.geode.redis.internal.netty.Coder.bytesToString;
+import static org.apache.geode.redis.internal.netty.StringBytesGlossary.bMESSAGE;
+import static org.apache.geode.redis.internal.netty.StringBytesGlossary.bPMESSAGE;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.util.ReferenceCountUtil;
+import org.apache.logging.log4j.Logger;
+
+import org.apache.geode.DataSerializer;
+import org.apache.geode.annotations.VisibleForTesting;
+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.distributed.DistributedMember;
+import org.apache.geode.internal.InternalDataSerializer;
+import org.apache.geode.internal.cache.execute.InternalFunction;
+import org.apache.geode.internal.serialization.DataSerializableFixedID;
+import org.apache.geode.internal.serialization.DeserializationContext;
+import org.apache.geode.internal.serialization.KnownVersion;
+import org.apache.geode.internal.serialization.SerializationContext;
+import org.apache.geode.logging.internal.log4j.api.LogService;
+import org.apache.geode.redis.internal.RegionProvider;
+import org.apache.geode.redis.internal.netty.Client;
+import org.apache.geode.redis.internal.netty.Coder;
+import org.apache.geode.redis.internal.pubsub.Subscriptions.PatternSubscriptions;
+
+/**
+ * This class deals with doing a redis pubsub publish operation.
+ * Since publish requests from the same client must have order
+ * preserved, and since publish requests myst be done in a thread
+ * other than the one the request arrived from, this class has an
+ * instance of ClientPublisher for any Client that does a publish.
+ * The ClientPublisher maintains an ordered queue of requests
+ * and is careful to process them in the order they arrived.
+ * It supports batching them which can be a significant win as
+ * soon as a second geode server is added to the cluster.
+ * WARNING: The queue on the ClientPublisher is unbounded
+ * which can cause the server to run out of memory if the
+ * client keeps sending publish requests faster than the server
+ * can process them. It would be nice for this queue to have
+ * a bound but that could cause a hang because of the way
+ * we use Netty.
+ */
+public class Publisher {
+ private static final Logger logger = LogService.getLogger();
+
+ private final ExecutorService executor;
+ private final RegionProvider regionProvider;
+ private final Subscriptions subscriptions;
+ private final Map<Client, ClientPublisher> clientPublishers = new ConcurrentHashMap<>();
+
+ public Publisher(RegionProvider regionProvider, Subscriptions subscriptions) {
+ this.executor = createExecutorService();
+ this.regionProvider = regionProvider;
+ this.subscriptions = subscriptions;
+ registerPublishFunction();
+ }
+
+ @VisibleForTesting
+ Publisher(RegionProvider regionProvider, Subscriptions subscriptions, ExecutorService executor) {
+ this.executor = executor;
+ this.regionProvider = regionProvider;
+ this.subscriptions = subscriptions;
+ // no need to register function in unit tests
+ }
+
+ public void publish(Client client, byte[] channel, byte[] message) {
+ ClientPublisher clientPublisher =
+ computeIfAbsent(clientPublishers, client, key -> new ClientPublisher());
+ clientPublisher.publish(channel, message);
+ }
+
+ public void disconnect(Client client) {
+ clientPublishers.remove(client);
+ }
+
+ public void close() {
+ executor.shutdown();
+ try {
+ if (!executor.awaitTermination(1, TimeUnit.MINUTES)) {
+ logger.warn("Timed out waiting for queued publish requests to be sent.");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ @VisibleForTesting
+ int getClientCount() {
+ return clientPublishers.size();
+ }
+
+
+ private static ExecutorService createExecutorService() {
+ // return newFixedThreadPool(2, "GeodeRedisServer-Publish-", true);
+ return newCachedThreadPool("GeodeRedisServer-Publish-", true);
+ }
+
+ private void registerPublishFunction() {
+ FunctionService.registerFunction(new PublishFunction(this));
+ }
+
+ @SuppressWarnings("unchecked")
+ private void internalPublish(PublishRequestBatch batch) {
+ Set<DistributedMember> remoteMembers = regionProvider.getRemoteRegionMembers();
+ List<PublishRequest> optimizedBatch = batch.optimize();
+ try {
+ ResultCollector<?, ?> resultCollector = null;
+ try {
+ if (!remoteMembers.isEmpty()) {
+ // send function to remotes
+ resultCollector = FunctionService
+ .onMembers(remoteMembers)
+ .setArguments(optimizedBatch)
+ .execute(PublishFunction.ID);
+ }
+ } finally {
+ // execute it locally
+ publishBatchToLocalSubscribers(optimizedBatch);
+ if (resultCollector != null) {
+ // block until remote execute completes
+ resultCollector.getResult();
+ }
+ }
+ } catch (Exception e) {
+ // the onMembers contract is for execute to throw an exception
+ // if one of the members goes down.
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "Exception executing publish function on batch {}. If a server departed during the publish then an exception is expected.",
+ batch, e);
+ }
+ }
+ }
+
+ private void publishBatchToLocalSubscribers(List<PublishRequest> batch) {
+ batch.forEach(request -> {
+ publishRequestToLocalChannelSubscribers(request);
+ publishRequestToLocalPatternSubscribers(request);
+ });
+ }
+
+ private void publishRequestToLocalChannelSubscribers(PublishRequest request) {
+ Collection<Subscription> channelSubscriptions =
+ subscriptions.getChannelSubscriptions(request.getChannel());
+ if (channelSubscriptions.isEmpty()) {
+ return;
+ }
+ ByteBuf writeBuf = channelSubscriptions.iterator().next().getChannelWriteBuffer();
+ for (byte[] message : request.getMessages()) {
+ Coder.writeArrayResponse(writeBuf, bMESSAGE, request.getChannel(), message);
+ }
+ if (channelSubscriptions.size() == 1) {
+ Subscription singleSubscription = channelSubscriptions.iterator().next();
+ singleSubscription.waitUntilReadyToPublish();
+ singleSubscription.writeBufferToChannel(writeBuf);
Review comment:
nice
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org