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