zookeeper-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From eolivelli <...@git.apache.org>
Subject [GitHub] zookeeper pull request #669: ZOOKEEPER-3152: Port ZK netty stack to netty4
Date Fri, 19 Oct 2018 15:38:52 GMT
Github user eolivelli commented on a diff in the pull request:

    https://github.com/apache/zookeeper/pull/669#discussion_r226682998
  
    --- 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 --
    
    Can this turn into an NPE? As channel() may return null. 


---

Mime
View raw message