Repository: netty/netty
Issue: SocketChannel.shutdownInput()?
Labels: feature

Is there a way to shutdownInput? I see that there is SocketChannel.shutdownOutput(), but no shutdownInput().


Issue comments:
At the moment not through this interface. What type of socket channel are you dealing with as there is likely a way to do it using an implementation of this interface.

NioSocketChannel, EpollSocketChannel or OioSocketChannel

I see you want it all ... stand by for a PR.

Fixed by https://github.com/netty/netty/pull/5269


Pull request review comments:
@normanmaurer - making `isInputShutdown()` abstract and removing this method is an API change ... i can be more strict with the API but this will change will only be for 4.1, and its OIO... let me know if you think I should spend more effort here to preserve the API

this is also an API change, but IMO was "broken" before (see the commit message)

@trustin WDYT ?

@trustin WDYT ?

@Scottmitch does this somehow change the behaviour of our current impl and so may break stuff ?

javadocs for protected methods please

same question as before 

setSuccess() ?

should we log t if cause is not null ?

should we log t if cause is not null ?

setSuccess()

I think this is not correct as voidPromise() will fail when you try to add listeners to it. So better just return a `newFailedFuture()` ?

same as below

done

done

done

done

done

yes I debated this but no-one is currently attaching to the future ... it only returns a ChannelFuture so it doesn't conflict with the existing `ChannelFuture shutdownInput()` from the base interface.

However I can change this in case future code changes i guess

@normanmaurer - previously `shutdownInput` set state internally and did not attempt to set the channel state or change the state of the underlying transport. Now that we do interact with the underlying transport we either close the underlying channel or shutdown channel (if the user enabled this behavior). NIO already [checks the input is open](http://grepcode.com/file/repository.grepcode.com/java/root/jdk/openjdk/8u40-b25/sun/nio/ch/SocketChannelImpl.java#300) before reading anything. This is also now more consistent with EPOLL.

done

This should be `UnsupportedOperationException` 

This should be `UnsupportedOperationException` 

debug ?

debug ?

factory out the above code into a new method `shutdownOutput0(...)` and delegate to it.

See above and delegate to `shutdownOutput0(...)`

same as above

same as above

same as above

+1

+1

sure

sure

I'm not sure it is necessary here. It is makes sense where this is currently done in `NioSocketChannel` because it can be called in a few spots ... but here this method just calls itself (so no duplication). The `shutdown` method is different because it must treat the exceptions differently so we can't really share must there.

see https://github.com/netty/netty/pull/5192#discussion_r62754131

see https://github.com/netty/netty/pull/5192#discussion_r62754131

see https://github.com/netty/netty/pull/5192#discussion_r62754131

This code is different than above. We treat exceptions differently here.

see https://github.com/netty/netty/pull/5192#discussion_r62543513

I think it's fine because it was broken before and it affects the transport implementations only.

notify -> shutdown the input and notify

Isn't `inEventLoop()` relatively an expensive operation? I'd prefer not letting it called twice when we can avoid.

Sgtm I will factor it out


Patch that resolved the issue:
diff --git a/transport-native-epoll/src/main/java/io/netty/channel/epoll/AbstractEpollStreamChannel.java b/transport-native-epoll/src/main/java/io/netty/channel/epoll/AbstractEpollStreamChannel.java
index bc22b37370e..8cdbba542ef 100644
--- a/transport-native-epoll/src/main/java/io/netty/channel/epoll/AbstractEpollStreamChannel.java
+++ b/transport-native-epoll/src/main/java/io/netty/channel/epoll/AbstractEpollStreamChannel.java
@@ -530,7 +530,7 @@ protected Object filterOutboundMessage(Object msg) {
                 "unsupported message type: " + StringUtil.simpleClassName(msg) + EXPECTED_TYPES);
     }
 
-    protected void shutdownOutput0(final ChannelPromise promise) {
+    private void shutdownOutput0(final ChannelPromise promise) {
         try {
             fd().shutdown(false, true);
             promise.setSuccess();
@@ -539,14 +539,37 @@ protected void shutdownOutput0(final ChannelPromise promise) {
         }
     }
 
+    private void shutdownInput0(final ChannelPromise promise) {
+        try {
+            fd().shutdown(true, false);
+            promise.setSuccess();
+        } catch (Throwable cause) {
+            promise.setFailure(cause);
+        }
+    }
+
+    private void shutdown0(final ChannelPromise promise) {
+        try {
+            fd().shutdown(true, true);
+            promise.setSuccess();
+        } catch (Throwable cause) {
+            promise.setFailure(cause);
+        }
+    }
+
+    @Override
+    public boolean isOutputShutdown() {
+        return fd().isOutputShutdown();
+    }
+
     @Override
     public boolean isInputShutdown() {
         return fd().isInputShutdown();
     }
 
     @Override
-    public boolean isOutputShutdown() {
-        return fd().isOutputShutdown();
+    public boolean isShutdown() {
+        return fd().isShutdown();
     }
 
     @Override
@@ -580,6 +603,68 @@ public void run() {
         return promise;
     }
 
+    @Override
+    public ChannelFuture shutdownInput() {
+        return shutdownInput(newPromise());
+    }
+
+    @Override
+    public ChannelFuture shutdownInput(final ChannelPromise promise) {
+        Executor closeExecutor = ((EpollStreamUnsafe) unsafe()).prepareToClose();
+        if (closeExecutor != null) {
+            closeExecutor.execute(new OneTimeTask() {
+                @Override
+                public void run() {
+                    shutdownInput0(promise);
+                }
+            });
+        } else {
+            EventLoop loop = eventLoop();
+            if (loop.inEventLoop()) {
+                shutdownInput0(promise);
+            } else {
+                loop.execute(new OneTimeTask() {
+                    @Override
+                    public void run() {
+                        shutdownInput0(promise);
+                    }
+                });
+            }
+        }
+        return promise;
+    }
+
+    @Override
+    public ChannelFuture shutdown() {
+        return shutdown(newPromise());
+    }
+
+    @Override
+    public ChannelFuture shutdown(final ChannelPromise promise) {
+        Executor closeExecutor = ((EpollStreamUnsafe) unsafe()).prepareToClose();
+        if (closeExecutor != null) {
+            closeExecutor.execute(new OneTimeTask() {
+                @Override
+                public void run() {
+                    shutdown0(promise);
+                }
+            });
+        } else {
+            EventLoop loop = eventLoop();
+            if (loop.inEventLoop()) {
+                shutdown0(promise);
+            } else {
+                loop.execute(new OneTimeTask() {
+                    @Override
+                    public void run() {
+                        shutdown0(promise);
+                    }
+                });
+            }
+        }
+        return promise;
+    }
+
     @Override
     protected void doClose() throws Exception {
         try {
diff --git a/transport-rxtx/src/main/java/io/netty/channel/rxtx/RxtxChannel.java b/transport-rxtx/src/main/java/io/netty/channel/rxtx/RxtxChannel.java
index 07628a944ae..7ff98e06ef9 100644
--- a/transport-rxtx/src/main/java/io/netty/channel/rxtx/RxtxChannel.java
+++ b/transport-rxtx/src/main/java/io/netty/channel/rxtx/RxtxChannel.java
@@ -18,6 +18,7 @@
 import gnu.io.CommPort;
 import gnu.io.CommPortIdentifier;
 import gnu.io.SerialPort;
+import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelPromise;
 import io.netty.channel.oio.OioByteStreamChannel;
 import io.netty.util.internal.OneTimeTask;
@@ -25,7 +26,14 @@
 import java.net.SocketAddress;
 import java.util.concurrent.TimeUnit;
 
-import static io.netty.channel.rxtx.RxtxChannelOption.*;
+import static io.netty.channel.rxtx.RxtxChannelOption.BAUD_RATE;
+import static io.netty.channel.rxtx.RxtxChannelOption.DATA_BITS;
+import static io.netty.channel.rxtx.RxtxChannelOption.DTR;
+import static io.netty.channel.rxtx.RxtxChannelOption.PARITY_BIT;
+import static io.netty.channel.rxtx.RxtxChannelOption.READ_TIMEOUT;
+import static io.netty.channel.rxtx.RxtxChannelOption.RTS;
+import static io.netty.channel.rxtx.RxtxChannelOption.STOP_BITS;
+import static io.netty.channel.rxtx.RxtxChannelOption.WAIT_TIME;
 
 /**
  * A channel to a serial device using the RXTX library.
@@ -129,6 +137,16 @@ protected void doClose() throws Exception {
         }
     }
 
+    @Override
+    protected boolean isInputShutdown() {
+        return !open;
+    }
+
+    @Override
+    protected ChannelFuture shutdownInput() {
+        return newFailedFuture(new UnsupportedOperationException("shutdownInput"));
+    }
+
     private final class RxtxUnsafe extends AbstractUnsafe {
         @Override
         public void connect(
diff --git a/transport-udt/src/main/java/io/netty/channel/udt/nio/NioUdtByteConnectorChannel.java b/transport-udt/src/main/java/io/netty/channel/udt/nio/NioUdtByteConnectorChannel.java
index 20adae249ba..62dbb9ffd7e 100644
--- a/transport-udt/src/main/java/io/netty/channel/udt/nio/NioUdtByteConnectorChannel.java
+++ b/transport-udt/src/main/java/io/netty/channel/udt/nio/NioUdtByteConnectorChannel.java
@@ -17,10 +17,10 @@
 
 import com.barchart.udt.TypeUDT;
 import com.barchart.udt.nio.SocketChannelUDT;
-
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelException;
+import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelMetadata;
 import io.netty.channel.FileRegion;
 import io.netty.channel.RecvByteBufAllocator;
@@ -34,7 +34,7 @@
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
 
-import static java.nio.channels.SelectionKey.*;
+import static java.nio.channels.SelectionKey.OP_CONNECT;
 
 /**
  * Byte Channel Connector for UDT Streams.
@@ -149,6 +149,11 @@ protected int doWriteBytes(final ByteBuf byteBuf) throws Exception {
         return byteBuf.readBytes(javaChannel(), expectedWrittenBytes);
     }
 
+    @Override
+    protected ChannelFuture shutdownInput() {
+        return newFailedFuture(new UnsupportedOperationException("shutdownInput"));
+    }
+
     @Override
     protected long doWriteFileRegion(FileRegion region) throws Exception {
         throw new UnsupportedOperationException();
diff --git a/transport/src/main/java/io/netty/channel/nio/AbstractNioByteChannel.java b/transport/src/main/java/io/netty/channel/nio/AbstractNioByteChannel.java
index 69f99b15758..cc59b84d8a7 100644
--- a/transport/src/main/java/io/netty/channel/nio/AbstractNioByteChannel.java
+++ b/transport/src/main/java/io/netty/channel/nio/AbstractNioByteChannel.java
@@ -19,6 +19,7 @@
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelConfig;
+import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.ChannelOutboundBuffer;
 import io.netty.channel.ChannelPipeline;
@@ -52,6 +53,11 @@ protected AbstractNioByteChannel(Channel parent, SelectableChannel ch) {
         super(parent, ch, SelectionKey.OP_READ);
     }
 
+    /**
+     * Shutdown the input side of the channel.
+     */
+    protected abstract ChannelFuture shutdownInput();
+
     @Override
     protected AbstractNioUnsafe newUnsafe() {
         return new NioByteUnsafe();
@@ -60,10 +66,10 @@ protected AbstractNioUnsafe newUnsafe() {
     protected class NioByteUnsafe extends AbstractNioUnsafe {
 
         private void closeOnRead(ChannelPipeline pipeline) {
-            SelectionKey key = selectionKey();
-            setInputShutdown();
             if (isOpen()) {
                 if (Boolean.TRUE.equals(config().getOption(ChannelOption.ALLOW_HALF_CLOSURE))) {
+                    shutdownInput();
+                    SelectionKey key = selectionKey();
                     key.interestOps(key.interestOps() & ~readInterestOp);
                     pipeline.fireUserEventTriggered(ChannelInputShutdownEvent.INSTANCE);
                 } else {
diff --git a/transport/src/main/java/io/netty/channel/nio/AbstractNioChannel.java b/transport/src/main/java/io/netty/channel/nio/AbstractNioChannel.java
index 582e6ae6faf..8de66aba187 100644
--- a/transport/src/main/java/io/netty/channel/nio/AbstractNioChannel.java
+++ b/transport/src/main/java/io/netty/channel/nio/AbstractNioChannel.java
@@ -60,7 +60,6 @@ public abstract class AbstractNioChannel extends AbstractChannel {
     private final SelectableChannel ch;
     protected final int readInterestOp;
     volatile SelectionKey selectionKey;
-    private volatile boolean inputShutdown;
     boolean readPending;
     private final Runnable clearReadPendingRunnable = new Runnable() {
         @Override
@@ -197,20 +196,6 @@ private void clearReadPending0() {
         ((AbstractNioUnsafe) unsafe()).removeReadOp();
     }
 
-    /**
-     * Return {@code true} if the input of this {@link Channel} is shutdown
-     */
-    protected boolean isInputShutdown() {
-        return inputShutdown;
-    }
-
-    /**
-     * Shutdown the input of this {@link Channel}.
-     */
-    void setInputShutdown() {
-        inputShutdown = true;
-    }
-
     /**
      * Special {@link Unsafe} sub-type which allows to access the underlying {@link SelectableChannel}
      */
@@ -422,10 +407,6 @@ protected void doDeregister() throws Exception {
     @Override
     protected void doBeginRead() throws Exception {
         // Channel.read() or ChannelHandlerContext.read() was called
-        if (inputShutdown) {
-            return;
-        }
-
         final SelectionKey selectionKey = this.selectionKey;
         if (!selectionKey.isValid()) {
             return;
diff --git a/transport/src/main/java/io/netty/channel/nio/AbstractNioMessageChannel.java b/transport/src/main/java/io/netty/channel/nio/AbstractNioMessageChannel.java
index 1d606e075b1..1c7f92aa2b1 100644
--- a/transport/src/main/java/io/netty/channel/nio/AbstractNioMessageChannel.java
+++ b/transport/src/main/java/io/netty/channel/nio/AbstractNioMessageChannel.java
@@ -33,6 +33,7 @@
  * {@link AbstractNioChannel} base class for {@link Channel}s that operate on messages.
  */
 public abstract class AbstractNioMessageChannel extends AbstractNioChannel {
+    boolean inputShutdown;
 
     /**
      * @see {@link AbstractNioChannel#AbstractNioChannel(Channel, SelectableChannel, int)}
@@ -46,6 +47,14 @@ protected AbstractNioUnsafe newUnsafe() {
         return new NioMessageUnsafe();
     }
 
+    @Override
+    protected void doBeginRead() throws Exception {
+        if (inputShutdown) {
+            return;
+        }
+        super.doBeginRead();
+    }
+
     private final class NioMessageUnsafe extends AbstractNioUnsafe {
 
         private final List<Object> readBuf = new ArrayList<Object>();
@@ -98,7 +107,7 @@ public void read() {
                 }
 
                 if (closed) {
-                    setInputShutdown();
+                    inputShutdown = true;
                     if (isOpen()) {
                         close(voidPromise());
                     }
diff --git a/transport/src/main/java/io/netty/channel/oio/AbstractOioByteChannel.java b/transport/src/main/java/io/netty/channel/oio/AbstractOioByteChannel.java
index 8c10cf2b206..15da4099b6b 100644
--- a/transport/src/main/java/io/netty/channel/oio/AbstractOioByteChannel.java
+++ b/transport/src/main/java/io/netty/channel/oio/AbstractOioByteChannel.java
@@ -19,6 +19,7 @@
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelConfig;
+import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelMetadata;
 import io.netty.channel.ChannelOption;
 import io.netty.channel.ChannelOutboundBuffer;
@@ -40,8 +41,6 @@ public abstract class AbstractOioByteChannel extends AbstractOioChannel {
             " (expected: " + StringUtil.simpleClassName(ByteBuf.class) + ", " +
             StringUtil.simpleClassName(FileRegion.class) + ')';
 
-    private volatile boolean inputShutdown;
-
     /**
      * @see AbstractOioByteChannel#AbstractOioByteChannel(Channel)
      */
@@ -49,39 +48,27 @@ protected AbstractOioByteChannel(Channel parent) {
         super(parent);
     }
 
-    protected boolean isInputShutdown() {
-        return inputShutdown;
-    }
-
     @Override
     public ChannelMetadata metadata() {
         return METADATA;
     }
 
     /**
-     * Check if the input was shutdown and if so return {@code true}. The default implementation sleeps also for
-     * {@link #SO_TIMEOUT} milliseconds to simulate some blocking.
+     * Determine if the input side of this channel is shutdown.
+     * @return {@code true} if the input side of this channel is shutdown.
      */
-    protected boolean checkInputShutdown() {
-        if (inputShutdown) {
-            try {
-                Thread.sleep(SO_TIMEOUT);
-            } catch (InterruptedException e) {
-                // ignore
-            }
-            return true;
-        }
-        return false;
-    }
+    protected abstract boolean isInputShutdown();
 
-    void setInputShutdown() {
-        inputShutdown = true;
-    }
+    /**
+     * Shutdown the input side of this channel.
+     * @return A channel future that will complete when the shutdown is complete.
+     */
+    protected abstract ChannelFuture shutdownInput();
 
     private void closeOnRead(ChannelPipeline pipeline) {
-        setInputShutdown();
         if (isOpen()) {
             if (Boolean.TRUE.equals(config().getOption(ChannelOption.ALLOW_HALF_CLOSURE))) {
+                shutdownInput();
                 pipeline.fireUserEventTriggered(ChannelInputShutdownEvent.INSTANCE);
             } else {
                 unsafe().close(unsafe().voidPromise());
diff --git a/transport/src/main/java/io/netty/channel/socket/DuplexChannel.java b/transport/src/main/java/io/netty/channel/socket/DuplexChannel.java
index d34ec36bff1..fc35b0d0158 100644
--- a/transport/src/main/java/io/netty/channel/socket/DuplexChannel.java
+++ b/transport/src/main/java/io/netty/channel/socket/DuplexChannel.java
@@ -32,6 +32,17 @@ public interface DuplexChannel extends Channel {
      */
     boolean isInputShutdown();
 
+    /**
+     * @see Socket#shutdownInput()
+     */
+    ChannelFuture shutdownInput();
+
+    /**
+     * Will notify the given {@link ChannelPromise}
+     * @see Socket#shutdownInput()
+     */
+    ChannelFuture shutdownInput(ChannelPromise promise);
+
     /**
      * @see Socket#isOutputShutdown()
      */
@@ -48,4 +59,22 @@ public interface DuplexChannel extends Channel {
      * Will notify the given {@link ChannelPromise}
      */
     ChannelFuture shutdownOutput(ChannelPromise promise);
+
+    /**
+     * Determine if both the input and output of this channel have been shutdown.
+     */
+    boolean isShutdown();
+
+    /**
+     * Will shutdown the input and output sides of this channel.
+     * @return will be completed when both shutdown operations complete.
+     */
+    ChannelFuture shutdown();
+
+    /**
+     * Will shutdown the input and output sides of this channel.
+     * @param promise will be completed when both shutdown operations complete.
+     * @return will be completed when both shutdown operations complete.
+     */
+    ChannelFuture shutdown(ChannelPromise promise);
 }
diff --git a/transport/src/main/java/io/netty/channel/socket/nio/NioSocketChannel.java b/transport/src/main/java/io/netty/channel/socket/nio/NioSocketChannel.java
index 86253ab8510..dc1cbe1cae2 100644
--- a/transport/src/main/java/io/netty/channel/socket/nio/NioSocketChannel.java
+++ b/transport/src/main/java/io/netty/channel/socket/nio/NioSocketChannel.java
@@ -31,6 +31,8 @@
 import io.netty.channel.socket.SocketChannelConfig;
 import io.netty.util.concurrent.GlobalEventExecutor;
 import io.netty.util.internal.OneTimeTask;
+import io.netty.util.internal.logging.InternalLogger;
+import io.netty.util.internal.logging.InternalLoggerFactory;
 
 import java.io.IOException;
 import java.net.InetSocketAddress;
@@ -46,7 +48,7 @@
  * {@link io.netty.channel.socket.SocketChannel} which uses NIO selector based implementation.
  */
 public class NioSocketChannel extends AbstractNioByteChannel implements io.netty.channel.socket.SocketChannel {
-
+    private static final InternalLogger logger = InternalLoggerFactory.getInstance(NioSocketChannel.class);
     private static final ChannelMetadata METADATA = new ChannelMetadata(false, 16);
     private static final SelectorProvider DEFAULT_SELECTOR_PROVIDER = SelectorProvider.provider();
 
@@ -124,9 +126,20 @@ public boolean isActive() {
         return ch.isOpen() && ch.isConnected();
     }
 
+    @Override
+    public boolean isOutputShutdown() {
+        return javaChannel().socket().isOutputShutdown() || !isActive();
+    }
+
     @Override
     public boolean isInputShutdown() {
-        return super.isInputShutdown();
+        return javaChannel().socket().isInputShutdown() || !isActive();
+    }
+
+    @Override
+    public boolean isShutdown() {
+        Socket socket = javaChannel().socket();
+        return socket.isInputShutdown() && socket.isOutputShutdown() || !isActive();
     }
 
     @Override
@@ -139,11 +152,6 @@ public InetSocketAddress remoteAddress() {
         return (InetSocketAddress) super.remoteAddress();
     }
 
-    @Override
-    public boolean isOutputShutdown() {
-        return javaChannel().socket().isOutputShutdown() || !isActive();
-    }
-
     @Override
     public ChannelFuture shutdownOutput() {
         return shutdownOutput(newPromise());
@@ -175,6 +183,68 @@ public void run() {
         return promise;
     }
 
+    @Override
+    public ChannelFuture shutdownInput() {
+        return shutdownInput(newPromise());
+    }
+
+    @Override
+    public ChannelFuture shutdownInput(final ChannelPromise promise) {
+        Executor closeExecutor = ((NioSocketChannelUnsafe) unsafe()).prepareToClose();
+        if (closeExecutor != null) {
+            closeExecutor.execute(new OneTimeTask() {
+                @Override
+                public void run() {
+                    shutdownInput0(promise);
+                }
+            });
+        } else {
+            EventLoop loop = eventLoop();
+            if (loop.inEventLoop()) {
+                shutdownInput0(promise);
+            } else {
+                loop.execute(new OneTimeTask() {
+                    @Override
+                    public void run() {
+                        shutdownInput0(promise);
+                    }
+                });
+            }
+        }
+        return promise;
+    }
+
+    @Override
+    public ChannelFuture shutdown() {
+        return shutdown(newPromise());
+    }
+
+    @Override
+    public ChannelFuture shutdown(final ChannelPromise promise) {
+        Executor closeExecutor = ((NioSocketChannelUnsafe) unsafe()).prepareToClose();
+        if (closeExecutor != null) {
+            closeExecutor.execute(new OneTimeTask() {
+                @Override
+                public void run() {
+                    shutdown0(promise);
+                }
+            });
+        } else {
+            EventLoop loop = eventLoop();
+            if (loop.inEventLoop()) {
+                shutdown0(promise);
+            } else {
+                loop.execute(new OneTimeTask() {
+                    @Override
+                    public void run() {
+                        shutdown0(promise);
+                    }
+                });
+            }
+        }
+        return promise;
+    }
+
     private void shutdownOutput0(final ChannelPromise promise) {
         try {
             javaChannel().socket().shutdownOutput();
@@ -184,6 +254,41 @@ private void shutdownOutput0(final ChannelPromise promise) {
         }
     }
 
+    private void shutdownInput0(final ChannelPromise promise) {
+        try {
+            javaChannel().socket().shutdownInput();
+            promise.setSuccess();
+        } catch (Throwable t) {
+            promise.setFailure(t);
+        }
+    }
+
+    private void shutdown0(final ChannelPromise promise) {
+        Socket socket = javaChannel().socket();
+        Throwable cause = null;
+        try {
+            socket.shutdownOutput();
+        } catch (Throwable t) {
+            cause = t;
+        }
+        try {
+            socket.shutdownInput();
+        } catch (Throwable t) {
+            if (cause == null) {
+                promise.setFailure(t);
+            } else {
+                logger.debug("Exception suppressed because a previous exception occurred.", t);
+                promise.setFailure(cause);
+            }
+            return;
+        }
+        if (cause == null) {
+            promise.setSuccess();
+        } else {
+            promise.setFailure(cause);
+        }
+    }
+
     @Override
     protected SocketAddress localAddress0() {
         return javaChannel().socket().getLocalSocketAddress();
diff --git a/transport/src/main/java/io/netty/channel/socket/oio/OioSocketChannel.java b/transport/src/main/java/io/netty/channel/socket/oio/OioSocketChannel.java
index 3b34e213ecc..231d9fcd74a 100644
--- a/transport/src/main/java/io/netty/channel/socket/oio/OioSocketChannel.java
+++ b/transport/src/main/java/io/netty/channel/socket/oio/OioSocketChannel.java
@@ -38,11 +38,9 @@
 /**
  * A {@link SocketChannel} which is using Old-Blocking-IO
  */
-public class OioSocketChannel extends OioByteStreamChannel
-                              implements SocketChannel {
+public class OioSocketChannel extends OioByteStreamChannel implements SocketChannel {
 
-    private static final InternalLogger logger =
-            InternalLoggerFactory.getInstance(OioSocketChannel.class);
+    private static final InternalLogger logger = InternalLoggerFactory.getInstance(OioSocketChannel.class);
 
     private final Socket socket;
     private final OioSocketChannelConfig config;
@@ -115,14 +113,19 @@ public boolean isActive() {
         return !socket.isClosed() && socket.isConnected();
     }
 
+    @Override
+    public boolean isOutputShutdown() {
+        return socket.isOutputShutdown() || !isActive();
+    }
+
     @Override
     public boolean isInputShutdown() {
-        return super.isInputShutdown();
+        return socket.isInputShutdown() || !isActive();
     }
 
     @Override
-    public boolean isOutputShutdown() {
-        return socket.isOutputShutdown() || !isActive();
+    public boolean isShutdown() {
+        return socket.isInputShutdown() && socket.isOutputShutdown() || !isActive();
     }
 
     @Override
@@ -130,6 +133,16 @@ public ChannelFuture shutdownOutput() {
         return shutdownOutput(newPromise());
     }
 
+    @Override
+    public ChannelFuture shutdownInput() {
+        return shutdownInput(newPromise());
+    }
+
+    @Override
+    public ChannelFuture shutdown() {
+        return shutdown(newPromise());
+    }
+
     @Override
     protected int doReadBytes(ByteBuf buf) throws Exception {
         if (socket.isClosed()) {
@@ -143,24 +156,82 @@ protected int doReadBytes(ByteBuf buf) throws Exception {
     }
 
     @Override
-    public ChannelFuture shutdownOutput(final ChannelPromise future) {
+    public ChannelFuture shutdownOutput(final ChannelPromise promise) {
+        EventLoop loop = eventLoop();
+        if (loop.inEventLoop()) {
+            try {
+                socket.shutdownOutput();
+                promise.setSuccess();
+            } catch (Throwable t) {
+                promise.setFailure(t);
+            }
+        } else {
+            loop.execute(new OneTimeTask() {
+                @Override
+                public void run() {
+                    shutdownOutput(promise);
+                }
+            });
+        }
+        return promise;
+    }
+
+    @Override
+    public ChannelFuture shutdownInput(final ChannelPromise promise) {
         EventLoop loop = eventLoop();
         if (loop.inEventLoop()) {
+            try {
+                socket.shutdownInput();
+                promise.setSuccess();
+            } catch (Throwable t) {
+                promise.setFailure(t);
+            }
+        } else {
+            loop.execute(new OneTimeTask() {
+                @Override
+                public void run() {
+                    shutdownInput(promise);
+                }
+            });
+        }
+        return promise;
+    }
+
+    @Override
+    public ChannelFuture shutdown(final ChannelPromise promise) {
+        EventLoop loop = eventLoop();
+        if (loop.inEventLoop()) {
+            Throwable cause = null;
             try {
                 socket.shutdownOutput();
-                future.setSuccess();
             } catch (Throwable t) {
-                future.setFailure(t);
+                cause = t;
+            }
+            try {
+                socket.shutdownInput();
+            } catch (Throwable t) {
+                if (cause == null) {
+                    promise.setFailure(t);
+                } else {
+                    logger.debug("Exception suppressed because a previous exception occurred.", t);
+                    promise.setFailure(cause);
+                }
+                return promise;
+            }
+            if (cause == null) {
+                promise.setSuccess();
+            } else {
+                promise.setFailure(cause);
             }
         } else {
             loop.execute(new OneTimeTask() {
                 @Override
                 public void run() {
-                    shutdownOutput(future);
+                    shutdown(promise);
                 }
             });
         }
-        return future;
+        return promise;
     }
 
     @Override
@@ -221,7 +292,6 @@ protected void doClose() throws Exception {
         socket.close();
     }
 
-    @Override
     protected boolean checkInputShutdown() {
         if (isInputShutdown()) {
             try {


Test patch:
diff --git a/transport-native-epoll/src/test/java/io/netty/channel/epoll/EpollSocketShutdownOutputByPeerTest.java b/transport-native-epoll/src/test/java/io/netty/channel/epoll/EpollSocketShutdownOutputByPeerTest.java
new file mode 100644
index 00000000000..5a7bdc1b93f
--- /dev/null
+++ b/transport-native-epoll/src/test/java/io/netty/channel/epoll/EpollSocketShutdownOutputByPeerTest.java
@@ -0,0 +1,29 @@
+/*
+ * Copyright 2016 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.channel.epoll;
+
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.testsuite.transport.TestsuitePermutation;
+import io.netty.testsuite.transport.socket.SocketShutdownOutputByPeerTest;
+
+import java.util.List;
+
+public class EpollSocketShutdownOutputByPeerTest extends SocketShutdownOutputByPeerTest {
+    @Override
+    protected List<TestsuitePermutation.BootstrapFactory<ServerBootstrap>> newFactories() {
+        return EpollSocketTestPermutation.INSTANCE.serverSocket();
+    }
+}
diff --git a/transport-native-epoll/src/test/java/io/netty/channel/epoll/EpollSocketShutdownOutputBySelfTest.java b/transport-native-epoll/src/test/java/io/netty/channel/epoll/EpollSocketShutdownOutputBySelfTest.java
new file mode 100644
index 00000000000..3ad80e472bd
--- /dev/null
+++ b/transport-native-epoll/src/test/java/io/netty/channel/epoll/EpollSocketShutdownOutputBySelfTest.java
@@ -0,0 +1,29 @@
+/*
+ * Copyright 2016 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.channel.epoll;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.testsuite.transport.TestsuitePermutation;
+import io.netty.testsuite.transport.socket.SocketShutdownOutputBySelfTest;
+
+import java.util.List;
+
+public class EpollSocketShutdownOutputBySelfTest extends SocketShutdownOutputBySelfTest {
+    @Override
+    protected List<TestsuitePermutation.BootstrapFactory<Bootstrap>> newFactories() {
+        return EpollSocketTestPermutation.INSTANCE.clientSocket();
+    }
+}