BaseStreamSocketChannel half-close allows outstanding writes to complete #3148

rnro · 2025-03-18T15:40:50Z

Motivation:

At the moment half-closes are actioned immediately and fails all outstanding writes. We should refuse new writes but allow these writes to complete before completing the close.

Modifications:

Modify the PendingWritesManager internal buffer to hold an enum of either writes or close events. We use this to store the close and only action it when the preceding writes have been handled.

Result:

Outbound close should no longer fail outstanding writes

Should resolve #3139

Motivation: At the moment half-closes are actioned immediately and fails all outstanding writes. We should refuse new writes but allow these writes to complete before completing the close. Modifications: Modify the PendingWritesManager internal buffer to hold an enum of either writes or close events. We use this to store the close and only action it when the preceding writes have been handled. Result: Outbound close should no longer fail outstanding writes

Sources/NIOPosix/BaseStreamSocketChannel.swift

Lukasa · 2025-03-21T16:54:17Z

Sources/NIOPosix/BaseStreamSocketChannel.swift

+                let outboundCloseState = self.pendingWrites.close(promise)
+                switch outboundCloseState {
+                case .open:
+                    preconditionFailure("Close resulted in an open state, this should never happen")


Let's avoid this precondition failure. Instead, we should produce an error just before line 219, and go ahead and close outbound anyway. We can then drop this to assertionFailure so that it crashes in testing.

Lukasa · 2025-03-21T16:55:28Z

Sources/NIOPosix/BaseStreamSocketChannel.swift

+                    self.pendingWrites.outboundCloseState = .closed
+                case .closed:
+                    ()  // nothing to do
+                }

                self.pipeline.fireUserInboundEventTriggered(ChannelEvent.outputClosed)


This user event now triggers unconditionally, and at the wrong time. It should trigger only after the shutdown. Also, don't forget the unregister for writable

Lukasa · 2025-03-21T16:57:32Z

Sources/NIOPosix/PendingWritesManager.swift

+            case .readyForClose:
+                ()
+            case .pending, .closed:
+                preconditionFailure("close called on channel in unexpected state: \(self.outboundCloseState)")


This is a touch nerve wracking. I can't see any reason this is actually impossible: I think if the user called close(mode: .output) twice in a row without the close having happened, they'd trap here. The same is true for the closed state.

More broadly, we have to tolerate the user doing this more than once. It's an error if they do, but that's ok.

I've attempted account for tolerating being called more than once by cascading promises.

Lukasa · 2025-03-21T16:58:13Z

Sources/NIOPosix/PendingWritesManager.swift

+            case .open:
+                self.outboundCloseState = .readyForClose(promise)
+            case .readyForClose:
+                ()


This code path leaks the promise.

Generally speaking, at this point we've taken ownership of the promise. It's our responsibility to do something with it, which we must do on all code paths.

I've changed the API of this object so that the promise is never surfaced after it has been passed in. The promise must be completed by a call to `closeComplete.

Sources/NIOPosix/PendingWritesManager.swift

Lukasa · 2025-03-21T16:58:50Z

Sources/NIOPosix/PendingWritesManager.swift

@@ -310,6 +321,8 @@ final class PendingStreamWritesManager: PendingWritesManager {

    private(set) var isOpen = true

+    internal var outboundCloseState: CloseState = .open


I think we need a very good reason to have this be anything but private(set).

Lukasa · 2025-03-21T16:59:27Z

Sources/NIOPosix/BaseSocketChannel.swift

+                case .open, .pending:
+                    ()
+                case .readyForClose(let eventLoopPromise):
+                    // TODO: it doesn't seem right that I have to pass an error in here)


Sources/NIOPosix/BaseStreamSocketChannel.swift

Lukasa · 2025-03-21T17:08:52Z

Sources/NIOPosix/PendingWritesManager.swift

+                }
+
+            case .couldNotWriteEverything:
+                ()


This one also feels like it is worth an assertionFailure.

glbrntt · 2025-04-01T14:00:31Z

Sources/NIOPosix/PendingWritesManager.swift

+        static func == (lhs: Self, rhs: Self) -> Bool {
+            switch (lhs, rhs) {
+            case (.writtenCompletely, .writtenCompletely):
+                return true
+            case (.couldNotWriteEverything, .couldNotWriteEverything):
+                return true
+            default:
+                return false
+            }
+        }


I get a bit nervous about ignoring state when comparing. Do we actually need the equality check? Also, it might make more sense to do this on the CloseState and compare the identity of its underlying futures where applicable.

It turns out we don't need it anymore so I've just deleted it.

🤦‍♂️ it's needed in tests. Looking again.

I think this has exposed that I was exposing the promise in a way which wasn't needed anyway because we just passed it right into pendingWrites.close which already held it. I've changed the method to return the CloseResult instead.

glbrntt · 2025-04-01T14:09:40Z

Sources/NIOPosix/PendingWritesManager.swift

    func failAll(error: Error, close: Bool) {
        if close {
            assert(self.isOpen)
            self.isOpen = false
+            self.state.removeAll()?.fail(error)


General rule of thumb: update your state before making outcalls. Failing the promise here might lead to calling back into this type again and we haven't fully reconciled our state yet. We should grab the promise and fail it afterwards.

Thanks for the general guidance 🙂

glbrntt · 2025-04-01T14:11:31Z

Sources/NIOPosix/PendingWritesManager.swift

+                closePromise?.fail(error)
+                self.outboundCloseState = .closed


Same here, update the state then complete the promise

glbrntt · 2025-04-01T14:15:12Z

Sources/NIOPosix/PendingWritesManager.swift

    func failAll(error: Error, close: Bool) {
        if close {
            assert(self.isOpen)
            self.isOpen = false
+            self.state.removeAll()?.fail(error)


Not sure if it's just the diff but it looks like this function also failed all the promises if !close, that doesn't seem to happen anymore.

glbrntt · 2025-04-01T14:22:12Z

Sources/NIOPosix/PendingWritesManager.swift

+                closePromise?.succeed(())
+            }
+
+            self.outboundCloseState = .closed


The state should be updated before completing the promise

glbrntt

This LGTM now but please wait for Cory's review as well.

Lukasa

Cool, this is looking closer, I think I just have some suggestions around promise hygiene and naming.

Lukasa · 2025-04-10T10:06:08Z

Sources/NIOPosix/PendingWritesManager.swift

+    ///
+    /// - Parameters:
+    ///   - promise: Optionally an `EventLoopPromise` that will be succeeded once all outstanding writes have been dealt with
+    func close(_ promise: EventLoopPromise<Void>?) -> CloseResult {


This should probably be called closeOutbound.

Sources/NIOPosix/BaseStreamSocketChannel.swift

Lukasa · 2025-04-10T11:59:19Z

Sources/NIOPosix/PendingWritesManager.swift

+                }
+
+            case .couldNotWriteEverything:
+                assertionFailure("Write result is .couldNotWriteEverything but we have no more writes to perform.")


If this is going to be an assertion failure we should have some fallback logic. Probably that should be throwing an error.

Lukasa · 2025-04-10T12:00:41Z

Sources/NIOPosix/PendingWritesManager.swift

+                self.outboundCloseState = .closed
+            case .pending(let closePromise), .readyForClose(let closePromise):
+                self.outboundCloseState = .closed
+                closePromise?.fail(error)


Hrm, should we return this instead of closing it at this stage?

Can you explain the thinking here? My understanding from our previous discussions was that it would be better that once the higher-level caller handed-off the promise it was the responsibility of the PendingWritesManager to complete the promise, so passing it in was a one-way street. This is why I separated-out CloseResult as its own type, so we have clarity on where the promise is exposed and not exposed.

Are you proposing that this method returns EventLoopPromise<Void>? where the promise is non-nil if the caller needs to do something with it?

Ah, I've just scrolled up and seen #3148 (comment) - that makes sense.

Lukasa · 2025-04-10T13:50:29Z

Sources/NIOPosix/PendingWritesManager.swift

+                closePromise.setOrCascade(to: promise)
+                self.outboundCloseState = .readyForClose(closePromise)
+            case .closed:
+                promise?.succeed(())


As noted above, it's probably better to let the caller complete the promise.

In this case are you proposing that this method returns (CloseResult, EventLoopPromise<Void>?) where the promise is non-nil if the caller needs to do something with it?

Or put the promise in the close result as associated data.

Lukasa · 2025-04-14T09:59:17Z

Sources/NIOPosix/BaseStreamSocketChannel.swift


-                self.pipeline.fireUserInboundEventTriggered(ChannelEvent.outputClosed)
+                let writesCloseResult = self.pendingWrites.closeOutbound(promise)


Ah, ok: there's a test missing here. The unregisterForWritable above isn't appropriate. I believe the following situation is possible:

Many writes are done, such that we are no longer writable at the socket layer.

The user issues a final flush.

The user calls close(mode: .output)

In this case, we'll have flushed, unwritten data, but we will no longer be registered for writable. As long as no other flush or close comes along, we'll stay unregistered for writable, and so we will make no attempt to empty the buffer. That will cause us to be wedged open.

I recommend, before fixing the bug, you write a test that correctly reproduces this wedge, and use that to validate the bug is actually fixed. This is going to be a challenging test to write. It can be written using both regular sockets, but it can also be written using the SAL, which may be easier. Let me know if you'd like to pair on a test for this.

Lukasa · 2025-04-14T10:00:37Z

Sources/NIOPosix/BaseStreamSocketChannel.swift

+                case .closed(let closePromise):
+                    closePromise?.succeed(())
+                case .open:
+                    promise?.fail(ChannelError.inappropriateOperationForState)


This path appears to violate the "pending writes manager takes ownership of the promise" construction.

Lukasa · 2025-04-14T10:02:42Z

Sources/NIOPosix/PendingWritesManager.swift

-    /// Fail all the outstanding writes. This is useful if for example the `Channel` is closed.
-    func failAll(error: Error, close: Bool) {
+    /// Fail all the outstanding writes.
+    func failAll(error: Error, close: Bool) -> CloseResult? {


This "boolean parameter with optional return contingent on the value of the bool" smells like there are two methods here. It may be worth factoring them apart. A safe way to do this, if we want to keep this patch understandable, is to make a separate refactoring PR that only refactors this method into two.

(Sidebar: the two methods are actually implemented as one calling the other: the one with close: true does some stuff, then it calls the one with close: false.)

Lukasa · 2025-04-14T10:06:24Z

Sources/NIOPosix/PendingWritesManager.swift

+                    "We are in .readyForClose state but we still have pending writes. This should never happen."
+                )
+            case .closed:
+                preconditionFailure(


I'm a touch nervous about these precondition failures. Can we come up with a logical behaviour here? Probably it's "fail the promise" and "assertionFailure to get crashes in debug".

Lukasa · 2025-04-14T10:06:56Z

Sources/NIOPosix/PendingWritesManager.swift

+        case .closed:
+            ()  // nothing to do
+        case .open, .pending:
+            preconditionFailure("close complete called on channel in unexpected state: \(self.outboundCloseState)")


Can we make this a thrown error instead of a crash?

Lukasa · 2025-04-14T10:07:37Z

Sources/NIOPosix/PendingWritesManager.swift

+
+        switch self.outboundCloseState {
+        case .readyForClose(let closePromise):
+            self.outboundCloseState = .closed


Ok, let's return the promise here rather than have this type complete it. It's a touch easier for the channel to manage things by completing things itself, rather than having this type do it, because the channel can ensure the state is squared away before making the outcall.

Lukasa · 2025-04-14T10:07:57Z

Sources/NIOPosix/PendingWritesManager.swift

+                        case .pending: .pending
+                        case .readyForClose(let closePromise): .readyForClose(closePromise)
+                        case .closed: .closed(nil)
+                        }


Let's outline this. Looks like an initializer to me.

rnro added the 🔨 semver/patch No public API change. label Mar 18, 2025

rnro force-pushed the queue_outbound_close branch from ef380ee to a9d9404 Compare March 18, 2025 16:32

Lukasa reviewed Mar 18, 2025

View reviewed changes

Sources/NIOPosix/BaseStreamSocketChannel.swift Show resolved Hide resolved

rnro added 5 commits March 21, 2025 10:57

use state machine instead of promise

94f9573

reverting unrelated changes

2603614

DatagramWritesManager only supports full closure

f762b29

formatting

716fc47

Merge branch 'main' into queue_outbound_close

0e6ee40

rnro requested a review from Lukasa March 21, 2025 15:29

Lukasa reviewed Mar 21, 2025

View reviewed changes

rnro added 3 commits March 31, 2025 13:53

review comments

68397e0

formatting

1f0cf18

Merge branch 'main' into queue_outbound_close

081a205

rnro requested a review from glbrntt March 31, 2025 14:30

review comments

32b232a

glbrntt reviewed Apr 1, 2025

View reviewed changes

rnro requested a review from glbrntt April 2, 2025 09:02

review comments

e439537

rnro force-pushed the queue_outbound_close branch from c7f73ff to e439537 Compare April 2, 2025 09:23

Merge branch 'main' into queue_outbound_close

92fa820

glbrntt approved these changes Apr 7, 2025

View reviewed changes

Merge branch 'main' into queue_outbound_close

d41ef1b

rnro requested a review from Lukasa April 10, 2025 05:18

Lukasa reviewed Apr 10, 2025

View reviewed changes

rnro added 4 commits April 11, 2025 09:28

PendingWritesManager returns close promise

e60b390

Merge branch 'main' into queue_outbound_close

828a113

formatting

ddbb053

Merge branch 'main' into queue_outbound_close

4c4bdd9

rnro requested a review from Lukasa April 11, 2025 13:40

Lukasa reviewed Apr 14, 2025

View reviewed changes

		@@ -310,6 +321,8 @@ final class PendingStreamWritesManager: PendingWritesManager {

		private(set) var isOpen = true

		internal var outboundCloseState: CloseState = .open


		self.pipeline.fireUserInboundEventTriggered(ChannelEvent.outputClosed)
		let writesCloseResult = self.pendingWrites.closeOutbound(promise)

BaseStreamSocketChannel half-close allows outstanding writes to complete #3148

Are you sure you want to change the base?

BaseStreamSocketChannel half-close allows outstanding writes to complete #3148

Conversation

rnro commented Mar 18, 2025 • edited Loading

Motivation:

Modifications:

Result:

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rnro Apr 2, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

glbrntt left a comment

Choose a reason for hiding this comment

Lukasa left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rnro commented Mar 18, 2025 •

edited

Loading

rnro Apr 2, 2025 •

edited

Loading