Apache9 commented on a change in pull request #1858:
URL: https://github.com/apache/hbase/pull/1858#discussion_r436244190
##########
File path:
hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/NettyRpcConnection.java
##########
@@ -73,40 +73,53 @@
private static final Logger LOG =
LoggerFactory.getLogger(NettyRpcConnection.class);
private static final ScheduledExecutorService RELOGIN_EXECUTOR =
-
Executors.newSingleThreadScheduledExecutor(Threads.newDaemonThreadFactory("Relogin"));
+
Executors.newSingleThreadScheduledExecutor(Threads.newDaemonThreadFactory("Relogin"));
private final NettyRpcClient rpcClient;
+ // the event loop used to set up the connection, we will also execute other
operations for this
+ // connection in this event loop, to avoid locking everywhere.
+ private final EventLoop eventLoop;
+
private ByteBuf connectionHeaderPreamble;
private ByteBuf connectionHeaderWithLength;
- @edu.umd.cs.findbugs.annotations.SuppressWarnings(value =
"IS2_INCONSISTENT_SYNC",
- justification = "connect is also under lock as notifyOnCancel will call
our action directly")
- private Channel channel;
+ private volatile Channel channel;
NettyRpcConnection(NettyRpcClient rpcClient, ConnectionId remoteId) throws
IOException {
super(rpcClient.conf, AbstractRpcClient.WHEEL_TIMER, remoteId,
rpcClient.clusterId,
- rpcClient.userProvider.isHBaseSecurityEnabled(), rpcClient.codec,
rpcClient.compressor);
+ rpcClient.userProvider.isHBaseSecurityEnabled(), rpcClient.codec,
rpcClient.compressor);
this.rpcClient = rpcClient;
+ this.eventLoop = rpcClient.group.next();
byte[] connectionHeaderPreamble = getConnectionHeaderPreamble();
this.connectionHeaderPreamble =
-
Unpooled.directBuffer(connectionHeaderPreamble.length).writeBytes(connectionHeaderPreamble);
+
Unpooled.directBuffer(connectionHeaderPreamble.length).writeBytes(connectionHeaderPreamble);
ConnectionHeader header = getConnectionHeader();
this.connectionHeaderWithLength = Unpooled.directBuffer(4 +
header.getSerializedSize());
this.connectionHeaderWithLength.writeInt(header.getSerializedSize());
header.writeTo(new ByteBufOutputStream(this.connectionHeaderWithLength));
}
- @Override
- protected synchronized void callTimeout(Call call) {
- if (channel != null) {
- channel.pipeline().fireUserEventTriggered(new CallEvent(TIMEOUT, call));
+ private void execute(Runnable action) {
+ if (eventLoop.inEventLoop()) {
Review comment:
If your client is also in the same event loop thread then it will return
yes.
----------------------------------------------------------------
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]
