[
https://issues.apache.org/jira/browse/GEODE-8333?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17164589#comment-17164589
]
ASF GitHub Bot commented on GEODE-8333:
---------------------------------------
jdeppe-pivotal commented on a change in pull request #5349:
URL: https://github.com/apache/geode/pull/5349#discussion_r460228306
##########
File path:
geode-redis/src/main/java/org/apache/geode/redis/internal/executor/pubsub/PublishExecutor.java
##########
@@ -27,12 +30,34 @@
@Override
public RedisResponse executeCommand(Command command,
ExecutionHandlerContext context) {
- List<byte[]> args = command.getProcessedCommand();
-
- long publishCount =
- context.getPubSub().publish(getDataRegion(context), args.get(1),
args.get(2));
-
- return RedisResponse.integer(publishCount);
+ context.getBackgroundExecutor().submit(new PublishingRunnable(context,
command));
+ return null;
}
+ public static class PublishingRunnable implements Runnable {
+
+ private final ExecutionHandlerContext context;
+ private final Command command;
+
+ public PublishingRunnable(ExecutionHandlerContext context, Command
command) {
+ this.context = context;
+ this.command = command;
+ }
+
+ @Override
+ public void run() {
+ List<byte[]> args = command.getProcessedCommand();
+ byte[] channelName = args.get(1);
+ byte[] message = args.get(2);
+ try {
+ long publishCount =
+ context.getPubSub()
+ .publish(context.getRegionProvider().getDataRegion(),
channelName, message);
+ ByteBuf response =
Coder.getIntegerResponse(context.getByteBufAllocator(), publishCount);
Review comment:
I'd prefer to create a `RedisResponse` so that we keep netty specifics
out of the executors.
##########
File path:
geode-redis/src/main/java/org/apache/geode/redis/internal/netty/ExecutionHandlerContext.java
##########
@@ -113,17 +114,19 @@ public ChannelFuture writeToChannel(RedisResponse
response) {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws
Exception {
Command command = (Command) msg;
- try {
- if (logger.isDebugEnabled()) {
- logger.debug("Executing Redis command: {}", command);
+ command.setChannelHandlerContext(ctx);
Review comment:
I think this might read a bit better like this:
```
synchronized (commandQueue) {
boolean isEmpty = commandQueue.isEmpty();
boolean isAsync = command.getCommandType().isAsync();
if (!isEmpty) {
commandQueue.offer(command);
return;
}
if (!isAsync) {
executeCommand(command);
} else {
commandQueue.offer(command);
startAsyncCommandExecution(command);
}
}
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
> Fix PUBSUB hang
> ---------------
>
> Key: GEODE-8333
> URL: https://issues.apache.org/jira/browse/GEODE-8333
> Project: Geode
> Issue Type: Bug
> Components: redis
> Reporter: Sarah Abbey
> Priority: Major
> Labels: pull-request-available
>
> PUBSUB hangs with concurrent publishers and subscribers on multiple servers
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
