Avoid zombie pipe channels when receiving writeEOF (#3408)

Motivation

In rare cases where the inbound side of a pipe channel is already closed
(either because it was never open or because it was closed during use),
we can get into trouble if the reader drops the read side of our write
pipe. In that context, the Linux kernel will deliver an EPOLLHUP, but
while we'll close the FD we'll let the channel hang out as a zombie.
This is, obviously, suboptimal.

Modifications

In writeEOF, we check whether the read side is already closed. If it is,
we trigger a close internally to shut the channel down. Added a test for
this as needed, and modified an existing test that would now trip over
this behaviour. Also added a new test for the reverse case to ensure we
don't end up with zombies there (we don't).

Results

No zombie channels.

Author: Lukasa

Sources/NIOPosix/PipeChannel.swift

@@ -121,6 +121,17 @@ final class PipeChannel: BaseStreamSocketChannel<PipePair>, @unchecked Sendable
         }
         try! self.selectableEventLoop.deregister(channel: self, mode: .output)
         try! outputSPH.close()
+
+        // Only close the entire channel if the input is already closed.
+        // If input is still open, we can continue half-duplex operation.
+        var inputIsClosed = true
+        if let inputSPH = self.pipePair.input {
+            inputIsClosed = !inputSPH.isOpen
+        }
+        if inputIsClosed {
+            let error = IOError(errnoCode: EPIPE, reason: "Broken pipe")
+            self.close0(error: error, mode: .all, promise: nil)
+        }
     }
 
     override func shutdownSocket(mode: CloseMode) throws {

Tests/NIOPosixTests/PipeChannelTest.swift

@@ -99,10 +99,17 @@ final class PipeChannelTest: XCTestCase {
 
     func testWriteErrorsCloseChannel() {
         XCTAssertNoThrow(try self.channel.setOption(.allowRemoteHalfClosure, value: true).wait())
-        self.fromChannel.closeFile()
-        var buffer = self.channel.allocator.buffer(capacity: 1)
-        buffer.writeString("X")
-        XCTAssertThrowsError(try self.channel.writeAndFlush(buffer).wait()) { error in
+
+        // We need to wedge the EL open here to make sure that the close of `fromChannel` does
+        // not potentially cause us to handle writeEOF before we attempt to make the write. We
+        // want the _write_ to discover the issue first, not writeEOF.
+        let writeFuture = self.channel.eventLoop.flatSubmit { [fromChannel, channel] in
+            fromChannel!.closeFile()
+            var buffer = channel!.allocator.buffer(capacity: 1)
+            buffer.writeString("X")
+            return channel!.writeAndFlush(buffer)
+        }
+        XCTAssertThrowsError(try writeFuture.wait()) { error in
             if let error = error as? IOError {
                 XCTAssert([EPIPE, EBADF].contains(error.errnoCode), "unexpected errno: \(error)")
             } else {
@@ -199,6 +206,47 @@ final class PipeChannelTest: XCTestCase {
         )
         XCTAssertEqual(UInt8(ascii: "X"), spaceForX)
     }
+
+    func testWriteEndGoingAway() throws {
+        try withPipe { readHandle, writeHandle in
+            let chan = try NIOPipeBootstrap(group: self.group)
+                .takingOwnershipOfDescriptor(output: try writeHandle.takeDescriptorOwnership())
+                .wait()
+
+            let writeResult = chan.writeAndFlush(ByteBuffer(repeating: 0x41, count: 32 * 1024 * 1024))
+            try chan.eventLoop.submit {}.wait()  // wait for write to be enqueued
+
+            try readHandle.close()
+
+            XCTAssertThrowsError(try writeResult.wait()) { error in
+                XCTAssertTrue(error is IOError, "Expected IOError but got \(type(of: error))")
+                if let ioError = error as? IOError {
+                    XCTAssertEqual(ioError.errnoCode, EPIPE, "Expected EPIPE but got \(ioError.errnoCode)")
+                }
+            }
+
+            // Channel should be closed after the write error
+            XCTAssertNoThrow(try chan.closeFuture.wait())
+
+            return []
+        }
+    }
+
+    func testReadEndGoingAway() throws {
+        try withPipe { readHandle, writeHandle in
+            let chan = try NIOPipeBootstrap(group: self.group)
+                .channelOption(ChannelOptions.allowRemoteHalfClosure, value: true)
+                .takingOwnershipOfDescriptor(input: try readHandle.takeDescriptorOwnership())
+                .wait()
+
+            try writeHandle.close()
+
+            // Channel should be closed because the read side is dead.
+            XCTAssertNoThrow(try chan.closeFuture.wait())
+
+            return []
+        }
+    }
 }
 
 extension FileHandle {