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 2020/07/06 21:34:54 UTC
[GitHub] [geode] sabbeyPivotal opened a new pull request #5349: GEODE-8333: Fix PUBSUB hang
sabbeyPivotal opened a new pull request #5349:
URL: https://github.com/apache/geode/pull/5349
PUBSUB hangs with concurrent publishers and subscribers on multiple
servers. Changed Publish executor to execute publish on background
thread. Removed separate separate subscriber group. Added DUnit test to
recreate failure.
Co-authored-by: Murtuza Boxwala <mb...@pivotal.io>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [geode] dschneider-pivotal commented on pull request #5349: GEODE-8333: Fix PUBSUB hang
Posted by GitBox <gi...@apache.org>.
dschneider-pivotal commented on pull request #5349:
URL: https://github.com/apache/geode/pull/5349#issuecomment-663750841
the dunit failure is GEODE-7710 and unrelated to this pull request
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [geode] dschneider-pivotal commented on a change in pull request #5349: GEODE-8333: Fix PUBSUB hang
Posted by GitBox <gi...@apache.org>.
dschneider-pivotal commented on a change in pull request #5349:
URL: https://github.com/apache/geode/pull/5349#discussion_r460163178
##########
File path: geode-redis/src/main/java/org/apache/geode/redis/internal/netty/ExecutionHandlerContext.java
##########
@@ -191,38 +199,121 @@ public void channelInactive(ChannelHandlerContext ctx) {
ctx.close();
}
- private void executeCommand(ChannelHandlerContext ctx, Command command) {
- RedisResponse response;
-
- if (!isAuthenticated()) {
- response = handleUnAuthenticatedCommand(command);
- writeToChannel(response);
- return;
+ private void startAsyncCommandExecution(Command command) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Starting execution of async Redis command: {}", command);
}
+ final long start = redisStats.startCommand(command.getCommandType());
+ command.setAsyncStartTime(start);
+ command.execute(this);
+ }
- if (command.isUnsupported() && !allowUnsupportedCommands()) {
- writeToChannel(
- RedisResponse.error(command.getCommandType() + RedisConstants.ERROR_UNSUPPORTED_COMMAND));
- return;
+ public void endAsyncCommandExecution(Command command, ByteBuf response) {
+ synchronized (commandQueue) {
+ Command head = takeFromCommandQueue();
+ if (head != command) {
+ throw new IllegalStateException(
+ "expected " + command + " but found " + head + " in the queue");
Review comment:
No. This is basically a check to make sure the implementation is correct. It should never happen if it is.
So the only way I could think of getting it to happen under test would be to have a unit test that mocks takeFromCommandQueue to return a bogus command. But I don't see much value in adding that test.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [geode] dschneider-pivotal merged pull request #5349: GEODE-8333: Fix PUBSUB hang
Posted by GitBox <gi...@apache.org>.
dschneider-pivotal merged pull request #5349:
URL: https://github.com/apache/geode/pull/5349
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [geode] dschneider-pivotal commented on a change in pull request #5349: GEODE-8333: Fix PUBSUB hang
Posted by GitBox <gi...@apache.org>.
dschneider-pivotal commented on a change in pull request #5349:
URL: https://github.com/apache/geode/pull/5349#discussion_r460243756
##########
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:
Yes, and since isEmpty and isAsync are only used once I ended up changing it to this:
` synchronized (commandQueue) {
if (!commandQueue.isEmpty()) {
commandQueue.offer(command);
return;
}
if (command.getCommandType().isAsync()) {
commandQueue.offer(command);
startAsyncCommandExecution(command);
return;
}
executeCommand(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:
users@infra.apache.org
[GitHub] [geode] dschneider-pivotal commented on a change in pull request #5349: GEODE-8333: Fix PUBSUB hang
Posted by GitBox <gi...@apache.org>.
dschneider-pivotal commented on a change in pull request #5349:
URL: https://github.com/apache/geode/pull/5349#discussion_r460251331
##########
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:
Done. I also found one other place (in AbstractSubscription) that was not using RedisResponse. I change it also to do so and deleted the old writeToChannel(ByteBuf) which was no longer used after these changes.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [geode] dschneider-pivotal commented on a change in pull request #5349: GEODE-8333: Fix PUBSUB hang
Posted by GitBox <gi...@apache.org>.
dschneider-pivotal commented on a change in pull request #5349:
URL: https://github.com/apache/geode/pull/5349#discussion_r460243756
##########
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:
Yes, and since isEmpty and isAsync are only used once I ended up changing it to this:
```
synchronized (commandQueue) {
if (!commandQueue.isEmpty()) {
commandQueue.offer(command);
return;
}
if (command.getCommandType().isAsync()) {
commandQueue.offer(command);
startAsyncCommandExecution(command);
return;
}
executeCommand(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:
users@infra.apache.org
[GitHub] [geode] jdeppe-pivotal commented on a change in pull request #5349: GEODE-8333: Fix PUBSUB hang
Posted by GitBox <gi...@apache.org>.
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:
users@infra.apache.org
[GitHub] [geode] ringles commented on a change in pull request #5349: GEODE-8333: Fix PUBSUB hang
Posted by GitBox <gi...@apache.org>.
ringles commented on a change in pull request #5349:
URL: https://github.com/apache/geode/pull/5349#discussion_r460149939
##########
File path: geode-redis/src/main/java/org/apache/geode/redis/internal/netty/ExecutionHandlerContext.java
##########
@@ -191,38 +199,121 @@ public void channelInactive(ChannelHandlerContext ctx) {
ctx.close();
}
- private void executeCommand(ChannelHandlerContext ctx, Command command) {
- RedisResponse response;
-
- if (!isAuthenticated()) {
- response = handleUnAuthenticatedCommand(command);
- writeToChannel(response);
- return;
+ private void startAsyncCommandExecution(Command command) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Starting execution of async Redis command: {}", command);
}
+ final long start = redisStats.startCommand(command.getCommandType());
+ command.setAsyncStartTime(start);
+ command.execute(this);
+ }
- if (command.isUnsupported() && !allowUnsupportedCommands()) {
- writeToChannel(
- RedisResponse.error(command.getCommandType() + RedisConstants.ERROR_UNSUPPORTED_COMMAND));
- return;
+ public void endAsyncCommandExecution(Command command, ByteBuf response) {
+ synchronized (commandQueue) {
+ Command head = takeFromCommandQueue();
+ if (head != command) {
+ throw new IllegalStateException(
+ "expected " + command + " but found " + head + " in the queue");
Review comment:
Is there a test for this exception?
----------------------------------------------------------------
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:
users@infra.apache.org