POC output with URring

Sources/NIOPosix/BaseSocketChannel.swift

@@ -374,6 +374,11 @@ class BaseSocketChannel<SocketType: BaseSocketProtocol>: SelectableChannel, Chan
     }
 
     // MARK: Methods to override in subclasses.
+
+    func writeToSocketAsync() throws {
+        fatalError("must be overridden")
+    }
+
     func writeToSocket() throws -> OverallWriteResult {
         fatalError("must be overridden")
     }
@@ -464,17 +469,17 @@ class BaseSocketChannel<SocketType: BaseSocketProtocol>: SelectableChannel, Chan
         fatalError("this must be overridden by sub class")
     }
 
-    /// Buffer a write in preparation for a flush.
-    func bufferPendingWrite(data: NIOAny, promise: EventLoopPromise<Void>?) {
-        fatalError("this must be overridden by sub class")
-    }
-
     /// Mark a flush point. This is called when flush is received, and instructs
     /// the implementation to record the flush.
     func markFlushPoint() {
         fatalError("this must be overridden by sub class")
     }
 
+    /// Buffer a write in preparation for a flush.
+    func bufferPendingWrite(data: NIOAny, promise: EventLoopPromise<Void>?) {
+        fatalError("this must be overridden by sub class")
+    }
+
     /// Called when closing, to instruct the specific implementation to discard all pending
     /// writes.
     func cancelWritesOnClose(error: Error) {
@@ -528,6 +533,15 @@ class BaseSocketChannel<SocketType: BaseSocketProtocol>: SelectableChannel, Chan
         return try self.socket.remoteAddress()
     }
 
+    func flushNowAsync() {
+        do {
+            try self.writeToSocketAsync()
+        }
+        catch let err {
+            self.close0(error: err, mode: .all, promise: nil)
+        }
+    }
+
     /// Flush data to the underlying socket and return if this socket needs to be registered for write notifications.
     ///
     /// This method can be called re-entrantly but it will return immediately because the first call is responsible
@@ -551,8 +565,8 @@ class BaseSocketChannel<SocketType: BaseSocketProtocol>: SelectableChannel, Chan
             self.inFlushNow = false
         }
 
-        var newWriteRegistrationState: IONotificationState = .unregister
         do {
+            var newWriteRegistrationState: IONotificationState = .unregister
             while newWriteRegistrationState == .unregister && self.hasFlushedPendingWrites() && self.isOpen {
                 assert(self.lifecycleManager.isActive)
                 let writeResult = try self.writeToSocket()
@@ -568,7 +582,13 @@ class BaseSocketChannel<SocketType: BaseSocketProtocol>: SelectableChannel, Chan
                     self.pipeline.syncOperations.fireChannelWritabilityChanged()
                 }
             }
-        } catch let err {
+
+            assert((newWriteRegistrationState == .register && self.hasFlushedPendingWrites()) ||
+                   (newWriteRegistrationState == .unregister && !self.hasFlushedPendingWrites()),
+                   "illegal flushNow decision: \(newWriteRegistrationState) and \(self.hasFlushedPendingWrites())")
+            return newWriteRegistrationState
+        }
+        catch let err {
             // If there is a write error we should try drain the inbound before closing the socket as there may be some data pending.
             // We ignore any error that is thrown as we will use the original err to close the channel and notify the user.
             if self.readIfNeeded0() {
@@ -589,11 +609,6 @@ class BaseSocketChannel<SocketType: BaseSocketProtocol>: SelectableChannel, Chan
             // we handled all writes
             return .unregister
         }
-
-        assert((newWriteRegistrationState == .register && self.hasFlushedPendingWrites()) ||
-               (newWriteRegistrationState == .unregister && !self.hasFlushedPendingWrites()),
-               "illegal flushNow decision: \(newWriteRegistrationState) and \(self.hasFlushedPendingWrites())")
-        return newWriteRegistrationState
     }
 
     public final func setOption<Option: ChannelOption>(_ option: Option, value: Option.Value) -> EventLoopFuture<Void> {
@@ -746,16 +761,30 @@ class BaseSocketChannel<SocketType: BaseSocketProtocol>: SelectableChannel, Chan
             return
         }
 
+#if SWIFTNIO_USE_IO_URING && os(Linux)
+        guard self.lifecycleManager.isActive else {
+            return
+        }
+
+        let isFlushPending = self.hasFlushedPendingWrites()
+        self.markFlushPoint()
+
+        if !isFlushPending {
+            self.flushNowAsync()
+        }
+#else
         self.markFlushPoint()
 
         guard self.lifecycleManager.isActive else {
             return
         }
 
-        if !isWritePending() && flushNow() == .register {
+        let isWritePending = self.interestedEvent.contains(.write)
+        if !isWritePending && flushNow() == .register {
             assert(self.lifecycleManager.isPreRegistered)
             registerForWritable()
         }
+#endif
     }
 
     public func read0() {
@@ -901,7 +930,6 @@ class BaseSocketChannel<SocketType: BaseSocketProtocol>: SelectableChannel, Chan
         }
     }
 
-
     public final func register0(promise: EventLoopPromise<Void>?) {
         self.eventLoop.assertInEventLoop()
 
@@ -967,6 +995,15 @@ class BaseSocketChannel<SocketType: BaseSocketProtocol>: SelectableChannel, Chan
 
         self.finishConnect()  // If we were connecting, that has finished.
 
+#if SWIFTNIO_USE_IO_URING && os(Linux)
+        // the only case when it happen with Uring is a socket connect,
+        // writtable event is not needed anymore
+        //let isFlushPending = self.hasFlushedPendingWrites()
+        //if !isFlushPending {
+        //    self.flushNow()
+        //}
+        self.unregisterForWritable()
+#else
         switch self.flushNow() {
         case .unregister:
             // Everything was written or connect was complete, let's unregister from writable.
@@ -975,6 +1012,7 @@ class BaseSocketChannel<SocketType: BaseSocketProtocol>: SelectableChannel, Chan
             assert(!self.isOpen || self.interestedEvent.contains(.write))
             () // nothing to do because given that we just received `writable`, we're still registered for writable.
         }
+#endif
     }
 
     private func finishConnect() {
@@ -1259,10 +1297,6 @@ class BaseSocketChannel<SocketType: BaseSocketProtocol>: SelectableChannel, Chan
         // Do nothing
     }
 
-    private func isWritePending() -> Bool {
-        return self.interestedEvent.contains(.write)
-    }
-
     private final func safeReregister(interested: SelectorEventSet) {
         self.eventLoop.assertInEventLoop()
         assert(self.lifecycleManager.isRegisteredFully)
@@ -1333,6 +1367,12 @@ class BaseSocketChannel<SocketType: BaseSocketProtocol>: SelectableChannel, Chan
         }
         self.registerForReadEOF()
 
+#if SWIFTNIO_USE_IO_URING && os(Linux)
+        let isFlushPending = self.hasFlushedPendingWrites()
+        if !isFlushPending {
+            self.flushNowAsync()
+        }
+#else
         // Flush any pending writes. If after the flush we're still open, make sure
         // our registration is appropriate.
         switch self.flushNow() {
@@ -1345,6 +1385,7 @@ class BaseSocketChannel<SocketType: BaseSocketProtocol>: SelectableChannel, Chan
                 self.unregisterForWritable()
             }
         }
+#endif
 
         self.readIfNeeded0()
     }
@@ -1360,6 +1401,18 @@ class BaseSocketChannel<SocketType: BaseSocketProtocol>: SelectableChannel, Chan
     func reregister(selector: Selector<NIORegistration>, interested: SelectorEventSet) throws {
         fatalError("must override")
     }
+
+    func writeAsync(selector: Selector<NIORegistration>, pointer: UnsafeRawBufferPointer) throws {
+        fatalError("must override")
+    }
+
+    func writeAsync(selector: Selector<NIORegistration>, iovecs: UnsafeBufferPointer<IOVector>) throws {
+        fatalError("must override")
+    }
+
+    func sendFileAsync(selector: Selector<NIORegistration>, src: CInt, offset: Int64, count: UInt32) throws {
+        fatalError("must override")
+    }
 }
 
 extension BaseSocketChannel {

Sources/NIOPosix/BaseStreamSocketChannel.swift

@@ -158,6 +158,42 @@ class BaseStreamSocketChannel<Socket: SocketProtocol>: BaseSocketChannel<Socket>
         return result
     }
 
+    override func writeToSocketAsync() throws {
+        try self.pendingWrites.triggerAppropriateAsyncWriteOperations(
+            scalarBufferAsyncWriteOperation: { ptr in
+                try self.selectableEventLoop.writeAsync(channel: self, pointer: ptr)
+            },
+            vectorBufferAsyncWriteOperation: { iovecs in
+                try self.selectableEventLoop.writeAsync(channel: self, iovecs: iovecs)
+            },
+            scalarFileAsyncWriteOperation: { descriptor, index, endIndex in
+                let count = (endIndex - index)
+                try self.selectableEventLoop.sendFileAsync(channel: self, src: descriptor, offset: Int64(index), count: UInt32(count))
+            })
+    }
+
+    func didAsyncWrite(result: Int32) {
+        if (result > 0) {
+            let (writabilityChange, flushAgain) = self.pendingWrites.didAsyncWrite(written: Int(result))
+            if writabilityChange {
+                self.pipeline.syncOperations.fireChannelWritabilityChanged()
+            }
+            if flushAgain {
+                self.flushNowAsync()
+            }
+        }
+        else {
+            self.pendingWrites.releaseData()
+            let errnoCode = -result
+            if errnoCode == EAGAIN {
+                self.flushNowAsync()
+            }
+            else {
+                assert(false)
+            }
+        }
+    }
+
     final override func writeToSocket() throws -> OverallWriteResult {
         let result = try self.pendingWrites.triggerAppropriateWriteOperations(scalarBufferWriteOperation: { ptr in
             guard ptr.count > 0 else {