View Javadoc
1   /*
2    * Copyright 2016 The Netty Project
3    *
4    * The Netty Project licenses this file to you under the Apache License,
5    * version 2.0 (the "License"); you may not use this file except in compliance
6    * with the License. You may obtain a copy of the License at:
7    *
8    *   http://www.apache.org/licenses/LICENSE-2.0
9    *
10   * Unless required by applicable law or agreed to in writing, software
11   * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12   * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13   * License for the specific language governing permissions and limitations
14   * under the License.
15   */
16  package io.netty.channel.kqueue;
17  
18  import io.netty.buffer.ByteBuf;
19  import io.netty.buffer.ByteBufAllocator;
20  import io.netty.buffer.ByteBufUtil;
21  import io.netty.buffer.Unpooled;
22  import io.netty.channel.AbstractChannel;
23  import io.netty.channel.Channel;
24  import io.netty.channel.ChannelConfig;
25  import io.netty.channel.ChannelException;
26  import io.netty.channel.ChannelFuture;
27  import io.netty.channel.ChannelFutureListener;
28  import io.netty.channel.ChannelMetadata;
29  import io.netty.channel.ChannelOutboundBuffer;
30  import io.netty.channel.ChannelPromise;
31  import io.netty.channel.ConnectTimeoutException;
32  import io.netty.channel.EventLoop;
33  import io.netty.channel.RecvByteBufAllocator;
34  import io.netty.channel.socket.ChannelInputShutdownEvent;
35  import io.netty.channel.socket.ChannelInputShutdownReadComplete;
36  import io.netty.channel.socket.SocketChannelConfig;
37  import io.netty.channel.unix.FileDescriptor;
38  import io.netty.channel.unix.UnixChannel;
39  import io.netty.util.ReferenceCountUtil;
40  
41  import java.io.IOException;
42  import java.net.ConnectException;
43  import java.net.InetSocketAddress;
44  import java.net.SocketAddress;
45  import java.nio.ByteBuffer;
46  import java.nio.channels.AlreadyConnectedException;
47  import java.nio.channels.ConnectionPendingException;
48  import java.nio.channels.NotYetConnectedException;
49  import java.nio.channels.UnresolvedAddressException;
50  import java.util.concurrent.ScheduledFuture;
51  import java.util.concurrent.TimeUnit;
52  
53  import static io.netty.channel.internal.ChannelUtils.WRITE_STATUS_SNDBUF_FULL;
54  import static io.netty.channel.unix.UnixChannelUtil.computeRemoteAddr;
55  import static io.netty.util.internal.ObjectUtil.checkNotNull;
56  
57  abstract class AbstractKQueueChannel extends AbstractChannel implements UnixChannel {
58      private static final ChannelMetadata METADATA = new ChannelMetadata(false);
59      /**
60       * The future of the current connection attempt.  If not null, subsequent
61       * connection attempts will fail.
62       */
63      private ChannelPromise connectPromise;
64      private ScheduledFuture<?> connectTimeoutFuture;
65      private SocketAddress requestedRemoteAddress;
66  
67      final BsdSocket socket;
68      private boolean readFilterEnabled;
69      private boolean writeFilterEnabled;
70      boolean readReadyRunnablePending;
71      boolean inputClosedSeenErrorOnRead;
72      /**
73       * This member variable means we don't have to have a map in {@link KQueueEventLoop} which associates the FDs
74       * from kqueue to instances of this class. This field will be initialized by JNI when modifying kqueue events.
75       * If there is no global reference when JNI gets a kqueue evSet call (aka this field is 0) then a global reference
76       * will be created and the address will be saved in this member variable. Then when we process a kevent in Java
77       * we can ask JNI to give us the {@link AbstractKQueueChannel} that corresponds to that event.
78       */
79      long jniSelfPtr;
80  
81      protected volatile boolean active;
82      private volatile SocketAddress local;
83      private volatile SocketAddress remote;
84  
85      AbstractKQueueChannel(Channel parent, BsdSocket fd, boolean active) {
86          super(parent);
87          socket = checkNotNull(fd, "fd");
88          this.active = active;
89          if (active) {
90              // Directly cache the remote and local addresses
91              // See https://github.com/netty/netty/issues/2359
92              local = fd.localAddress();
93              remote = fd.remoteAddress();
94          }
95      }
96  
97      AbstractKQueueChannel(Channel parent, BsdSocket fd, SocketAddress remote) {
98          super(parent);
99          socket = checkNotNull(fd, "fd");
100         active = true;
101         // Directly cache the remote and local addresses
102         // See https://github.com/netty/netty/issues/2359
103         this.remote = remote;
104         local = fd.localAddress();
105     }
106 
107     static boolean isSoErrorZero(BsdSocket fd) {
108         try {
109             return fd.getSoError() == 0;
110         } catch (IOException e) {
111             throw new ChannelException(e);
112         }
113     }
114 
115     @Override
116     public final FileDescriptor fd() {
117         return socket;
118     }
119 
120     @Override
121     public boolean isActive() {
122         return active;
123     }
124 
125     @Override
126     public ChannelMetadata metadata() {
127         return METADATA;
128     }
129 
130     @Override
131     protected void doClose() throws Exception {
132         active = false;
133         // Even if we allow half closed sockets we should give up on reading. Otherwise we may allow a read attempt on a
134         // socket which has not even been connected yet. This has been observed to block during unit tests.
135         inputClosedSeenErrorOnRead = true;
136         try {
137             if (isRegistered()) {
138                 // The FD will be closed, which should take care of deleting any associated events from kqueue, but
139                 // since we rely upon jniSelfRef to be consistent we make sure that we clear this reference out for
140                 // all events which are pending in kqueue to avoid referencing a deleted pointer at a later time.
141 
142                 // Need to check if we are on the EventLoop as doClose() may be triggered by the GlobalEventExecutor
143                 // if SO_LINGER is used.
144                 //
145                 // See https://github.com/netty/netty/issues/7159
146                 EventLoop loop = eventLoop();
147                 if (loop.inEventLoop()) {
148                     doDeregister();
149                 } else {
150                     loop.execute(new Runnable() {
151                         @Override
152                         public void run() {
153                             try {
154                                 doDeregister();
155                             } catch (Throwable cause) {
156                                 pipeline().fireExceptionCaught(cause);
157                             }
158                         }
159                     });
160                 }
161             }
162         } finally {
163             socket.close();
164         }
165     }
166 
167     @Override
168     protected void doDisconnect() throws Exception {
169         doClose();
170     }
171 
172     @Override
173     protected boolean isCompatible(EventLoop loop) {
174         return loop instanceof KQueueEventLoop;
175     }
176 
177     @Override
178     public boolean isOpen() {
179         return socket.isOpen();
180     }
181 
182     @Override
183     protected void doDeregister() throws Exception {
184         // Make sure we unregister our filters from kqueue!
185         readFilter(false);
186         writeFilter(false);
187         evSet0(Native.EVFILT_SOCK, Native.EV_DELETE, 0);
188 
189         ((KQueueEventLoop) eventLoop()).remove(this);
190     }
191 
192     @Override
193     protected final void doBeginRead() throws Exception {
194         // Channel.read() or ChannelHandlerContext.read() was called
195         final AbstractKQueueUnsafe unsafe = (AbstractKQueueUnsafe) unsafe();
196         unsafe.readPending = true;
197 
198         // We must set the read flag here as it is possible the user didn't read in the last read loop, the
199         // executeReadReadyRunnable could read nothing, and if the user doesn't explicitly call read they will
200         // never get data after this.
201         readFilter(true);
202 
203         // If auto read was toggled off on the last read loop then we may not be notified
204         // again if we didn't consume all the data. So we force a read operation here if there maybe more data.
205         if (unsafe.maybeMoreDataToRead) {
206             unsafe.executeReadReadyRunnable(config());
207         }
208     }
209 
210     @Override
211     protected void doRegister() throws Exception {
212         // Just in case the previous EventLoop was shutdown abruptly, or an event is still pending on the old EventLoop
213         // make sure the readReadyRunnablePending variable is reset so we will be able to execute the Runnable on the
214         // new EventLoop.
215         readReadyRunnablePending = false;
216         // Add the write event first so we get notified of connection refused on the client side!
217         if (writeFilterEnabled) {
218             evSet0(Native.EVFILT_WRITE, Native.EV_ADD_CLEAR_ENABLE);
219         }
220         if (readFilterEnabled) {
221             evSet0(Native.EVFILT_READ, Native.EV_ADD_CLEAR_ENABLE);
222         }
223         evSet0(Native.EVFILT_SOCK, Native.EV_ADD, Native.NOTE_RDHUP);
224     }
225 
226     @Override
227     protected abstract AbstractKQueueUnsafe newUnsafe();
228 
229     @Override
230     public abstract KQueueChannelConfig config();
231 
232     /**
233      * Returns an off-heap copy of the specified {@link ByteBuf}, and releases the original one.
234      */
235     protected final ByteBuf newDirectBuffer(ByteBuf buf) {
236         return newDirectBuffer(buf, buf);
237     }
238 
239     /**
240      * Returns an off-heap copy of the specified {@link ByteBuf}, and releases the specified holder.
241      * The caller must ensure that the holder releases the original {@link ByteBuf} when the holder is released by
242      * this method.
243      */
244     protected final ByteBuf newDirectBuffer(Object holder, ByteBuf buf) {
245         final int readableBytes = buf.readableBytes();
246         if (readableBytes == 0) {
247             ReferenceCountUtil.release(holder);
248             return Unpooled.EMPTY_BUFFER;
249         }
250 
251         final ByteBufAllocator alloc = alloc();
252         if (alloc.isDirectBufferPooled()) {
253             return newDirectBuffer0(holder, buf, alloc, readableBytes);
254         }
255 
256         final ByteBuf directBuf = ByteBufUtil.threadLocalDirectBuffer();
257         if (directBuf == null) {
258             return newDirectBuffer0(holder, buf, alloc, readableBytes);
259         }
260 
261         directBuf.writeBytes(buf, buf.readerIndex(), readableBytes);
262         ReferenceCountUtil.safeRelease(holder);
263         return directBuf;
264     }
265 
266     private static ByteBuf newDirectBuffer0(Object holder, ByteBuf buf, ByteBufAllocator alloc, int capacity) {
267         final ByteBuf directBuf = alloc.directBuffer(capacity);
268         directBuf.writeBytes(buf, buf.readerIndex(), capacity);
269         ReferenceCountUtil.safeRelease(holder);
270         return directBuf;
271     }
272 
273     protected static void checkResolvable(InetSocketAddress addr) {
274         if (addr.isUnresolved()) {
275             throw new UnresolvedAddressException();
276         }
277     }
278 
279     /**
280      * Read bytes into the given {@link ByteBuf} and return the amount.
281      */
282     protected final int doReadBytes(ByteBuf byteBuf) throws Exception {
283         int writerIndex = byteBuf.writerIndex();
284         int localReadAmount;
285         unsafe().recvBufAllocHandle().attemptedBytesRead(byteBuf.writableBytes());
286         if (byteBuf.hasMemoryAddress()) {
287             localReadAmount = socket.readAddress(byteBuf.memoryAddress(), writerIndex, byteBuf.capacity());
288         } else {
289             ByteBuffer buf = byteBuf.internalNioBuffer(writerIndex, byteBuf.writableBytes());
290             localReadAmount = socket.read(buf, buf.position(), buf.limit());
291         }
292         if (localReadAmount > 0) {
293             byteBuf.writerIndex(writerIndex + localReadAmount);
294         }
295         return localReadAmount;
296     }
297 
298     protected final int doWriteBytes(ChannelOutboundBuffer in, ByteBuf buf) throws Exception {
299         if (buf.hasMemoryAddress()) {
300             int localFlushedAmount = socket.writeAddress(buf.memoryAddress(), buf.readerIndex(), buf.writerIndex());
301             if (localFlushedAmount > 0) {
302                 in.removeBytes(localFlushedAmount);
303                 return 1;
304             }
305         } else {
306             final ByteBuffer nioBuf = buf.nioBufferCount() == 1 ?
307                     buf.internalNioBuffer(buf.readerIndex(), buf.readableBytes()) : buf.nioBuffer();
308             int localFlushedAmount = socket.write(nioBuf, nioBuf.position(), nioBuf.limit());
309             if (localFlushedAmount > 0) {
310                 nioBuf.position(nioBuf.position() + localFlushedAmount);
311                 in.removeBytes(localFlushedAmount);
312                 return 1;
313             }
314         }
315         return WRITE_STATUS_SNDBUF_FULL;
316     }
317 
318     final boolean shouldBreakReadReady(ChannelConfig config) {
319         return socket.isInputShutdown() && (inputClosedSeenErrorOnRead || !isAllowHalfClosure(config));
320     }
321 
322     private static boolean isAllowHalfClosure(ChannelConfig config) {
323         return config instanceof SocketChannelConfig &&
324                 ((SocketChannelConfig) config).isAllowHalfClosure();
325     }
326 
327     final void clearReadFilter() {
328         // Only clear if registered with an EventLoop as otherwise
329         if (isRegistered()) {
330             final EventLoop loop = eventLoop();
331             final AbstractKQueueUnsafe unsafe = (AbstractKQueueUnsafe) unsafe();
332             if (loop.inEventLoop()) {
333                 unsafe.clearReadFilter0();
334             } else {
335                 // schedule a task to clear the EPOLLIN as it is not safe to modify it directly
336                 loop.execute(new Runnable() {
337                     @Override
338                     public void run() {
339                         if (!unsafe.readPending && !config().isAutoRead()) {
340                             // Still no read triggered so clear it now
341                             unsafe.clearReadFilter0();
342                         }
343                     }
344                 });
345             }
346         } else  {
347             // The EventLoop is not registered atm so just update the flags so the correct value
348             // will be used once the channel is registered
349             readFilterEnabled = false;
350         }
351     }
352 
353     void readFilter(boolean readFilterEnabled) throws IOException {
354         if (this.readFilterEnabled != readFilterEnabled) {
355             this.readFilterEnabled = readFilterEnabled;
356             evSet(Native.EVFILT_READ, readFilterEnabled ? Native.EV_ADD_CLEAR_ENABLE : Native.EV_DELETE_DISABLE);
357         }
358     }
359 
360     void writeFilter(boolean writeFilterEnabled) throws IOException {
361         if (this.writeFilterEnabled != writeFilterEnabled) {
362             this.writeFilterEnabled = writeFilterEnabled;
363             evSet(Native.EVFILT_WRITE, writeFilterEnabled ? Native.EV_ADD_CLEAR_ENABLE : Native.EV_DELETE_DISABLE);
364         }
365     }
366 
367     private void evSet(short filter, short flags) {
368         if (isOpen() && isRegistered()) {
369             evSet0(filter, flags);
370         }
371     }
372 
373     private void evSet0(short filter, short flags) {
374         evSet0(filter, flags, 0);
375     }
376 
377     private void evSet0(short filter, short flags, int fflags) {
378         ((KQueueEventLoop) eventLoop()).evSet(this, filter, flags, fflags);
379     }
380 
381     abstract class AbstractKQueueUnsafe extends AbstractUnsafe {
382         boolean readPending;
383         boolean maybeMoreDataToRead;
384         private KQueueRecvByteAllocatorHandle allocHandle;
385         private final Runnable readReadyRunnable = new Runnable() {
386             @Override
387             public void run() {
388                 readReadyRunnablePending = false;
389                 readReady(recvBufAllocHandle());
390             }
391         };
392 
393         final void readReady(long numberBytesPending) {
394             KQueueRecvByteAllocatorHandle allocHandle = recvBufAllocHandle();
395             allocHandle.numberBytesPending(numberBytesPending);
396             readReady(allocHandle);
397         }
398 
399         abstract void readReady(KQueueRecvByteAllocatorHandle allocHandle);
400 
401         final void readReadyBefore() { maybeMoreDataToRead = false; }
402 
403         final void readReadyFinally(ChannelConfig config) {
404             maybeMoreDataToRead = allocHandle.maybeMoreDataToRead();
405 
406             if (allocHandle.isReadEOF() || (readPending && maybeMoreDataToRead)) {
407                 // trigger a read again as there may be something left to read and because of ET we
408                 // will not get notified again until we read everything from the socket
409                 //
410                 // It is possible the last fireChannelRead call could cause the user to call read() again, or if
411                 // autoRead is true the call to channelReadComplete would also call read, but maybeMoreDataToRead is set
412                 // to false before every read operation to prevent re-entry into readReady() we will not read from
413                 // the underlying OS again unless the user happens to call read again.
414                 executeReadReadyRunnable(config);
415             } else if (!readPending && !config.isAutoRead()) {
416                 // Check if there is a readPending which was not processed yet.
417                 // This could be for two reasons:
418                 // * The user called Channel.read() or ChannelHandlerContext.read() in channelRead(...) method
419                 // * The user called Channel.read() or ChannelHandlerContext.read() in channelReadComplete(...) method
420                 //
421                 // See https://github.com/netty/netty/issues/2254
422                 clearReadFilter0();
423             }
424         }
425 
426         final boolean failConnectPromise(Throwable cause) {
427             if (connectPromise != null) {
428                 // SO_ERROR has been shown to return 0 on macOS if detect an error via read() and the write filter was
429                 // not set before calling connect. This means finishConnect will not detect any error and would
430                 // successfully complete the connectPromise and update the channel state to active (which is incorrect).
431                 ChannelPromise connectPromise = AbstractKQueueChannel.this.connectPromise;
432                 AbstractKQueueChannel.this.connectPromise = null;
433                 if (connectPromise.tryFailure((cause instanceof ConnectException) ? cause
434                                 : new ConnectException("failed to connect").initCause(cause))) {
435                     closeIfClosed();
436                     return true;
437                 }
438             }
439             return false;
440         }
441 
442         final void writeReady() {
443             if (connectPromise != null) {
444                 // pending connect which is now complete so handle it.
445                 finishConnect();
446             } else if (!socket.isOutputShutdown()) {
447                 // directly call super.flush0() to force a flush now
448                 super.flush0();
449             }
450         }
451 
452         /**
453          * Shutdown the input side of the channel.
454          */
455         void shutdownInput(boolean readEOF) {
456             // We need to take special care of calling finishConnect() if readEOF is true and we not
457             // fullfilled the connectPromise yet. If we fail to do so the connectPromise will be failed
458             // with a ClosedChannelException as a close() will happen and so the FD is closed before we
459             // have a chance to call finishConnect() later on. Calling finishConnect() here will ensure
460             // we observe the correct exception in case of a connect failure.
461             if (readEOF && connectPromise != null) {
462                 finishConnect();
463             }
464             if (!socket.isInputShutdown()) {
465                 if (isAllowHalfClosure(config())) {
466                     try {
467                         socket.shutdown(true, false);
468                     } catch (IOException ignored) {
469                         // We attempted to shutdown and failed, which means the input has already effectively been
470                         // shutdown.
471                         fireEventAndClose(ChannelInputShutdownEvent.INSTANCE);
472                         return;
473                     } catch (NotYetConnectedException ignore) {
474                         // We attempted to shutdown and failed, which means the input has already effectively been
475                         // shutdown.
476                     }
477                     pipeline().fireUserEventTriggered(ChannelInputShutdownEvent.INSTANCE);
478                 } else {
479                     close(voidPromise());
480                 }
481             } else if (!readEOF) {
482                 inputClosedSeenErrorOnRead = true;
483                 pipeline().fireUserEventTriggered(ChannelInputShutdownReadComplete.INSTANCE);
484             }
485         }
486 
487         final void readEOF() {
488             // This must happen before we attempt to read. This will ensure reading continues until an error occurs.
489             final KQueueRecvByteAllocatorHandle allocHandle = recvBufAllocHandle();
490             allocHandle.readEOF();
491 
492             if (isActive()) {
493                 // If it is still active, we need to call readReady as otherwise we may miss to
494                 // read pending data from the underlying file descriptor.
495                 // See https://github.com/netty/netty/issues/3709
496                 readReady(allocHandle);
497             } else {
498                 // Just to be safe make sure the input marked as closed.
499                 shutdownInput(true);
500             }
501         }
502 
503         @Override
504         public KQueueRecvByteAllocatorHandle recvBufAllocHandle() {
505             if (allocHandle == null) {
506                 allocHandle = new KQueueRecvByteAllocatorHandle(
507                         (RecvByteBufAllocator.ExtendedHandle) super.recvBufAllocHandle());
508             }
509             return allocHandle;
510         }
511 
512         @Override
513         protected final void flush0() {
514             // Flush immediately only when there's no pending flush.
515             // If there's a pending flush operation, event loop will call forceFlush() later,
516             // and thus there's no need to call it now.
517             if (!writeFilterEnabled) {
518                 super.flush0();
519             }
520         }
521 
522         final void executeReadReadyRunnable(ChannelConfig config) {
523             if (readReadyRunnablePending || !isActive() || shouldBreakReadReady(config)) {
524                 return;
525             }
526             readReadyRunnablePending = true;
527             eventLoop().execute(readReadyRunnable);
528         }
529 
530         protected final void clearReadFilter0() {
531             assert eventLoop().inEventLoop();
532             try {
533                 readPending = false;
534                 readFilter(false);
535             } catch (IOException e) {
536                 // When this happens there is something completely wrong with either the filedescriptor or epoll,
537                 // so fire the exception through the pipeline and close the Channel.
538                 pipeline().fireExceptionCaught(e);
539                 unsafe().close(unsafe().voidPromise());
540             }
541         }
542 
543         private void fireEventAndClose(Object evt) {
544             pipeline().fireUserEventTriggered(evt);
545             close(voidPromise());
546         }
547 
548         @Override
549         public void connect(
550                 final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) {
551             if (!promise.setUncancellable() || !ensureOpen(promise)) {
552                 return;
553             }
554 
555             try {
556                 if (connectPromise != null) {
557                     throw new ConnectionPendingException();
558                 }
559 
560                 boolean wasActive = isActive();
561                 if (doConnect(remoteAddress, localAddress)) {
562                     fulfillConnectPromise(promise, wasActive);
563                 } else {
564                     connectPromise = promise;
565                     requestedRemoteAddress = remoteAddress;
566 
567                     // Schedule connect timeout.
568                     int connectTimeoutMillis = config().getConnectTimeoutMillis();
569                     if (connectTimeoutMillis > 0) {
570                         connectTimeoutFuture = eventLoop().schedule(new Runnable() {
571                             @Override
572                             public void run() {
573                                 ChannelPromise connectPromise = AbstractKQueueChannel.this.connectPromise;
574                                 ConnectTimeoutException cause =
575                                         new ConnectTimeoutException("connection timed out: " + remoteAddress);
576                                 if (connectPromise != null && connectPromise.tryFailure(cause)) {
577                                     close(voidPromise());
578                                 }
579                             }
580                         }, connectTimeoutMillis, TimeUnit.MILLISECONDS);
581                     }
582 
583                     promise.addListener(new ChannelFutureListener() {
584                         @Override
585                         public void operationComplete(ChannelFuture future) throws Exception {
586                             if (future.isCancelled()) {
587                                 if (connectTimeoutFuture != null) {
588                                     connectTimeoutFuture.cancel(false);
589                                 }
590                                 connectPromise = null;
591                                 close(voidPromise());
592                             }
593                         }
594                     });
595                 }
596             } catch (Throwable t) {
597                 closeIfClosed();
598                 promise.tryFailure(annotateConnectException(t, remoteAddress));
599             }
600         }
601 
602         private void fulfillConnectPromise(ChannelPromise promise, boolean wasActive) {
603             if (promise == null) {
604                 // Closed via cancellation and the promise has been notified already.
605                 return;
606             }
607             active = true;
608 
609             // Get the state as trySuccess() may trigger an ChannelFutureListener that will close the Channel.
610             // We still need to ensure we call fireChannelActive() in this case.
611             boolean active = isActive();
612 
613             // trySuccess() will return false if a user cancelled the connection attempt.
614             boolean promiseSet = promise.trySuccess();
615 
616             // Regardless if the connection attempt was cancelled, channelActive() event should be triggered,
617             // because what happened is what happened.
618             if (!wasActive && active) {
619                 pipeline().fireChannelActive();
620             }
621 
622             // If a user cancelled the connection attempt, close the channel, which is followed by channelInactive().
623             if (!promiseSet) {
624                 close(voidPromise());
625             }
626         }
627 
628         private void fulfillConnectPromise(ChannelPromise promise, Throwable cause) {
629             if (promise == null) {
630                 // Closed via cancellation and the promise has been notified already.
631                 return;
632             }
633 
634             // Use tryFailure() instead of setFailure() to avoid the race against cancel().
635             promise.tryFailure(cause);
636             closeIfClosed();
637         }
638 
639         private void finishConnect() {
640             // Note this method is invoked by the event loop only if the connection attempt was
641             // neither cancelled nor timed out.
642 
643             assert eventLoop().inEventLoop();
644 
645             boolean connectStillInProgress = false;
646             try {
647                 boolean wasActive = isActive();
648                 if (!doFinishConnect()) {
649                     connectStillInProgress = true;
650                     return;
651                 }
652                 fulfillConnectPromise(connectPromise, wasActive);
653             } catch (Throwable t) {
654                 fulfillConnectPromise(connectPromise, annotateConnectException(t, requestedRemoteAddress));
655             } finally {
656                 if (!connectStillInProgress) {
657                     // Check for null as the connectTimeoutFuture is only created if a connectTimeoutMillis > 0 is used
658                     // See https://github.com/netty/netty/issues/1770
659                     if (connectTimeoutFuture != null) {
660                         connectTimeoutFuture.cancel(false);
661                     }
662                     connectPromise = null;
663                 }
664             }
665         }
666 
667         private boolean doFinishConnect() throws Exception {
668             if (socket.finishConnect()) {
669                 writeFilter(false);
670                 if (requestedRemoteAddress instanceof InetSocketAddress) {
671                     remote = computeRemoteAddr((InetSocketAddress) requestedRemoteAddress, socket.remoteAddress());
672                 }
673                 requestedRemoteAddress = null;
674                 return true;
675             }
676             writeFilter(true);
677             return false;
678         }
679     }
680 
681     @Override
682     protected void doBind(SocketAddress local) throws Exception {
683         if (local instanceof InetSocketAddress) {
684             checkResolvable((InetSocketAddress) local);
685         }
686         socket.bind(local);
687         this.local = socket.localAddress();
688     }
689 
690     /**
691      * Connect to the remote peer
692      */
693     protected boolean doConnect(SocketAddress remoteAddress, SocketAddress localAddress) throws Exception {
694         if (localAddress instanceof InetSocketAddress) {
695             checkResolvable((InetSocketAddress) localAddress);
696         }
697 
698         InetSocketAddress remoteSocketAddr = remoteAddress instanceof InetSocketAddress
699                 ? (InetSocketAddress) remoteAddress : null;
700         if (remoteSocketAddr != null) {
701             checkResolvable(remoteSocketAddr);
702         }
703 
704         if (remote != null) {
705             // Check if already connected before trying to connect. This is needed as connect(...) will not return -1
706             // and set errno to EISCONN if a previous connect(...) attempt was setting errno to EINPROGRESS and finished
707             // later.
708             throw new AlreadyConnectedException();
709         }
710 
711         if (localAddress != null) {
712             socket.bind(localAddress);
713         }
714 
715         boolean connected = doConnect0(remoteAddress);
716         if (connected) {
717             remote = remoteSocketAddr == null ?
718                     remoteAddress : computeRemoteAddr(remoteSocketAddr, socket.remoteAddress());
719         }
720         // We always need to set the localAddress even if not connected yet as the bind already took place.
721         //
722         // See https://github.com/netty/netty/issues/3463
723         local = socket.localAddress();
724         return connected;
725     }
726 
727     private boolean doConnect0(SocketAddress remote) throws Exception {
728         boolean success = false;
729         try {
730             boolean connected = socket.connect(remote);
731             if (!connected) {
732                 writeFilter(true);
733             }
734             success = true;
735             return connected;
736         } finally {
737             if (!success) {
738                 doClose();
739             }
740         }
741     }
742 
743     @Override
744     protected SocketAddress localAddress0() {
745         return local;
746     }
747 
748     @Override
749     protected SocketAddress remoteAddress0() {
750         return remote;
751     }
752 }