From dev-return-74838-archive-asf-public=cust-asf.ponee.io@zookeeper.apache.org Fri Oct 19 21:19:18 2018 Return-Path: X-Original-To: archive-asf-public@cust-asf.ponee.io Delivered-To: archive-asf-public@cust-asf.ponee.io Received: from mail.apache.org (hermes.apache.org [140.211.11.3]) by mx-eu-01.ponee.io (Postfix) with SMTP id 2BA83180652 for ; Fri, 19 Oct 2018 21:19:18 +0200 (CEST) Received: (qmail 9710 invoked by uid 500); 19 Oct 2018 19:19:17 -0000 Mailing-List: contact dev-help@zookeeper.apache.org; run by ezmlm Precedence: bulk List-Help: List-Unsubscribe: List-Post: List-Id: Reply-To: dev@zookeeper.apache.org Delivered-To: mailing list dev@zookeeper.apache.org Received: (qmail 9699 invoked by uid 99); 19 Oct 2018 19:19:16 -0000 Received: from git1-us-west.apache.org (HELO git1-us-west.apache.org) (140.211.11.23) by apache.org (qpsmtpd/0.29) with ESMTP; Fri, 19 Oct 2018 19:19:16 +0000 Received: by git1-us-west.apache.org (ASF Mail Server at git1-us-west.apache.org, from userid 33) id 7332FDFC65; Fri, 19 Oct 2018 19:19:16 +0000 (UTC) From: ivmaykov To: dev@zookeeper.apache.org Reply-To: dev@zookeeper.apache.org References: In-Reply-To: Subject: [GitHub] zookeeper pull request #669: ZOOKEEPER-3152: Port ZK netty stack to netty4 Content-Type: text/plain Message-Id: <20181019191916.7332FDFC65@git1-us-west.apache.org> Date: Fri, 19 Oct 2018 19:19:16 +0000 (UTC) Github user ivmaykov commented on a diff in the pull request: https://github.com/apache/zookeeper/pull/669#discussion_r226755419 --- Diff: zookeeper-common/src/main/java/org/apache/zookeeper/ClientCnxnSocketNetty.java --- @@ -103,71 +108,95 @@ boolean isConnected() { // Assuming that isConnected() is only used to initiate connection, // not used by some other connection status judgement. - return channel != null; + connectLock.lock(); + try { + return connectFuture != null || channel != null; + } finally { + connectLock.unlock(); + } } @Override void connect(InetSocketAddress addr) throws IOException { firstConnect = new CountDownLatch(1); - ClientBootstrap bootstrap = new ClientBootstrap(channelFactory); - - bootstrap.setPipelineFactory(new ZKClientPipelineFactory(addr.getHostString(), addr.getPort())); - bootstrap.setOption("soLinger", -1); - bootstrap.setOption("tcpNoDelay", true); - - connectFuture = bootstrap.connect(addr); - connectFuture.addListener(new ChannelFutureListener() { - @Override - public void operationComplete(ChannelFuture channelFuture) throws Exception { - // this lock guarantees that channel won't be assgined after cleanup(). - connectLock.lock(); - try { - if (!channelFuture.isSuccess() || connectFuture == null) { - LOG.info("future isn't success, cause: {}", channelFuture.getCause()); - return; - } - // setup channel, variables, connection, etc. - channel = channelFuture.getChannel(); - - disconnected.set(false); - initialized = false; - lenBuffer.clear(); - incomingBuffer = lenBuffer; - - sendThread.primeConnection(); - updateNow(); - updateLastSendAndHeard(); - - if (sendThread.tunnelAuthInProgress()) { - waitSasl.drainPermits(); - needSasl.set(true); - sendPrimePacket(); - } else { - needSasl.set(false); - } + Bootstrap bootstrap = new Bootstrap(); + bootstrap.group(Objects.requireNonNull(eventLoopGroup)) + .channel(NioSocketChannel.class) + .handler(new ZKClientPipelineFactory(addr.getHostString(), addr.getPort())) + .option(ChannelOption.SO_LINGER, -1) + .option(ChannelOption.TCP_NODELAY, true); + ByteBufAllocator testAllocator = TEST_ALLOCATOR.get(); + if (testAllocator != null) { + bootstrap.option(ChannelOption.ALLOCATOR, testAllocator); + } + bootstrap.validate(); + + connectLock.lock(); + try { + connectFuture = bootstrap.connect(addr); + connectFuture.addListener(new ChannelFutureListener() { + @Override + public void operationComplete(ChannelFuture channelFuture) throws Exception { + // this lock guarantees that channel won't be assigned after cleanup(). + connectLock.lock(); + try { + if (!channelFuture.isSuccess()) { + LOG.info("future isn't success, cause:", channelFuture.cause()); + return; + } else if (connectFuture == null) { + LOG.info("connect attempt cancelled"); + // If the connect attempt was cancelled but succeeded + // anyway, make sure to close the channel, otherwise + // we may leak a file descriptor. + channelFuture.channel().close(); --- End diff -- I don't think so, since this code can only trigger if the connect future is successful. If the future is not successful, the previous if branch will be taken. ---