github-forks
/
redisson
mirror of https://github.com/redisson/redisson.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Improvement - response decoding optimization. #1908

Browse Source
pull/1907/head
Nikita Koksharov 6 years ago
parent 2b3db36d27
commit 2c98073532
1 changed files with 93 additions and 102 deletions
Show Stats Download Patch File Download Diff File

161
redisson/src/main/java/org/redisson/client/handler/CommandDecoder.java
Unescape Escape View File

@ -44,14 +44,11 @@ import org.redisson.client.RedisMovedException;
import org.redisson.client.RedisOutOfMemoryException; import org.redisson.client.RedisOutOfMemoryException;
import org.redisson.client.RedisTimeoutException; import org.redisson.client.RedisTimeoutException;
import org.redisson.client.RedisTryAgainException; import org.redisson.client.RedisTryAgainException;
import org.redisson.client.codec.BaseCodec;
import org.redisson.client.codec.ByteArrayCodec;
import org.redisson.client.codec.Codec; import org.redisson.client.codec.Codec;
import org.redisson.client.codec.StringCodec; import org.redisson.client.codec.StringCodec;
import org.redisson.client.protocol.CommandData; import org.redisson.client.protocol.CommandData;
import org.redisson.client.protocol.CommandsData; import org.redisson.client.protocol.CommandsData;
import org.redisson.client.protocol.Decoder; import org.redisson.client.protocol.Decoder;
import org.redisson.client.protocol.Encoder;
import org.redisson.client.protocol.QueueCommand; import org.redisson.client.protocol.QueueCommand;
import org.redisson.client.protocol.RedisCommand; import org.redisson.client.protocol.RedisCommand;
import org.redisson.client.protocol.RedisCommand.ValueType; import org.redisson.client.protocol.RedisCommand.ValueType;
@ -63,7 +60,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import io.netty.buffer.ByteBuf; import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.channel.Channel; import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ReplayingDecoder; import io.netty.handler.codec.ReplayingDecoder;
@ -78,30 +74,6 @@ import io.netty.util.concurrent.FastThreadLocal;
*/ */
public class CommandDecoder extends ReplayingDecoder<State> { public class CommandDecoder extends ReplayingDecoder<State> {
public static class NullCodec extends BaseCodec {
public static final NullCodec INSTANCE = new NullCodec();
private final Decoder<Object> decoder = new Decoder<Object>() {
@Override
public Object decode(ByteBuf buf, State state) {
return new Object();
}
};
@Override
public Decoder<Object> getValueDecoder() {
return decoder;
}
@Override
public Encoder getValueEncoder() {
throw new UnsupportedOperationException();
}
}
final Logger log = LoggerFactory.getLogger(getClass()); final Logger log = LoggerFactory.getLogger(getClass());
private static final char CR = '\r'; private static final char CR = '\r';
@ -113,13 +85,6 @@ public class CommandDecoder extends ReplayingDecoder<State> {
final ExecutorService executor; final ExecutorService executor;
private final boolean decodeInExecutor; private final boolean decodeInExecutor;
private final FastThreadLocal<Status> decoderStatus = new FastThreadLocal<Status>() {
@Override
protected Status initialValue() {
return Status.NORMAL;
};
};
private final FastThreadLocal<State> state = new FastThreadLocal<State>(); private final FastThreadLocal<State> state = new FastThreadLocal<State>();
public CommandDecoder(ExecutorService executor, boolean decodeInExecutor) { public CommandDecoder(ExecutorService executor, boolean decodeInExecutor) {
@ -128,50 +93,69 @@ public class CommandDecoder extends ReplayingDecoder<State> {
} }
@Override @Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { protected final void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
final QueueCommand data = ctx.channel().attr(CommandsQueue.CURRENT_COMMAND).get(); QueueCommand data = ctx.channel().attr(CommandsQueue.CURRENT_COMMAND).get();
if (log.isTraceEnabled()) { if (log.isTraceEnabled()) {
log.trace("reply: {}, channel: {}, command: {}", in.toString(0, in.writerIndex(), CharsetUtil.UTF_8), ctx.channel(), data); log.trace("reply: {}, channel: {}, command: {}", in.toString(0, in.writerIndex(), CharsetUtil.UTF_8), ctx.channel(), data);
} }
if (state.get() == null) { if (state.get() == null) {
state.set(new State()); state.set(new State());
} }
state.get().setDecoderState(null); state.get().setDecoderState(null);
if (data == null) {
while (in.writerIndex() > in.readerIndex()) {
in.markReaderIndex(); in.markReaderIndex();
decodeCommand(ctx.channel(), in, data); skipCommand(in);
in.resetReaderIndex();
if (decoderStatus.get() == Status.FILL_BUFFER) { decode(ctx, in, data);
}
} else {
in.markReaderIndex();
if (data instanceof CommandsData) {
CommandsData cmd = (CommandsData) data;
if (!cmd.isSkipResult()) {
for (int j = 0; j < cmd.getCommands().size(); j++) {
skipCommand(in);
}
}
} else {
skipCommand(in);
}
in.resetReaderIndex(); in.resetReaderIndex();
final ByteBuf copy = ByteBufAllocator.DEFAULT.buffer(in.writerIndex());
in.readBytes(copy);
state.set(null);
decoderStatus.set(Status.NORMAL);
final Channel channel = ctx.channel(); decode(ctx, in, data);
}
}
private void decode(ChannelHandlerContext ctx, ByteBuf in, QueueCommand data) throws Exception {
if (decodeInExecutor) {
ByteBuf copy = in.copy(in.readerIndex(), in.writerIndex() - in.readerIndex());
in.skipBytes(in.writerIndex() - in.readerIndex());
executor.execute(() -> { executor.execute(() -> {
decoderStatus.set(Status.DECODE_BUFFER);
state.set(new State()); state.set(new State());
state.get().setDecoderState(null); state.get().setDecoderState(null);
try { try {
decodeCommand(channel, copy, data); decodeCommand(ctx.channel(), copy, data);
} catch (Exception e) { } catch (Exception e) {
log.error("Unable to decode data in separate thread: " + LogHelper.toString(data), e); log.error("Unable to decode data in separate thread: " + LogHelper.toString(data), e);
} finally { } finally {
copy.release(); copy.release();
decoderStatus.remove();
state.remove();
} }
}); });
} else {
decodeCommand(ctx.channel(), in, data);
} }
} }
protected void sendNext(Channel channel, QueueCommand data) { protected void sendNext(Channel channel, QueueCommand data) {
if (data != null) { if (data != null) {
if (decoderStatus.get() == Status.FILL_BUFFER || data.isExecuted()) { if (data.isExecuted()) {
sendNext(channel); sendNext(channel);
} }
} else { } else {
@ -179,6 +163,46 @@ public class CommandDecoder extends ReplayingDecoder<State> {
} }
} }
protected void skipCommand(ByteBuf in) throws Exception {
skipDecode(in);
}
protected void skipDecode(ByteBuf in) throws IOException{
int code = in.readByte();
if (code == '+') {
skipString(in);
} else if (code == '-') {
skipString(in);
} else if (code == ':') {
skipString(in);
} else if (code == '$') {
skipBytes(in);
} else if (code == '*') {
long size = readLong(in);
for (int i = 0; i < size; i++) {
skipDecode(in);
}
}
}
private void skipBytes(ByteBuf is) throws IOException {
long l = readLong(is);
if (l > Integer.MAX_VALUE) {
throw new IllegalArgumentException(
"Java only supports arrays up to " + Integer.MAX_VALUE + " in size");
}
int size = (int) l;
if (size == -1) {
return;
}
is.skipBytes(size + 2);
}
private void skipString(ByteBuf in) {
int len = in.bytesBefore((byte) '\r');
in.skipBytes(len + 2);
}
protected void decodeCommand(Channel channel, ByteBuf in, QueueCommand data) throws Exception { protected void decodeCommand(Channel channel, ByteBuf in, QueueCommand data) throws Exception {
if (data instanceof CommandData) { if (data instanceof CommandData) {
CommandData<Object, Object> cmd = (CommandData<Object, Object>) data; CommandData<Object, Object> cmd = (CommandData<Object, Object>) data;
@ -188,7 +212,6 @@ public class CommandDecoder extends ReplayingDecoder<State> {
} catch (Exception e) { } catch (Exception e) {
log.error("Unable to decode data. channel: " + channel + ", reply: " + LogHelper.toString(in) + ", command: " + LogHelper.toString(data), e); log.error("Unable to decode data. channel: " + channel + ", reply: " + LogHelper.toString(in) + ", command: " + LogHelper.toString(data), e);
cmd.tryFailure(e); cmd.tryFailure(e);
decoderStatus.set(Status.NORMAL);
sendNext(channel); sendNext(channel);
throw e; throw e;
} }
@ -198,7 +221,6 @@ public class CommandDecoder extends ReplayingDecoder<State> {
decodeCommandBatch(channel, in, data, commands); decodeCommandBatch(channel, in, data, commands);
} catch (Exception e) { } catch (Exception e) {
commands.getPromise().tryFailure(e); commands.getPromise().tryFailure(e);
decoderStatus.set(Status.NORMAL);
sendNext(channel); sendNext(channel);
throw e; throw e;
} }
@ -210,7 +232,6 @@ public class CommandDecoder extends ReplayingDecoder<State> {
sendNext(channel); sendNext(channel);
} catch (Exception e) { } catch (Exception e) {
log.error("Unable to decode data. channel: " + channel + ", reply: " + LogHelper.toString(in), e); log.error("Unable to decode data. channel: " + channel + ", reply: " + LogHelper.toString(in), e);
decoderStatus.set(Status.NORMAL);
sendNext(channel); sendNext(channel);
throw e; throw e;
} }
@ -218,27 +239,19 @@ public class CommandDecoder extends ReplayingDecoder<State> {
} }
protected void sendNext(Channel channel) { protected void sendNext(Channel channel) {
if (decoderStatus.get() != Status.DECODE_BUFFER) {
channel.pipeline().get(CommandsQueue.class).sendNextCommand(channel); channel.pipeline().get(CommandsQueue.class).sendNextCommand(channel);
state.set(null); state.set(null);
} }
}
private void decodeCommandBatch(Channel channel, ByteBuf in, QueueCommand data, private void decodeCommandBatch(Channel channel, ByteBuf in, QueueCommand data,
CommandsData commandBatch) throws Exception { CommandsData commandBatch) throws Exception {
int i = 0; int i = state.get().getBatchIndex();
if (decoderStatus.get() != Status.DECODE_BUFFER) {
i = state.get().getBatchIndex();
}
Throwable error = null; Throwable error = null;
while (in.writerIndex() > in.readerIndex()) { while (in.writerIndex() > in.readerIndex()) {
CommandData<Object, Object> commandData = null; CommandData<Object, Object> commandData = null;
try { try {
if (decoderStatus.get() != Status.DECODE_BUFFER) {
checkpoint();
state.get().setBatchIndex(i); state.get().setBatchIndex(i);
}
RedisCommand<?> cmd = commandBatch.getCommands().get(i).getCommand(); RedisCommand<?> cmd = commandBatch.getCommands().get(i).getCommand();
boolean skipConvertor = commandBatch.isQueued(); boolean skipConvertor = commandBatch.isQueued();
List<CommandData<?, ?>> commandsData = null; List<CommandData<?, ?>> commandsData = null;
@ -291,7 +304,6 @@ public class CommandDecoder extends ReplayingDecoder<State> {
} }
if (commandBatch.isSkipResult() || i == commandBatch.getCommands().size()) { if (commandBatch.isSkipResult() || i == commandBatch.getCommands().size()) {
if (decoderStatus.get() != Status.FILL_BUFFER) {
RPromise<Void> promise = commandBatch.getPromise(); RPromise<Void> promise = commandBatch.getPromise();
if (error != null) { if (error != null) {
if (!promise.tryFailure(error) && promise.cause() instanceof RedisTimeoutException) { if (!promise.tryFailure(error) && promise.cause() instanceof RedisTimeoutException) {
@ -302,16 +314,12 @@ public class CommandDecoder extends ReplayingDecoder<State> {
log.warn("response has been skipped due to timeout! channel: {}, command: {}", channel, LogHelper.toString(data)); log.warn("response has been skipped due to timeout! channel: {}, command: {}", channel, LogHelper.toString(data));
} }
} }
}
sendNext(channel); sendNext(channel);
} else { } else {
if (decoderStatus.get() != Status.DECODE_BUFFER) {
checkpoint();
state.get().setBatchIndex(i); state.get().setBatchIndex(i);
} }
} }
}
protected void decode(ByteBuf in, CommandData<Object, Object> data, List<Object> parts, Channel channel, boolean skipConvertor, List<CommandData<?, ?>> commandsData) throws IOException { protected void decode(ByteBuf in, CommandData<Object, Object> data, List<Object> parts, Channel channel, boolean skipConvertor, List<CommandData<?, ?>> commandsData) throws IOException {
int code = in.readByte(); int code = in.readByte();
@ -412,10 +420,6 @@ public class CommandDecoder extends ReplayingDecoder<State> {
return; return;
} }
if (decoderStatus.get() == Status.FILL_BUFFER) {
return;
}
Object result = decoder.decode(respParts, state.get()); Object result = decoder.decode(respParts, state.get());
decodeResult(data, parts, channel, result); decodeResult(data, parts, channel, result);
} }
@ -428,7 +432,7 @@ public class CommandDecoder extends ReplayingDecoder<State> {
} }
private void handleResult(CommandData<Object, Object> data, List<Object> parts, Object result, boolean skipConvertor, Channel channel) { private void handleResult(CommandData<Object, Object> data, List<Object> parts, Object result, boolean skipConvertor, Channel channel) {
if (data != null && !skipConvertor && decoderStatus.get() != Status.FILL_BUFFER) { if (data != null && !skipConvertor) {
result = data.getCommand().getConvertor().convert(result); result = data.getCommand().getConvertor().convert(result);
} }
if (parts != null) { if (parts != null) {
@ -439,10 +443,6 @@ public class CommandDecoder extends ReplayingDecoder<State> {
} }
protected void completeResponse(CommandData<Object, Object> data, Object result, Channel channel) { protected void completeResponse(CommandData<Object, Object> data, Object result, Channel channel) {
if (decoderStatus.get() == Status.FILL_BUFFER) {
return;
}
if (data != null && !data.getPromise().trySuccess(result) && data.cause() instanceof RedisTimeoutException) { if (data != null && !data.getPromise().trySuccess(result) && data.cause() instanceof RedisTimeoutException) {
log.warn("response has been skipped due to timeout! channel: {}, command: {}, result: {}", channel, LogHelper.toString(data), LogHelper.toString(result)); log.warn("response has been skipped due to timeout! channel: {}, command: {}, result: {}", channel, LogHelper.toString(data), LogHelper.toString(result));
} }
@ -473,15 +473,6 @@ public class CommandDecoder extends ReplayingDecoder<State> {
} }
Codec codec = data.getCodec(); Codec codec = data.getCodec();
if (decodeInExecutor && !(codec instanceof StringCodec || codec instanceof ByteArrayCodec)) {
if (decoderStatus.get() == Status.NORMAL) {
decoderStatus.set(Status.FILL_BUFFER);
codec = NullCodec.INSTANCE;
} else if (decoderStatus.get() == Status.FILL_BUFFER) {
codec = NullCodec.INSTANCE;
}
}
Decoder<Object> decoder = data.getCommand().getReplayDecoder(); Decoder<Object> decoder = data.getCommand().getReplayDecoder();
if (decoder == null) { if (decoder == null) {
if (codec == null) { if (codec == null) {

Write Preview
Loading…
Cancel
Save