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