diff --git a/Sources/SwiftNetwork/Context/NetworkContext.swift b/Sources/SwiftNetwork/Context/NetworkContext.swift index 4de9107..a06d33a 100644 --- a/Sources/SwiftNetwork/Context/NetworkContext.swift +++ b/Sources/SwiftNetwork/Context/NetworkContext.swift @@ -164,6 +164,26 @@ public final class NetworkContext: NetworkContextProtocol, @unchecked Sendable { scheduler = externalScheduler schedulerIsDefault = false } + + /// Creates a context with a deterministic inline (no-thread) queue, driven by + /// `drainInline()` / `advanceInline(byMilliseconds:)`. For testing. + public static func inlineContext(identifier: String) -> NetworkContext { + let globals = Globals(inline: true) + return NetworkContext( + identifier: identifier, + globals: globals, + scheduler: DefaultScheduler(globals: globals), + schedulerIsDefault: true + ) + } + + public func drainInline() { + globals.queue.drain() + } + + public func advanceInline(byMilliseconds milliseconds: Int) { + globals.queue.advance(byMilliseconds: milliseconds) + } #endif var disableLogging: Bool { @@ -176,7 +196,7 @@ public final class NetworkContext: NetworkContextProtocol, @unchecked Sendable { #if !NETWORK_DRIVERKIT && !NETWORK_STANDALONE func assert() { if schedulerIsDefault { - dispatchPrecondition(condition: DispatchPredicate.onQueue(queue)) + queue.assertQueue() } else { precondition(scheduler.runningInScheduler, "Not running on context scheduler") } @@ -250,11 +270,11 @@ extension NetworkContext { final class Globals { final class TimerList { var entries = NetworkPriorityQueue() - var queue: DispatchQueue - var timerSource: (any DispatchSourceTimer)? + var queue: NetworkQueue + var timerSource: NetworkQueueSource? var currentTarget: DispatchTime = .distantFuture - init(queue: DispatchQueue) { + init(queue: NetworkQueue) { self.queue = queue } @@ -267,12 +287,12 @@ extension NetworkContext { return } while !entries.isEmpty { - let now = DispatchTime.now() + let now = queue.now let targetTime = entries.first.targetTime if targetTime > now { // Target is in future, reset and return currentTarget = targetTime - timerSource.schedule(deadline: targetTime) + timerSource.setTimerValues(fireTime: targetTime) return } guard var entry = entries.pop() else { @@ -320,25 +340,28 @@ extension NetworkContext { // Reset the target time if needed if needsReschedule { if timerSource == nil { - timerSource = DispatchSource.makeTimerSource(queue: queue) + timerSource = queue.createSource(.timer) { [weak self] in + self?.runTimer() + } } currentTarget = targetTime if let timerSource { - let timerHandler = DispatchWorkItem { - self.runTimer() - } - timerSource.setEventHandler(handler: timerHandler) - timerSource.schedule(deadline: currentTarget) + timerSource.setTimerValues(fireTime: currentTarget) timerSource.activate() } } } } var timerList: TimerList - var queue: DispatchQueue + var queue: NetworkQueue init(label: String) { - queue = DispatchQueue(label: "networking context") + queue = NetworkQueue(label: "networking context") + timerList = TimerList(queue: queue) + } + + init(inline: Bool) { + queue = inline ? NetworkQueue() : NetworkQueue(label: "networking context") timerList = TimerList(queue: queue) } } @@ -350,13 +373,13 @@ extension NetworkContext { } /// Runs an immediate task. No assumptions are made about how the task is run. func runImmediate(_ task: @escaping (() -> Void)) { - globals.queue.async(execute: DispatchWorkItem(block: task)) + globals.queue.async(task) } /// Schedules a task to run after a delay, using a reference. /// /// The `milliseconds` parameter specifies the delay before the task runs. func schedule(_ task: @escaping (() -> Void), milliseconds: Int64, reference: TimerReference) { - let targetTime = DispatchTime.now() + DispatchTimeInterval.milliseconds(Int(milliseconds)) + let targetTime = globals.queue.now + DispatchTimeInterval.milliseconds(Int(milliseconds)) globals.timerList.insert(targetTime: targetTime, reference: reference, task: task) } /// Unschedules a task with a reference. @@ -365,8 +388,7 @@ extension NetworkContext { } /// A Boolean value that indicates whether the current code is running in the scheduler. var runningInScheduler: Bool { - // TODO: Not supported by DispatchQueue - fatalError("Unsupported") + globals.queue.isCurrent } } } @@ -376,7 +398,7 @@ extension NetworkContext { @available(Network 0.1.0, *) extension NetworkContext { - var queue: DispatchQueue { + var queue: NetworkQueue { globals.queue } diff --git a/Sources/SwiftNetwork/Context/NetworkQueue.swift b/Sources/SwiftNetwork/Context/NetworkQueue.swift new file mode 100644 index 0000000..bb8a648 --- /dev/null +++ b/Sources/SwiftNetwork/Context/NetworkQueue.swift @@ -0,0 +1,275 @@ +//===----------------------------------------------------------------------===// +// +// This source file is part of the Swift open source project +// +// Copyright (c) 2026 Apple Inc. and the Swift project authors +// Licensed under Apache License v2.0 +// +// See LICENSE.txt for license information +// See CONTRIBUTORS.txt for the list of Swift project authors +// +// SPDX-License-Identifier: Apache-2.0 +// +//===----------------------------------------------------------------------===// + +#if canImport(Dispatch) +import Dispatch + +/// Serial execution context for the networking stack, over `DispatchQueue`. Work and event +/// sources are created through the queue rather than a raw `DispatchQueue`. +/// +/// Two backends: thread-backed dispatch, and an inline (no-thread) mode with a virtual clock +/// driven by `drain()` / `advance(byMilliseconds:)` for deterministic tests. Inline mode is a +/// single-threaded test tool: it must only be driven from one thread at a time. +@available(Network 0.1.0, *) +final class NetworkQueue: @unchecked Sendable { + enum SourceType { case read, write, timer } + + private let dispatchQueue: DispatchQueue? + private let specificKey = DispatchSpecificKey() + + // Inline-mode state. Only touched while draining on the single driving thread. + private var pending: [() -> Void] = [] + private var inlineSources: [NetworkQueueSource] = [] + private var virtualNanos: UInt64 = DispatchTime.now().uptimeNanoseconds + private var pumping = false + + var isInline: Bool { dispatchQueue == nil } + + init(label: String) { + let queue = DispatchQueue(label: label) + self.dispatchQueue = queue + queue.setSpecific(key: specificKey, value: ObjectIdentifier(self)) + } + + /// Creates an inline (no-thread) queue. + init() { + self.dispatchQueue = nil + } + + /// Real time for dispatch, virtual time for inline. + var now: DispatchTime { + isInline ? DispatchTime(uptimeNanoseconds: virtualNanos) : .now() + } + + var isCurrent: Bool { + if isInline { return pumping } + return DispatchQueue.getSpecific(key: specificKey) == ObjectIdentifier(self) + } + + func async(_ block: @escaping () -> Void) { + if let dispatchQueue { + dispatchQueue.async(execute: DispatchWorkItem(block: block)) + } else { + pending.append(block) + } + } + + func barrierAsync(_ block: @escaping () -> Void) { + if let dispatchQueue { + dispatchQueue.async(execute: DispatchWorkItem(flags: .barrier, block: block)) + } else { + pending.append(block) + } + } + + /// Runs inline if already on the queue, otherwise enqueues. + func asyncIfNeeded(_ block: @escaping () -> Void) { + isCurrent ? block() : async(block) + } + + func assertQueue() { + if let dispatchQueue { + dispatchPrecondition(condition: DispatchPredicate.onQueue(dispatchQueue)) + } else { + precondition(pumping, "Not running on inline queue") + } + } + + // MARK: - Event sources + + /// Creates a suspended source. `fileDescriptor`/`mask` are unused for `.timer`. + func createSource( + _ type: SourceType, + fileDescriptor: Int32 = -1, + mask: UInt = 0, + block: @escaping () -> Void, + cancelBlock: (() -> Void)? = nil + ) -> NetworkQueueSource { + if let dispatchQueue { + let source: any DispatchSourceProtocol + switch type { + case .read: source = DispatchSource.makeReadSource(fileDescriptor: fileDescriptor, queue: dispatchQueue) + case .write: source = DispatchSource.makeWriteSource(fileDescriptor: fileDescriptor, queue: dispatchQueue) + case .timer: source = DispatchSource.makeTimerSource(queue: dispatchQueue) + } + source.setEventHandler(handler: block) + if let cancelBlock { + source.setCancelHandler(handler: cancelBlock) + } + return NetworkQueueSource(dispatch: source) + } + // Inline mode has no kernel event delivery; only timers are supported. + precondition(type == .timer, "Inline NetworkQueue supports only timer sources") + let source = NetworkQueueSource(inlineTimerOn: self, block: block, cancelBlock: cancelBlock) + inlineSources.append(source) + return source + } + + fileprivate func removeInlineSource(_ source: NetworkQueueSource) { + inlineSources.removeAll { $0 === source } + } + + // MARK: - Inline pump + + /// Runs pending work and fires timers due at the current virtual time. + func drain() { + guard isInline else { return } + precondition(!pumping, "Reentrant or concurrent drain of inline NetworkQueue") + pumping = true + defer { pumping = false } + while true { + if !pending.isEmpty { + pending.removeFirst()() + } else if let timer = earliestDueTimer() { + timer.fireInline() + } else { + break + } + } + } + + func advance(byMilliseconds milliseconds: Int) { + guard isInline else { return } + virtualNanos &+= UInt64(milliseconds) * 1_000_000 + drain() + } + + private func earliestDueTimer() -> NetworkQueueSource? { + let deadline = now + var earliest: NetworkQueueSource? + for source in inlineSources where source.isDueTimer(at: deadline) { + if earliest == nil || source.fireTime < earliest!.fireTime { + earliest = source + } + } + return earliest + } +} + +/// Opaque cancellable event source vended by `NetworkQueue`, backed by a `DispatchSource` +/// or an inline timer. +@available(Network 0.1.0, *) +final class NetworkQueueSource: @unchecked Sendable { + /// Inline timer state, allocated only for inline sources. + fileprivate final class Inline { + weak var queue: NetworkQueue? + let block: () -> Void + let cancelBlock: (() -> Void)? + var fireTime: DispatchTime = .distantFuture + var interval: UInt64 = .max + var armed = false + var cancelled = false + + init(queue: NetworkQueue, block: @escaping () -> Void, cancelBlock: (() -> Void)?) { + self.queue = queue + self.block = block + self.cancelBlock = cancelBlock + } + } + + private enum Backend { + case dispatch(any DispatchSourceProtocol) + case inline(Inline) + } + + private let backend: Backend + + fileprivate init(dispatch source: any DispatchSourceProtocol) { + self.backend = .dispatch(source) + } + + fileprivate init(inlineTimerOn queue: NetworkQueue, block: @escaping () -> Void, cancelBlock: (() -> Void)?) { + self.backend = .inline(Inline(queue: queue, block: block, cancelBlock: cancelBlock)) + } + + var data: UInt { + if case .dispatch(let source) = backend { return source.data } + return 0 + } + + fileprivate var fireTime: DispatchTime { + if case .inline(let inline) = backend { return inline.fireTime } + return .distantFuture + } + + /// Timer sources only. + func setTimerValues(fireTime: DispatchTime, interval: UInt64 = .max, leeway: UInt64 = 0) { + switch backend { + case .dispatch(let source): + guard let timer = source as? any DispatchSourceTimer else { return } + if interval == .max { + timer.schedule(deadline: fireTime, leeway: .nanoseconds(Int(leeway))) + } else { + timer.schedule(deadline: fireTime, repeating: .nanoseconds(Int(interval)), leeway: .nanoseconds(Int(leeway))) + } + case .inline(let inline): + inline.fireTime = fireTime + inline.interval = interval + } + } + + func activate() { + switch backend { + case .dispatch(let source): source.activate() + case .inline(let inline): inline.armed = true + } + } + + func resume() { + switch backend { + case .dispatch(let source): source.resume() + case .inline(let inline): inline.armed = true + } + } + + func suspend() { + switch backend { + case .dispatch(let source): source.suspend() + case .inline(let inline): inline.armed = false + } + } + + func cancel() { + switch backend { + case .dispatch(let source): + source.cancel() + case .inline(let inline): + guard !inline.cancelled else { return } + inline.cancelled = true + inline.armed = false + inline.queue?.removeInlineSource(self) + inline.cancelBlock?() + } + } + + fileprivate func isDueTimer(at deadline: DispatchTime) -> Bool { + guard case .inline(let inline) = backend else { return false } + return inline.armed && !inline.cancelled && inline.fireTime <= deadline + } + + fileprivate func fireInline() { + guard case .inline(let inline) = backend else { return } + let scheduledAt = inline.fireTime + inline.block() + // If the block rescheduled or cancelled, honor that; otherwise consume a one-shot + // or advance a repeating timer so drain() makes progress. + guard !inline.cancelled, inline.fireTime == scheduledAt else { return } + if inline.interval != .max && inline.interval > 0 { + inline.fireTime = DispatchTime(uptimeNanoseconds: inline.fireTime.uptimeNanoseconds &+ inline.interval) + } else { + inline.armed = false + } + } +} +#endif // canImport(Dispatch) diff --git a/Sources/SwiftNetwork/Protocols/SocketProtocol.swift b/Sources/SwiftNetwork/Protocols/SocketProtocol.swift index 072ce5d..e245d1d 100644 --- a/Sources/SwiftNetwork/Protocols/SocketProtocol.swift +++ b/Sources/SwiftNetwork/Protocols/SocketProtocol.swift @@ -36,8 +36,8 @@ public final class SocketDatagramProtocol: BottomDatagramProtocol, ProtocolInsta var log = NetworkLoggerState() private var socket: SystemSocket? = nil - private var dispatchReadSource: (any DispatchSourceRead)? = nil - private var dispatchWriteSource: (any DispatchSourceWrite)? = nil + private var dispatchReadSource: NetworkQueueSource? = nil + private var dispatchWriteSource: NetworkQueueSource? = nil private var waitingForWritable = false private var inputUnacknowledged = false private var inputSourceSuspended = false @@ -91,7 +91,6 @@ public final class SocketDatagramProtocol: BottomDatagramProtocol, ProtocolInsta inputSourceSuspended = false } inputUnacknowledged = false - dispatchReadSource?.setEventHandler(handler: nil) dispatchReadSource?.cancel() dispatchReadSource = nil cancelWriteSource() @@ -212,8 +211,7 @@ public final class SocketDatagramProtocol: BottomDatagramProtocol, ProtocolInsta private func setupReadSource() { socket?.withFileDescriptor { fileDescriptor in - dispatchReadSource = DispatchSource.makeReadSource(fileDescriptor: fileDescriptor, queue: context.queue) - dispatchReadSource?.setEventHandler { + dispatchReadSource = context.queue.createSource(.read, fileDescriptor: fileDescriptor) { self.handleSocketReadEvent() } dispatchReadSource?.resume() @@ -261,8 +259,7 @@ public final class SocketDatagramProtocol: BottomDatagramProtocol, ProtocolInsta private func setupWriteSource() { socket?.withFileDescriptor { fileDescriptor -> Void in - dispatchWriteSource = DispatchSource.makeWriteSource(fileDescriptor: fileDescriptor, queue: context.queue) - dispatchWriteSource?.setEventHandler { + dispatchWriteSource = context.queue.createSource(.write, fileDescriptor: fileDescriptor) { self.serviceWrites() self.triggerOutboundRoomAvailable() } @@ -283,7 +280,6 @@ public final class SocketDatagramProtocol: BottomDatagramProtocol, ProtocolInsta // DispatchSource must be resumed before cancel dispatchWriteSource.resume() } - dispatchWriteSource.setEventHandler(handler: nil) dispatchWriteSource.cancel() self.dispatchWriteSource = nil waitingForWritable = false diff --git a/Tests/SwiftNetworkTests/NetworkQueueTests.swift b/Tests/SwiftNetworkTests/NetworkQueueTests.swift new file mode 100644 index 0000000..dc95c8a --- /dev/null +++ b/Tests/SwiftNetworkTests/NetworkQueueTests.swift @@ -0,0 +1,144 @@ +//===----------------------------------------------------------------------===// +// +// This source file is part of the Swift open source project +// +// Copyright (c) 2026 Apple Inc. and the Swift project authors +// Licensed under Apache License v2.0 +// +// See LICENSE.txt for license information +// See CONTRIBUTORS.txt for the list of Swift project authors +// +// SPDX-License-Identifier: Apache-2.0 +// +//===----------------------------------------------------------------------===// + +#if canImport(Dispatch) +import XCTest +@_spi(Essentials) @testable import SwiftNetwork + +@available(Network 0.1.0, *) +final class NetworkQueueTests: XCTestCase { + + // MARK: - async / ordering + + func testInlineAsyncRunsInOrderOnlyWhenDrained() { + let queue = NetworkQueue() + var log: [Int] = [] + queue.async { log.append(1) } + queue.async { log.append(2) } + XCTAssertEqual(log, []) + queue.drain() + XCTAssertEqual(log, [1, 2]) + } + + func testAsyncIfNeededRunsInlineWhileDraining() { + let queue = NetworkQueue() + var order: [String] = [] + queue.async { + order.append("outer") + queue.asyncIfNeeded { order.append("inner") } + } + queue.drain() + XCTAssertEqual(order, ["outer", "inner"]) + } + + func testIsCurrentOnlyTrueWhileDraining() { + let queue = NetworkQueue() + XCTAssertFalse(queue.isCurrent) + queue.async { XCTAssertTrue(queue.isCurrent) } + queue.drain() + XCTAssertFalse(queue.isCurrent) + } + + // MARK: - timers + + func testOneShotTimerFiresExactlyOnce() { + let queue = NetworkQueue() + var count = 0 + let source = queue.createSource(.timer) { count += 1 } + source.setTimerValues(fireTime: queue.now + .milliseconds(10)) + source.activate() + + queue.advance(byMilliseconds: 100) + queue.advance(byMilliseconds: 100) + XCTAssertEqual(count, 1, "one-shot must not re-fire and must not loop") + } + + func testRepeatingTimerFiresPerInterval() { + let queue = NetworkQueue() + var count = 0 + let source = queue.createSource(.timer) { count += 1 } + source.setTimerValues( + fireTime: queue.now + .milliseconds(10), + interval: 10_000_000 // 10ms + ) + source.activate() + + queue.advance(byMilliseconds: 35) // fires at 10, 20, 30 + XCTAssertEqual(count, 3) + } + + func testSuspendedTimerDoesNotFire() { + let queue = NetworkQueue() + var fired = false + let source = queue.createSource(.timer) { fired = true } + source.setTimerValues(fireTime: queue.now + .milliseconds(10)) + // never activated + queue.advance(byMilliseconds: 100) + XCTAssertFalse(fired) + } + + func testCancelledTimerDoesNotFireAndRunsCancelBlock() { + let queue = NetworkQueue() + var fired = false + var cancelledRan = false + let source = queue.createSource(.timer, block: { fired = true }, cancelBlock: { cancelledRan = true }) + source.setTimerValues(fireTime: queue.now + .milliseconds(10)) + source.activate() + source.cancel() + + queue.advance(byMilliseconds: 100) + XCTAssertFalse(fired) + XCTAssertTrue(cancelledRan) + } + + func testCancelIsIdempotent() { + let queue = NetworkQueue() + var cancelCount = 0 + let source = queue.createSource(.timer, block: {}, cancelBlock: { cancelCount += 1 }) + source.activate() + source.cancel() + source.cancel() + XCTAssertEqual(cancelCount, 1) + } + + func testEarliestTimerFiresFirst() { + let queue = NetworkQueue() + var order: [String] = [] + let late = queue.createSource(.timer) { order.append("late") } + late.setTimerValues(fireTime: queue.now + .milliseconds(50)) + late.activate() + let early = queue.createSource(.timer) { order.append("early") } + early.setTimerValues(fireTime: queue.now + .milliseconds(10)) + early.activate() + + queue.advance(byMilliseconds: 100) + XCTAssertEqual(order, ["early", "late"]) + } + + // MARK: - inline restrictions + + func testInlineTimerSourceConstructs() { + let queue = NetworkQueue() + // Read/write sources need kernel event delivery and trap inline; timer sources are fine. + let timer = queue.createSource(.timer) {} + XCTAssertNotNil(timer) + } + + func testDataIsZeroForInlineSource() { + let queue = NetworkQueue() + let source = queue.createSource(.timer) {} + XCTAssertEqual(source.data, 0) + } +} +#endif diff --git a/Tests/SwiftNetworkTests/SwiftNetworkContextTests.swift b/Tests/SwiftNetworkTests/SwiftNetworkContextTests.swift index 2e3dae8..e5a7036 100644 --- a/Tests/SwiftNetworkTests/SwiftNetworkContextTests.swift +++ b/Tests/SwiftNetworkTests/SwiftNetworkContextTests.swift @@ -46,4 +46,42 @@ final class SwiftNetworkContextTests: NetTestCase { wait(for: [expectation], timeout: 5.0) } + + func testInlineAsyncDrains() { + let context = NetworkContext.inlineContext(identifier: "test") + var log: [String] = [] + + context.async { log.append("a") } + context.async { log.append("b") } + XCTAssertEqual(log, [], "inline work must not run until drained") + + context.drainInline() + XCTAssertEqual(log, ["a", "b"]) + } + + func testInlineTimerFiresOnVirtualClock() { + let context = NetworkContext.inlineContext(identifier: "test") + var fired = false + + context.resetTimer(for: TimerReference(index: 1), to: .milliseconds(100) { fired = true }) + context.drainInline() + XCTAssertFalse(fired, "timer not yet due") + + context.advanceInline(byMilliseconds: 50) + XCTAssertFalse(fired, "still not due at t=50") + + context.advanceInline(byMilliseconds: 60) + XCTAssertTrue(fired, "timer due at t=110") + } + + func testInlineTimerUnschedule() { + let context = NetworkContext.inlineContext(identifier: "test") + var fired = false + let ref = TimerReference(index: 2) + + context.resetTimer(for: ref, to: .milliseconds(100) { fired = true }) + context.resetTimer(for: ref, to: .unschedule) + context.advanceInline(byMilliseconds: 200) + XCTAssertFalse(fired, "unscheduled timer must not fire") + } }