Add AddressedEnvelope to PendingWritesManager.

Author: zaneenders

Sources/NIOCore/AddressedEnvelope.swift

@@ -39,15 +39,24 @@ public struct AddressedEnvelope<DataType> {
         /// Details of any congestion state.
         public var ecnState: NIOExplicitCongestionNotificationState
         public var packetInfo: NIOPacketInfo?
+        public var fd: Int?
 
         public init(ecnState: NIOExplicitCongestionNotificationState) {
             self.ecnState = ecnState
             self.packetInfo = nil
+            self.fd = nil
         }
 
         public init(ecnState: NIOExplicitCongestionNotificationState, packetInfo: NIOPacketInfo?) {
             self.ecnState = ecnState
             self.packetInfo = packetInfo
+            self.fd = nil
+        }
+
+        public init(ecnState: NIOExplicitCongestionNotificationState, fd: Int?) {
+            self.ecnState = ecnState
+            self.packetInfo = nil
+            self.fd = fd
         }
     }
 }

Sources/NIOPosix/BaseStreamSocketChannel.swift

@@ -161,6 +161,9 @@ class BaseStreamSocketChannel<Socket: SocketProtocol>: BaseSocketChannel<Socket>
     }
 
     final override func writeToSocket() throws -> OverallWriteResult {
+        print(#fileID, #function)
+        // TODO: self.socket.sendmmsg(msgs: )
+        // TODO: self.socket.sendmsg(pointer: , destinationPtr: UnsafePointer<sockaddr>?, destinationSize: socklen_t, controlBytes: UnsafeMutableRawBufferPointer)
         let result = try self.pendingWrites.triggerAppropriateWriteOperations(
             scalarBufferWriteOperation: { ptr in
                 guard ptr.count > 0 else {
@@ -294,6 +297,13 @@ class BaseStreamSocketChannel<Socket: SocketProtocol>: BaseSocketChannel<Socket>
             return
         }
 
+        if let envelope = self.tryUnwrapData(data, as: AddressedEnvelope<ByteBuffer>.self) {
+            if self.pendingWrites.add(envelope: envelope, promise: promise) {
+                self.pipeline.syncOperations.fireChannelWritabilityChanged()
+            }
+            return
+        }
+
         let data = self.unwrapData(data, as: IOData.self)
 
         if !self.pendingWrites.add(data: data, promise: promise) {

Sources/NIOPosix/PendingWritesManager.swift

@@ -16,43 +16,10 @@ import Atomics
 import CNIOLinux
 import NIOCore
 
-public struct MetadataEnvelope {
-    public var data: ByteBuffer
-    public var controlData: MessageMetadata? = nil
-
-    public init(data: ByteBuffer) {
-        self.data = data
-    }
-
-    public init(data: ByteBuffer, controlData: MessageMetadata?) {
-        self.data = data
-        self.controlData = controlData
-    }
-
-    /// Any metadata associated with an
-    public struct MessageMetadata: Hashable, Sendable {
-        init() {
-            var m: msghdr = .init()
-            // Create buffer for file descriptor
-            let fake_fd = 6666
-            let fd_pointer = UnsafeMutableRawPointer(bitPattern: fake_fd)
-            m.msg_control = fd_pointer
-            m.msg_controllen = MemoryLayout.size(ofValue: fake_fd)
-            // TODO other platoforms
-            let cm: UnsafeMutablePointer<cmsghdr> = CNIOLinux_CMSG_FIRSTHDR(&m)
-            cm.pointee.cmsg_level = SOL_SOCKET
-            cm.pointee.cmsg_type = Int32(SCM_RIGHTS)
-            cm.pointee.cmsg_len = MemoryLayout.size(ofValue: fake_fd)
-
-            // TODO send `m`
-        }
-    }
-}
-
 private struct PendingStreamWrite {
     var data: IOData
     var promise: Optional<EventLoopPromise<Void>>
-    var metadata: MetadataEnvelope?
+    var metadata: AddressedEnvelope<ByteBuffer>.Metadata?
 }
 
 /// Write result is `.couldNotWriteEverything` but we have no more writes to perform.
@@ -404,7 +371,10 @@ final class PendingStreamWritesManager: PendingWritesManager {
     func add(data: IOData, promise: EventLoopPromise<Void>?) -> Bool {
         assert(self.isOpen)
         self.state.append(PendingStreamWrite(data: data, promise: promise))
+        return _add()
+    }
 
+    private func _add() -> Bool {
         if self.state.bytes > waterMark.high
             && channelWritabilityFlag.compareExchange(expected: true, desired: false, ordering: .relaxed).exchanged
         {
@@ -415,6 +385,14 @@ final class PendingStreamWritesManager: PendingWritesManager {
         return true
     }
 
+    func add(envelope: AddressedEnvelope<ByteBuffer>, promise: EventLoopPromise<Void>?) -> Bool {
+        assert(self.isOpen)
+        self.state.append(
+            PendingStreamWrite(data: .byteBuffer(envelope.data), promise: promise, metadata: envelope.metadata)
+        )
+        return _add()
+    }
+
     /// Returns the best mechanism to write pending data at the current point in time.
     var currentBestWriteMechanism: WriteMechanism {
         self.state.currentBestWriteMechanism
@@ -434,6 +412,7 @@ final class PendingStreamWritesManager: PendingWritesManager {
         scalarFileWriteOperation: (CInt, Int, Int) throws -> IOResult<Int>
     ) throws -> OverallWriteResult {
         try self.triggerWriteOperations { writeMechanism in
+            print(#function, writeMechanism)
             // TODO: add with metadata calls.
             switch writeMechanism {
             case .scalarBufferWrite: