Apache9 commented on a change in pull request #1858: URL: https://github.com/apache/hbase/pull/1858#discussion_r436244428
########## File path: hbase-client/src/main/java/org/apache/hadoop/hbase/ipc/NettyRpcConnection.java ########## @@ -253,52 +268,38 @@ public void operationComplete(Future<Boolean> future) throws Exception { } private void connect() { + assert eventLoop.inEventLoop(); LOG.trace("Connecting to {}", remoteId.address); - this.channel = new Bootstrap().group(rpcClient.group).channel(rpcClient.channelClass) - .option(ChannelOption.TCP_NODELAY, rpcClient.isTcpNoDelay()) - .option(ChannelOption.SO_KEEPALIVE, rpcClient.tcpKeepAlive) - .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, rpcClient.connectTO) - .handler(new BufferCallBeforeInitHandler()).localAddress(rpcClient.localAddr) - .remoteAddress(remoteId.address).connect().addListener(new ChannelFutureListener() { - - @Override - public void operationComplete(ChannelFuture future) throws Exception { - Channel ch = future.channel(); - if (!future.isSuccess()) { - failInit(ch, toIOE(future.cause())); - rpcClient.failedServers.addToFailedServers(remoteId.address, future.cause()); - return; - } - ch.writeAndFlush(connectionHeaderPreamble.retainedDuplicate()); - if (useSasl) { - saslNegotiate(ch); - } else { - // send the connection header to server - ch.write(connectionHeaderWithLength.retainedDuplicate()); - established(ch); - } - } - }).channel(); - } + this.channel = new Bootstrap().group(eventLoop).channel(rpcClient.channelClass) + .option(ChannelOption.TCP_NODELAY, rpcClient.isTcpNoDelay()) + .option(ChannelOption.SO_KEEPALIVE, rpcClient.tcpKeepAlive) + .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, rpcClient.connectTO) + .handler(new BufferCallBeforeInitHandler()).localAddress(rpcClient.localAddr) + .remoteAddress(remoteId.address).connect().addListener(new ChannelFutureListener() { - private void write(Channel ch, final Call call) { - ch.writeAndFlush(call).addListener(new ChannelFutureListener() { - - @Override - public void operationComplete(ChannelFuture future) throws Exception { - // Fail the call if we failed to write it out. This usually because the channel is - // closed. This is needed because we may shutdown the channel inside event loop and - // there may still be some pending calls in the event loop queue after us. - if (!future.isSuccess()) { - call.setException(toIOE(future.cause())); + @Override + public void operationComplete(ChannelFuture future) throws Exception { + Channel ch = future.channel(); + if (!future.isSuccess()) { + failInit(ch, toIOE(future.cause())); + rpcClient.failedServers.addToFailedServers(remoteId.address, future.cause()); + return; + } + ch.writeAndFlush(connectionHeaderPreamble.retainedDuplicate()); + if (useSasl) { + saslNegotiate(ch); + } else { + // send the connection header to server + ch.write(connectionHeaderWithLength.retainedDuplicate()); + established(ch); + } } - } - }); + }).channel(); } - @Override - public synchronized void sendRequest(final Call call, HBaseRpcController hrc) throws IOException { + private void sendRequest0(Call call, HBaseRpcController hrc) throws IOException { + assert eventLoop.inEventLoop(); Review comment: I think you two are talking about different things? For the assert, just want to make sure that later developpers do not break the assumption as usually we will enable assert for UTs. For a production cluster where you want performance, you can just disable assert. And on the performance effect on event loop, I think it would be fine. As when you call write on a channel, finally they will switch to the event loop thread to call actual handlers. But anyway, we could run a PE to see the performance impact. ---------------------------------------------------------------- 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: us...@infra.apache.org