dnrops
/
swift-nio
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

support for initializing off the system file descriptors (#285) 

Motivation:

In some contexts it is important to be able to pass file descriptors
around instead of specifying the particular host and port. This allows removing
certain windows for race conditions, and has positive security implications.

Modifications:

Added `withConnectedSocket(_:)` call to the `ClientBootstrap` that can
be used to create a new `Channel` out of existing file descriptor
representing the connected socket.

Result:

Allows initializing `Channel`s off the existing file descriptors.
This commit is contained in:
Lev Walkin 2018-04-19 00:58:39 -07:00 committed by Norman Maurer
parent 2bce766c0c
commit 338cfb541b
11 changed files with 256 additions and 57 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
Sources/NIO/BaseSocketChannel.swift
View File

@ -718,6 +718,14 @@ class BaseSocketChannel<T: BaseSocket>: SelectableChannel, ChannelCore {
}
}
public final func registerAlreadyConfigured0(promise: EventLoopPromise<Void>?) {
assert(self.eventLoop.inEventLoop)
assert(self.isOpen)
assert(!self.lifecycleManager.isActive)
register0(promise: nil)
becomeActive0(promise: promise)
}
public final func triggerUserOutboundEvent0(_ event: Any, promise: EventLoopPromise<Void>?) {
promise?.succeed(result: ())
}

183
Sources/NIO/Bootstrap.swift
View File

@ -12,7 +12,7 @@
//
//===----------------------------------------------------------------------===//
/// A `ServerBootstrap` is an easy way to bootstrap a `ServerChannel` when creating network servers.
/// A `ServerBootstrap` is an easy way to bootstrap a `ServerSocketChannel` when creating network servers.
///
/// Example:
///
@ -63,7 +63,7 @@ public final class ServerBootstrap {
/// Create a `ServerBootstrap` for the `EventLoopGroup` `group`.
///
/// - parameters:
/// - group: The `EventLoopGroup` to use for the `ServerChannel`.
/// - group: The `EventLoopGroup` to use for the `ServerSocketChannel`.
public convenience init(group: EventLoopGroup) {
self.init(group: group, childGroup: group)
}
@ -132,12 +132,8 @@ public final class ServerBootstrap {
/// - host: The host to bind on.
/// - port: The port to bind on.
public func bind(host: String, port: Int) -> EventLoopFuture<Channel> {
let evGroup = group
do {
let address = try SocketAddress.newAddressResolving(host: host, port: port)
return bind0(eventLoopGroup: evGroup, to: address)
} catch let err {
return evGroup.next().newFailedFuture(error: err)
return bind0 {
return try SocketAddress.newAddressResolving(host: host, port: port)
}
}
@ -146,7 +142,7 @@ public final class ServerBootstrap {
/// - parameters:
/// - address: The `SocketAddress` to bind on.
public func bind(to address: SocketAddress) -> EventLoopFuture<Channel> {
return bind0(eventLoopGroup: group, to: address)
return bind0 { address }
}
/// Bind the `ServerSocketChannel` to a UNIX Domain Socket.
@ -154,46 +150,74 @@ public final class ServerBootstrap {
/// - parameters:
/// - unixDomainSocketPath: The _Unix domain socket_ path to bind to. `unixDomainSocketPath` must not exist, it will be created by the system.
public func bind(unixDomainSocketPath: String) -> EventLoopFuture<Channel> {
let evGroup = group
do {
let address = try SocketAddress(unixDomainSocketPath: unixDomainSocketPath)
return bind0(eventLoopGroup: evGroup, to: address)
} catch let err {
return evGroup.next().newFailedFuture(error: err)
return bind0 {
try SocketAddress(unixDomainSocketPath: unixDomainSocketPath)
}
}
private func bind0(eventLoopGroup: EventLoopGroup, to address: SocketAddress) -> EventLoopFuture<Channel> {
/// Use the existing bound socket file descriptor.
///
/// - parameters:
/// - descriptor: The _Unix file descriptor_ representing the bound stream socket.
public func withBoundSocket(descriptor: CInt) -> EventLoopFuture<Channel> {
func makeChannel(_ eventLoop: SelectableEventLoop, _ childEventLoopGroup: EventLoopGroup) throws -> ServerSocketChannel {
return try ServerSocketChannel(descriptor: descriptor, eventLoop: eventLoop, group: childEventLoopGroup)
}
return bind0(makeServerChannel: makeChannel) { (eventLoop, serverChannel) in
let promise: EventLoopPromise<Void> = eventLoop.newPromise()
serverChannel.registerAlreadyConfigured0(promise: promise)
return promise.futureResult
}
}
private func bind0(_ makeSocketAddress: () throws -> SocketAddress) -> EventLoopFuture<Channel> {
let address: SocketAddress
do {
address = try makeSocketAddress()
} catch {
return group.next().newFailedFuture(error: error)
}
func makeChannel(_ eventLoop: SelectableEventLoop, _ childEventLoopGroup: EventLoopGroup) throws -> ServerSocketChannel {
return try ServerSocketChannel(eventLoop: eventLoop,
group: childEventLoopGroup,
protocolFamily: address.protocolFamily)
}
return bind0(makeServerChannel: makeChannel) { (eventGroup, serverChannel) in
serverChannel.registerAndDoSynchronously { serverChannel in
serverChannel.bind(to: address)
}
}
}
private func bind0(makeServerChannel: (_ eventLoop: SelectableEventLoop, _ childGroup: EventLoopGroup) throws -> ServerSocketChannel, _ register: @escaping (EventLoop, ServerSocketChannel) -> EventLoopFuture<Void>) -> EventLoopFuture<Channel> {
let eventLoop = self.group.next()
let childEventLoopGroup = self.childGroup
let serverChannelOptions = self.serverChannelOptions
let eventLoop = eventLoopGroup.next()
let serverChannelInit = self.serverChannelInit ?? { _ in eventLoop.newSucceededFuture(result: ()) }
let childChannelInit = self.childChannelInit
let childChannelOptions = self.childChannelOptions
let promise: EventLoopPromise<Channel> = eventLoop.newPromise()
let future: EventLoopFuture<Channel>
do {
let serverChannel = try ServerSocketChannel(eventLoop: eventLoop as! SelectableEventLoop,
group: childEventLoopGroup,
protocolFamily: address.protocolFamily)
let serverChannel = try makeServerChannel(eventLoop as! SelectableEventLoop, childEventLoopGroup)
serverChannelInit(serverChannel).then {
future = serverChannelInit(serverChannel).then {
serverChannel.pipeline.add(handler: AcceptHandler(childChannelInitializer: childChannelInit,
childChannelOptions: childChannelOptions))
}.then {
serverChannelOptions.applyAll(channel: serverChannel)
}.then {
serverChannel.registerAndDoSynchronously { serverChannel in
serverChannel.bind(to: address)
}
register(eventLoop, serverChannel)
}.map {
serverChannel
}.cascade(promise: promise)
} catch let err {
promise.fail(error: err)
}
} catch {
let promise: EventLoopPromise<Channel> = eventLoop.newPromise()
promise.fail(error: error)
future = promise.futureResult
}
return promise.futureResult
return future
}
private class AcceptHandler: ChannelInboundHandler {
@ -388,6 +412,31 @@ public final class ClientBootstrap {
}
}
/// Use the existing connected socket file descriptor.
///
/// - parameters:
/// - descriptor: The _Unix file descriptor_ representing the connected stream socket.
/// - returns: an `EventLoopFuture<Channel>` to deliver the `Channel` immediately.
public func withConnectedSocket(descriptor: CInt) -> EventLoopFuture<Channel> {
let eventLoop = group.next()
do {
let channelInitializer = self.channelInitializer ?? { _ in eventLoop.newSucceededFuture(result: ()) }
let channel = try SocketChannel(eventLoop: eventLoop as! SelectableEventLoop, descriptor: descriptor)
return channelInitializer(channel).then {
self.channelOptions.applyAll(channel: channel)
}.then {
let promise: EventLoopPromise<Void> = eventLoop.newPromise()
channel.registerAlreadyConfigured0(promise: promise)
return promise.futureResult
}.map {
channel
}
} catch {
return eventLoop.newFailedFuture(error: error)
}
}
private func execute(eventLoop: EventLoop,
protocolFamily: Int32,
_ body: @escaping (Channel) -> EventLoopFuture<Void>) -> EventLoopFuture<Channel> {
@ -470,18 +519,29 @@ public final class DatagramBootstrap {
return self
}
/// Use the existing bound socket file descriptor.
///
/// - parameters:
/// - descriptor: The _Unix file descriptor_ representing the bound datagram socket.
public func withBoundSocket(descriptor: CInt) -> EventLoopFuture<Channel> {
func makeChannel(_ eventLoop: SelectableEventLoop) throws -> DatagramChannel {
return try DatagramChannel(eventLoop: eventLoop, descriptor: descriptor)
}
return bind0(makeChannel: makeChannel) { (eventLoop, channel) in
let promise: EventLoopPromise<Void> = eventLoop.newPromise()
channel.registerAlreadyConfigured0(promise: promise)
return promise.futureResult
}
}
/// Bind the `DatagramChannel` to `host` and `port`.
///
/// - parameters:
/// - host: The host to bind on.
/// - port: The port to bind on.
public func bind(host: String, port: Int) -> EventLoopFuture<Channel> {
let evGroup = group
do {
let address = try SocketAddress.newAddressResolving(host: host, port: port)
return bind0(eventLoopGroup: evGroup, to: address)
} catch let err {
return evGroup.next().newFailedFuture(error: err)
return bind0 {
return try SocketAddress.newAddressResolving(host: host, port: port)
}
}
@ -490,7 +550,7 @@ public final class DatagramBootstrap {
/// - parameters:
/// - address: The `SocketAddress` to bind on.
public func bind(to address: SocketAddress) -> EventLoopFuture<Channel> {
return bind0(eventLoopGroup: group, to: address)
return bind0 { address }
}
/// Bind the `DatagramChannel` to a UNIX Domain Socket.
@ -498,39 +558,52 @@ public final class DatagramBootstrap {
/// - parameters:
/// - unixDomainSocketPath: The path of the UNIX Domain Socket to bind on. `path` must not exist, it will be created by the system.
public func bind(unixDomainSocketPath: String) -> EventLoopFuture<Channel> {
let evGroup = group
do {
let address = try SocketAddress(unixDomainSocketPath: unixDomainSocketPath)
return bind0(eventLoopGroup: evGroup, to: address)
} catch let err {
return evGroup.next().newFailedFuture(error: err)
return bind0 {
return try SocketAddress(unixDomainSocketPath: unixDomainSocketPath)
}
}
private func bind0(eventLoopGroup: EventLoopGroup, to address: SocketAddress) -> EventLoopFuture<Channel> {
let eventLoop = eventLoopGroup.next()
private func bind0(_ makeSocketAddress: () throws -> SocketAddress) -> EventLoopFuture<Channel> {
let address: SocketAddress
do {
address = try makeSocketAddress()
} catch {
return group.next().newFailedFuture(error: error)
}
func makeChannel(_ eventLoop: SelectableEventLoop) throws -> DatagramChannel {
return try DatagramChannel(eventLoop: eventLoop,
protocolFamily: address.protocolFamily)
}
return bind0(makeChannel: makeChannel) { (eventLoop, channel) in
channel.register().then {
_ in return channel.bind(to: address)
}
}
}
private func bind0(makeChannel: (_ eventLoop: SelectableEventLoop) throws -> DatagramChannel, _ registerAndBind: @escaping (EventLoop, DatagramChannel) -> EventLoopFuture<Void>) -> EventLoopFuture<Channel> {
let eventLoop = self.group.next()
let channelInitializer = self.channelInitializer ?? { _ in eventLoop.newSucceededFuture(result: ()) }
let channelOptions = self.channelOptions
let promise: EventLoopPromise<Channel> = eventLoop.newPromise()
let future: EventLoopFuture<Channel>
do {
let channel = try DatagramChannel(eventLoop: eventLoop as! SelectableEventLoop,
protocolFamily: address.protocolFamily)
let channel = try makeChannel(eventLoop as! SelectableEventLoop)
channelInitializer(channel).then {
future = channelInitializer(channel).then {
channelOptions.applyAll(channel: channel)
}.then {
channel.register()
}.then {
channel.bind(to: address)
registerAndBind(eventLoop, channel)
}.map {
channel
}.cascade(promise: promise)
} catch let err {
promise.fail(error: err)
}
} catch {
let promise: EventLoopPromise<Channel> = eventLoop.newPromise()
promise.fail(error: error)
future = promise.futureResult
}
return promise.futureResult
return future
}
}

9
Sources/NIO/Channel.swift
View File

@ -30,6 +30,11 @@ public protocol ChannelCore: class {
/// - promise: The `EventLoopPromise` which should be notified once the operation completes, or nil if no notification should take place.
func register0(promise: EventLoopPromise<Void>?)
/// Register channel as already connected or bound socket.
/// - parameters:
/// - promise: The `EventLoopPromise` which should be notified once the operation completes, or nil if no notification should take place.
func registerAlreadyConfigured0(promise: EventLoopPromise<Void>?)
/// Bind to a `SocketAddress`.
///
/// - parameters:
@ -205,6 +210,10 @@ extension Channel {
pipeline.register(promise: promise)
}
public func registerAlreadyConfigured0(promise: EventLoopPromise<Void>?) {
promise?.fail(error: ChannelError.operationUnsupported)
}
public func triggerUserOutboundEvent(_ event: Any, promise: EventLoopPromise<Void>?) {
pipeline.triggerUserOutboundEvent(event, promise: promise)
}

4
Sources/NIO/DeadChannel.swift
View File

@ -28,6 +28,10 @@ private final class DeadChannelCore: ChannelCore {
promise?.fail(error: ChannelError.ioOnClosedChannel)
}
func registerAlreadyConfigured0(promise: EventLoopPromise<Void>?) {
promise?.fail(error: ChannelError.ioOnClosedChannel)
}
func bind0(to: SocketAddress, promise: EventLoopPromise<Void>?) {
promise?.fail(error: ChannelError.ioOnClosedChannel)
}

6
Sources/NIO/Embedded.swift
View File

@ -203,6 +203,12 @@ class EmbeddedChannelCore: ChannelCore {
pipeline.fireChannelRegistered0()
}
func registerAlreadyConfigured0(promise: EventLoopPromise<Void>?) {
isActive = true
register0(promise: promise)
pipeline.fireChannelActive0()
}
func write0(_ data: NIOAny, promise: EventLoopPromise<Void>?) {
guard let data = data.tryAsIOData() else {
promise?.fail(error: ChannelError.writeDataUnsupported)

13
Sources/NIO/ServerSocket.swift
View File

@ -32,6 +32,19 @@
super.init(descriptor: sock)
}
/// Create a new instance.
///
/// - parameters:
/// - descriptor: The _Unix file descriptor_ representing the bound socket.
/// - setNonBlocking: Set non-blocking mode on the socket.
/// - throws: An `IOError` if socket is invalid.
init(descriptor: CInt, setNonBlocking: Bool = false) throws {
super.init(descriptor: descriptor)
if setNonBlocking {
try self.setNonBlocking()
}
}
/// Start to listen for new connections.
///
/// - parameters:

15
Sources/NIO/Socket.swift
View File

@ -38,6 +38,19 @@ public typealias IOVector = iovec
super.init(descriptor: sock)
}
/// Create a new instance out of an already established socket.
///
/// - parameters:
/// - descriptor: The existing socket descriptor.
/// - setNonBlocking: Set non-blocking mode on the socket.
/// - throws: An `IOError` if could not change the socket into non-blocking
init(descriptor: CInt, setNonBlocking: Bool) throws {
super.init(descriptor: descriptor)
if setNonBlocking {
try self.setNonBlocking()
}
}
/// Create a new instance.
///
/// The ownership of the passed in descriptor is transferred to this class. A user must call `close` to close the underlying
@ -45,7 +58,7 @@ public typealias IOVector = iovec
///
/// - parameters:
/// - descriptor: The file descriptor to wrap.
override init(descriptor: Int32) {
override init(descriptor: CInt) {
super.init(descriptor: descriptor)
}

25
Sources/NIO/SocketChannel.swift
View File

@ -59,6 +59,12 @@ final class SocketChannel: BaseSocketChannel<Socket> {
try super.init(socket: socket, eventLoop: eventLoop, recvAllocator: AdaptiveRecvByteBufferAllocator())
}
init(eventLoop: SelectableEventLoop, descriptor: CInt) throws {
let socket = try Socket(descriptor: descriptor, setNonBlocking: true)
self.pendingWrites = PendingStreamWritesManager(iovecs: eventLoop.iovecs, storageRefs: eventLoop.storageRefs)
try super.init(socket: socket, eventLoop: eventLoop, recvAllocator: AdaptiveRecvByteBufferAllocator())
}
deinit {
// We should never have any pending writes left as otherwise we may leak callbacks
assert(pendingWrites.isEmpty)
@ -316,12 +322,18 @@ final class ServerSocketChannel: BaseSocketChannel<ServerSocket> {
convenience init(eventLoop: SelectableEventLoop, group: EventLoopGroup, protocolFamily: Int32) throws {
try self.init(serverSocket: try ServerSocket(protocolFamily: protocolFamily, setNonBlocking: true), eventLoop: eventLoop, group: group)
}
init(serverSocket: ServerSocket, eventLoop: SelectableEventLoop, group: EventLoopGroup) throws {
self.group = group
try super.init(socket: serverSocket, eventLoop: eventLoop, recvAllocator: AdaptiveRecvByteBufferAllocator())
}
convenience init(descriptor: CInt, eventLoop: SelectableEventLoop, group: EventLoopGroup) throws {
let socket = try ServerSocket(descriptor: descriptor, setNonBlocking: true)
try self.init(serverSocket: socket, eventLoop: eventLoop, group: group)
try self.socket.listen(backlog: backlog)
}
override func registrationFor(interested: SelectorEventSet) -> NIORegistration {
return .serverSocketChannel(self, interested)
}
@ -491,6 +503,17 @@ final class DatagramChannel: BaseSocketChannel<Socket> {
return super.isOpen
}
convenience init(eventLoop: SelectableEventLoop, descriptor: CInt) throws {
let socket = Socket(descriptor: descriptor)
do {
try self.init(socket: socket, eventLoop: eventLoop)
} catch {
_ = try? socket.close()
throw error
}
}
init(eventLoop: SelectableEventLoop, protocolFamily: Int32) throws {
let socket = try Socket(protocolFamily: protocolFamily, type: Posix.SOCK_DGRAM)
do {

4
Tests/NIOTests/CustomChannelTests.swift
View File

@ -37,6 +37,10 @@ private class IntChannelCore: ChannelCore {
promise?.fail(error: NotImplementedError())
}
func registerAlreadyConfigured0(promise: EventLoopPromise<Void>?) {
promise?.fail(error: NotImplementedError())
}
func bind0(to: SocketAddress, promise: EventLoopPromise<Void>?) {
promise?.fail(error: NotImplementedError())
}

2
Tests/NIOTests/SocketChannelTest+XCTest.swift
View File

@ -40,6 +40,8 @@ extension SocketChannelTest {
("testWriteAndFlushServerSocketChannel", testWriteAndFlushServerSocketChannel),
("testConnectServerSocketChannel", testConnectServerSocketChannel),
("testCloseDuringWriteFailure", testCloseDuringWriteFailure),
("testWithConfiguredStreamSocket", testWithConfiguredStreamSocket),
("testWithConfiguredDatagramSocket", testWithConfiguredDatagramSocket),
]
}
}

44
Tests/NIOTests/SocketChannelTest.swift
View File

@ -306,4 +306,48 @@ public class SocketChannelTest : XCTestCase {
XCTFail("Unexpected error \(error)")
}
}
public func testWithConfiguredStreamSocket() throws {
let group = MultiThreadedEventLoopGroup(numThreads: 1)
defer { XCTAssertNoThrow(try group.syncShutdownGracefully()) }
let serverSock = try Socket(protocolFamily: AF_INET, type: Posix.SOCK_STREAM)
try serverSock.bind(to: SocketAddress(ipAddress: "127.0.0.1", port: 0))
let serverChannelFuture = try serverSock.withUnsafeFileDescriptor {
ServerBootstrap(group: group).withBoundSocket(descriptor: dup($0))
}
try serverSock.close()
let serverChannel = try serverChannelFuture.wait()
let clientSock = try Socket(protocolFamily: AF_INET, type: Posix.SOCK_STREAM)
let connected = try clientSock.connect(to: serverChannel.localAddress!)
XCTAssertEqual(connected, true)
let clientChannelFuture = try clientSock.withUnsafeFileDescriptor {
ClientBootstrap(group: group).withConnectedSocket(descriptor: dup($0))
}
try clientSock.close()
let clientChannel = try clientChannelFuture.wait()
XCTAssertEqual(true, clientChannel.isActive)
try serverChannel.close().wait()
try clientChannel.close().wait()
}
public func testWithConfiguredDatagramSocket() throws {
let group = MultiThreadedEventLoopGroup(numThreads: 1)
defer { XCTAssertNoThrow(try group.syncShutdownGracefully()) }
let serverSock = try Socket(protocolFamily: AF_INET, type: Posix.SOCK_DGRAM)
try serverSock.bind(to: SocketAddress(ipAddress: "127.0.0.1", port: 0))
let serverChannelFuture = try serverSock.withUnsafeFileDescriptor {
DatagramBootstrap(group: group).withBoundSocket(descriptor: dup($0))
}
try serverSock.close()
let serverChannel = try serverChannelFuture.wait()
XCTAssertEqual(true, serverChannel.isActive)
try serverChannel.close().wait()
}
}