Don't call to user code before reconciling Channel state. (#310)

Motivation:

Callouts to user code allows the user to make calls that re-enter
channel code. In the case of channel closure, we could call out to the
user before the channel knew it was completely closed, which would
trigger a safety assertion (in debug mode) or hit a fatalError
(in release mode).

Modifications:

Reconciled channel state before we call out to user code.
Added a test for this case.

Result:

Fewer crashes, better channel state management.

Author: Lukasa

Sources/NIO/BaseSocketChannel.swift

@@ -667,9 +667,13 @@ class BaseSocketChannel<T: BaseSocket>: SelectableChannel, ChannelCore {
 
         // Fail all pending writes and so ensure all pending promises are notified
         self.unsetCachedAddressesFromSocket()
-        self.cancelWritesOnClose(error: error)
 
-        self.lifecycleManager.close(promise: p)(self.pipeline)
+        // Transition our internal state.
+        let callouts = self.lifecycleManager.close(promise: p)
+
+        // Now that our state is sensible, we can call out to user code.
+        self.cancelWritesOnClose(error: error)
+        callouts(self.pipeline)
 
         eventLoop.execute {
             // ensure this is executed in a delayed fashion as the users code may still traverse the pipeline

Tests/NIOTests/SocketChannelTest+XCTest.swift

@@ -39,6 +39,7 @@ extension SocketChannelTest {
                 ("testWriteServerSocketChannel", testWriteServerSocketChannel),
                 ("testWriteAndFlushServerSocketChannel", testWriteAndFlushServerSocketChannel),
                 ("testConnectServerSocketChannel", testConnectServerSocketChannel),
+                ("testCloseDuringWriteFailure", testCloseDuringWriteFailure),
            ]
    }
 }

Tests/NIOTests/SocketChannelTest.swift

@@ -261,4 +261,33 @@ public class SocketChannelTest : XCTestCase {
         try serverChannel.close().wait()
     }
 
+    public func testCloseDuringWriteFailure() throws {
+        let group = MultiThreadedEventLoopGroup(numThreads: 1)
+        defer { XCTAssertNoThrow(try group.syncShutdownGracefully()) }
+
+        let serverChannel = try ServerBootstrap(group: group).bind(host: "127.0.0.1", port: 0).wait()
+        let clientChannel = try ClientBootstrap(group: group).connect(to: serverChannel.localAddress!).wait()
+
+        // Put a write in the channel but don't flush it. We're then going to
+        // close the channel. This should trigger an error callback that will
+        // re-close the channel, which should fail with `alreadyClosed`.
+        var buffer = clientChannel.allocator.buffer(capacity: 12)
+        buffer.write(staticString: "hello")
+        let writeFut = clientChannel.write(buffer).map {
+            XCTFail("Must not succeed")
+        }.thenIfError { error in
+            XCTAssertEqual(error as? ChannelError, ChannelError.alreadyClosed)
+            return clientChannel.close()
+        }
+        XCTAssertNoThrow(try clientChannel.close().wait())
+
+        do {
+            try writeFut.wait()
+            XCTFail("Did not throw")
+        } catch ChannelError.alreadyClosed {
+            // ok
+        } catch {
+            XCTFail("Unexpected error \(error)")
+        }
+    }
 }