mirror of
https://github.com/apple/swift-nio.git
synced 2026-05-20 20:30:36 +00:00
8d2347a56b
Fixes #3262 by adding the missing APIs. ### Motivation: As explained by the issue linked above, having a non-`Sendable`-requiring variant of the new `scheduleCallback` APIs on `NIOIsolatedEventLoop` can be useful since we're not always dealing with `Sendable` types. ### Modifications: This adds the required `scheduleCallback` APIs to `NIOIsolatedEventLoop` by wrapping the non-`Sendable` `NIOScheduleCallbackHandler` in a `NIOLoopBound`-based handler. ### Result: The two `scheduleCallback`s can be used on `NIOIsolatedEventLoop` too. --------- Co-authored-by: Fabian Fett <fabianfett@apple.com>
1616 lines
68 KiB
Swift
1616 lines
68 KiB
Swift
//===----------------------------------------------------------------------===//
|
|
//
|
|
// This source file is part of the SwiftNIO open source project
|
|
//
|
|
// Copyright (c) 2017-2021 Apple Inc. and the SwiftNIO project authors
|
|
// Licensed under Apache License v2.0
|
|
//
|
|
// See LICENSE.txt for license information
|
|
// See CONTRIBUTORS.txt for the list of SwiftNIO project authors
|
|
//
|
|
// SPDX-License-Identifier: Apache-2.0
|
|
//
|
|
//===----------------------------------------------------------------------===//
|
|
|
|
import NIOConcurrencyHelpers
|
|
|
|
#if canImport(Dispatch)
|
|
import Dispatch
|
|
#endif
|
|
|
|
#if canImport(WASILibc)
|
|
@preconcurrency import WASILibc
|
|
import CNIOWASI
|
|
#endif
|
|
|
|
#if os(Linux)
|
|
import CNIOLinux
|
|
#endif // os(Linux)
|
|
|
|
/// Returned once a task was scheduled on the `EventLoop` for later execution.
|
|
///
|
|
/// A `Scheduled` allows the user to either `cancel()` the execution of the scheduled task (if possible) or obtain a reference to the `EventLoopFuture` that
|
|
/// will be notified once the execution is complete.
|
|
public struct Scheduled<T> {
|
|
@usableFromInline typealias CancelationCallback = @Sendable () -> Void
|
|
@usableFromInline let _promise: EventLoopPromise<T>
|
|
@usableFromInline let _cancellationTask: CancelationCallback
|
|
|
|
@inlinable
|
|
@preconcurrency
|
|
public init(promise: EventLoopPromise<T>, cancellationTask: @escaping @Sendable () -> Void) {
|
|
self._promise = promise
|
|
self._cancellationTask = cancellationTask
|
|
}
|
|
|
|
/// Try to cancel the execution of the scheduled task.
|
|
///
|
|
/// Whether this is successful depends on whether the execution of the task already begun.
|
|
/// This means that cancellation is not guaranteed.
|
|
@inlinable
|
|
public func cancel() {
|
|
self._promise.fail(EventLoopError._cancelled)
|
|
self._cancellationTask()
|
|
}
|
|
|
|
/// Returns the `EventLoopFuture` which will be notified once the execution of the scheduled task completes.
|
|
@inlinable
|
|
public var futureResult: EventLoopFuture<T> {
|
|
self._promise.futureResult
|
|
}
|
|
}
|
|
|
|
extension Scheduled: Sendable {}
|
|
|
|
/// Returned once a task was scheduled to be repeatedly executed on the `EventLoop`.
|
|
///
|
|
/// A `RepeatedTask` allows the user to `cancel()` the repeated scheduling of further tasks.
|
|
public final class RepeatedTask {
|
|
typealias RepeatedTaskCallback = @Sendable (RepeatedTask) -> EventLoopFuture<Void>
|
|
private let delay: TimeAmount
|
|
private let eventLoop: EventLoop
|
|
private let cancellationPromise: EventLoopPromise<Void>?
|
|
private var scheduled: Optional<Scheduled<EventLoopFuture<Void>>>
|
|
private var task: Optional<RepeatedTaskCallback>
|
|
|
|
internal init(
|
|
interval: TimeAmount,
|
|
eventLoop: EventLoop,
|
|
cancellationPromise: EventLoopPromise<Void>? = nil,
|
|
task: @escaping RepeatedTaskCallback
|
|
) {
|
|
self.delay = interval
|
|
self.eventLoop = eventLoop
|
|
self.cancellationPromise = cancellationPromise
|
|
self.task = task
|
|
self.scheduled = nil
|
|
}
|
|
|
|
internal func begin(in delay: TimeAmount) {
|
|
if self.eventLoop.inEventLoop {
|
|
self.begin0(in: delay)
|
|
} else {
|
|
self.eventLoop.execute {
|
|
self.begin0(in: delay)
|
|
}
|
|
}
|
|
}
|
|
|
|
private func begin0(in delay: TimeAmount) {
|
|
self.eventLoop.assertInEventLoop()
|
|
guard let task = self.task else {
|
|
return
|
|
}
|
|
self.scheduled = self.eventLoop.scheduleTask(in: delay) {
|
|
task(self)
|
|
}
|
|
self.reschedule()
|
|
}
|
|
|
|
/// Try to cancel the execution of the repeated task.
|
|
///
|
|
/// Whether the execution of the task is immediately canceled depends on whether the execution of a task has already begun.
|
|
/// This means immediate cancellation is not guaranteed.
|
|
///
|
|
/// The safest way to cancel is by using the passed reference of `RepeatedTask` inside the task closure.
|
|
///
|
|
/// If the promise parameter is not `nil`, the passed promise is fulfilled when cancellation is complete.
|
|
/// Passing a promise does not prevent fulfillment of any promise provided on original task creation.
|
|
public func cancel(promise: EventLoopPromise<Void>? = nil) {
|
|
if self.eventLoop.inEventLoop {
|
|
self.cancel0(localCancellationPromise: promise)
|
|
} else {
|
|
self.eventLoop.execute {
|
|
self.cancel0(localCancellationPromise: promise)
|
|
}
|
|
}
|
|
}
|
|
|
|
private func cancel0(localCancellationPromise: EventLoopPromise<Void>?) {
|
|
self.eventLoop.assertInEventLoop()
|
|
self.scheduled?.cancel()
|
|
self.scheduled = nil
|
|
self.task = nil
|
|
|
|
// Possible states at this time are:
|
|
// 1) Task is scheduled but has not yet executed.
|
|
// 2) Task is currently executing and invoked `cancel()` on itself.
|
|
// 3) Task is currently executing and `cancel0()` has been reentrantly invoked.
|
|
// 4) NOT VALID: Task is currently executing and has NOT invoked `cancel()` (`EventLoop` guarantees serial execution)
|
|
// 5) NOT VALID: Task has completed execution in a success state (`reschedule()` ensures state #2).
|
|
// 6) Task has completed execution in a failure state.
|
|
// 7) Task has been fully cancelled at a previous time.
|
|
//
|
|
// It is desirable that the task has fully completed any execution before any cancellation promise is
|
|
// fulfilled. States 2 and 3 occur during execution, so the requirement is implemented by deferring
|
|
// fulfillment to the next `EventLoop` cycle. The delay is harmless to other states and distinguishing
|
|
// them from 2 and 3 is not practical (or necessarily possible), so is used unconditionally. Check the
|
|
// promises for nil so as not to otherwise invoke `execute()` unnecessarily.
|
|
if self.cancellationPromise != nil || localCancellationPromise != nil {
|
|
self.eventLoop.execute {
|
|
self.cancellationPromise?.succeed(())
|
|
localCancellationPromise?.succeed(())
|
|
}
|
|
}
|
|
}
|
|
|
|
private func reschedule() {
|
|
self.eventLoop.assertInEventLoop()
|
|
guard let scheduled = self.scheduled else {
|
|
return
|
|
}
|
|
|
|
scheduled.futureResult.whenSuccess { future in
|
|
future.hop(to: self.eventLoop).whenComplete { (_: Result<Void, Error>) in
|
|
self.reschedule0()
|
|
}
|
|
}
|
|
|
|
scheduled.futureResult.whenFailure { (_: Error) in
|
|
self.cancel0(localCancellationPromise: nil)
|
|
}
|
|
}
|
|
|
|
private func reschedule0() {
|
|
self.eventLoop.assertInEventLoop()
|
|
guard self.task != nil else {
|
|
return
|
|
}
|
|
self.scheduled = self.eventLoop.scheduleTask(in: self.delay) {
|
|
// we need to repeat this as we might have been cancelled in the meantime
|
|
guard let task = self.task else {
|
|
return self.eventLoop.makeSucceededFuture(())
|
|
}
|
|
return task(self)
|
|
}
|
|
self.reschedule()
|
|
}
|
|
}
|
|
|
|
extension RepeatedTask: @unchecked Sendable {}
|
|
|
|
/// An iterator over the `EventLoop`s forming an `EventLoopGroup`.
|
|
///
|
|
/// Usually returned by an `EventLoopGroup`'s `makeIterator()` method.
|
|
///
|
|
/// let group = MultiThreadedEventLoopGroup(numberOfThreads: 1)
|
|
/// group.makeIterator().forEach { loop in
|
|
/// // Do something with each loop
|
|
/// }
|
|
///
|
|
public struct EventLoopIterator: Sequence, IteratorProtocol {
|
|
public typealias Element = EventLoop
|
|
private var eventLoops: IndexingIterator<[EventLoop]>
|
|
|
|
/// Create an `EventLoopIterator` from an array of `EventLoop`s.
|
|
public init(_ eventLoops: [EventLoop]) {
|
|
self.eventLoops = eventLoops.makeIterator()
|
|
}
|
|
|
|
/// Advances to the next `EventLoop` and returns it, or `nil` if no next element exists.
|
|
///
|
|
/// - Returns: The next `EventLoop` if a next element exists; otherwise, `nil`.
|
|
public mutating func next() -> EventLoop? {
|
|
self.eventLoops.next()
|
|
}
|
|
}
|
|
|
|
extension EventLoopIterator: Sendable {}
|
|
|
|
/// An EventLoop processes IO / tasks in an endless loop for `Channel`s until it's closed.
|
|
///
|
|
/// Usually multiple `Channel`s share the same `EventLoop` for processing IO / tasks and so share the same processing `NIOThread`.
|
|
/// For a better understanding of how such an `EventLoop` works internally the following pseudo code may be helpful:
|
|
///
|
|
/// ```
|
|
/// while eventLoop.isOpen {
|
|
/// /// Block until there is something to process for 1...n Channels
|
|
/// let readyChannels = blockUntilIoOrTasksAreReady()
|
|
/// /// Loop through all the Channels
|
|
/// for channel in readyChannels {
|
|
/// /// Process IO and / or tasks for the Channel.
|
|
/// /// This may include things like:
|
|
/// /// - accept new connection
|
|
/// /// - connect to a remote host
|
|
/// /// - read from socket
|
|
/// /// - write to socket
|
|
/// /// - tasks that were submitted via EventLoop methods
|
|
/// /// and others.
|
|
/// processIoAndTasks(channel)
|
|
/// }
|
|
/// }
|
|
/// ```
|
|
///
|
|
/// Because an `EventLoop` may be shared between multiple `Channel`s it's important to _NOT_ block while processing IO / tasks. This also includes long running computations which will have the same
|
|
/// effect as blocking in this case.
|
|
public protocol EventLoop: EventLoopGroup {
|
|
/// Returns `true` if the current `NIOThread` is the same as the `NIOThread` that is tied to this `EventLoop`. `false` otherwise.
|
|
///
|
|
/// This method is intended principally as an optimization point, allowing users to write code that can provide fast-paths when this
|
|
/// property is true. It is _not suitable_ as a correctness guard. Code must be correct if this value returns `false` _even if_
|
|
/// the event loop context is actually held. That's because this property is allowed to produce false-negatives
|
|
///
|
|
/// Implementers may implement this method in a way that may produce false-negatives: that is, this value may return `false` in
|
|
/// cases where the code in question actually is executing on the event loop. It may _never_ produce false positives: this value
|
|
/// must never return `true` when event loop context is not actually held.
|
|
///
|
|
/// If it is necessary for correctness to confirm that you're on an event loop, prefer ``preconditionInEventLoop(file:line:)-7ukrq``.
|
|
var inEventLoop: Bool { get }
|
|
|
|
/// Submit a given task to be executed by the `EventLoop`
|
|
@preconcurrency
|
|
func execute(_ task: @escaping @Sendable () -> Void)
|
|
|
|
/// Submit a given task to be executed by the `EventLoop`. Once the execution is complete the returned `EventLoopFuture` is notified.
|
|
///
|
|
/// - Parameters:
|
|
/// - task: The closure that will be submitted to the `EventLoop` for execution.
|
|
/// - Returns: `EventLoopFuture` that is notified once the task was executed.
|
|
@preconcurrency
|
|
func submit<T>(_ task: @escaping @Sendable () throws -> T) -> EventLoopFuture<T>
|
|
|
|
/// The current time of the event loop.
|
|
var now: NIODeadline { get }
|
|
|
|
/// Schedule a `task` that is executed by this `EventLoop` at the given time.
|
|
///
|
|
/// - Parameters:
|
|
/// - deadline: The instant in time before which the task will not execute.
|
|
/// - task: The synchronous task to run. As with everything that runs on the `EventLoop`, it must not block.
|
|
/// - Returns: A `Scheduled` object which may be used to cancel the task if it has not yet run, or to wait
|
|
/// on the completion of the task.
|
|
///
|
|
/// - Note: You can only cancel a task before it has started executing.
|
|
@discardableResult
|
|
@preconcurrency
|
|
func scheduleTask<T>(deadline: NIODeadline, _ task: @escaping @Sendable () throws -> T) -> Scheduled<T>
|
|
|
|
/// Schedule a `task` that is executed by this `EventLoop` after the given amount of time.
|
|
///
|
|
/// - Parameters:
|
|
/// - in: The amount of time before which the task will not execute.
|
|
/// - task: The synchronous task to run. As with everything that runs on the `EventLoop`, it must not block.
|
|
/// - Returns: A `Scheduled` object which may be used to cancel the task if it has not yet run, or to wait
|
|
/// on the completion of the task.
|
|
///
|
|
/// - Note: You can only cancel a task before it has started executing.
|
|
/// - Note: The `in` value is clamped to a maximum when running on a Darwin-kernel.
|
|
@discardableResult
|
|
@preconcurrency
|
|
func scheduleTask<T>(in: TimeAmount, _ task: @escaping @Sendable () throws -> T) -> Scheduled<T>
|
|
|
|
/// Asserts that the current thread is the one tied to this `EventLoop`.
|
|
/// Otherwise, the process will be abnormally terminated as per the semantics of `preconditionFailure(_:file:line:)`.
|
|
///
|
|
/// This method may never produce false positives or false negatives in conforming implementations. It may never
|
|
/// terminate the process when event loop context is actually held, and it may never fail to terminate the process
|
|
/// when event loop context is not held.
|
|
func preconditionInEventLoop(file: StaticString, line: UInt)
|
|
|
|
/// Asserts that the current thread is _not_ the one tied to this `EventLoop`.
|
|
/// Otherwise, the process will be abnormally terminated as per the semantics of `preconditionFailure(_:file:line:)`.
|
|
///
|
|
/// This method may never produce false positives or false negatives in conforming implementations. It may never
|
|
/// terminate the process when event loop context is not held, and it may never fail to terminate the process
|
|
/// when event loop context is held.
|
|
func preconditionNotInEventLoop(file: StaticString, line: UInt)
|
|
|
|
/// Return a succeeded `Void` future.
|
|
///
|
|
/// Semantically, this function is equivalent to calling `makeSucceededFuture(())`.
|
|
/// Contrary to `makeSucceededFuture`, `makeSucceededVoidFuture` is a customization point for `EventLoop`s which
|
|
/// allows `EventLoop`s to cache a pre-succeeded `Void` future to prevent superfluous allocations.
|
|
func makeSucceededVoidFuture() -> EventLoopFuture<Void>
|
|
|
|
/// Returns a `SerialExecutor` corresponding to this ``EventLoop``.
|
|
///
|
|
/// This executor can be used to isolate an actor to a given ``EventLoop``. Implementers are encouraged to customise
|
|
/// this implementation by conforming their ``EventLoop`` to ``NIOSerialEventLoopExecutor`` which will provide an
|
|
/// optimised implementation of this method, and will conform their type to `SerialExecutor`. The default
|
|
/// implementation provides suboptimal performance.
|
|
@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
|
|
var executor: any SerialExecutor { get }
|
|
|
|
/// Submit a job to be executed by the `EventLoop`
|
|
@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
|
|
func enqueue(_ job: consuming ExecutorJob)
|
|
|
|
/// Must crash if it is not safe to call `wait()` on an `EventLoopFuture`.
|
|
///
|
|
/// This method is a debugging hook that can be used to override the behaviour of `EventLoopFuture.wait()` when called.
|
|
/// By default this simply becomes `preconditionNotInEventLoop`, but some `EventLoop`s are capable of more exhaustive
|
|
/// checking and can validate that the wait is not occurring on an entire `EventLoopGroup`, or even more broadly.
|
|
///
|
|
/// This method should not be called by users directly, it should only be implemented by `EventLoop` implementers that
|
|
/// need to customise the behaviour.
|
|
func _preconditionSafeToWait(file: StaticString, line: UInt)
|
|
|
|
/// Debug hook: track a promise creation and its location.
|
|
///
|
|
/// This debug hook is called by EventLoopFutures and EventLoopPromises when they are created, and tracks the location
|
|
/// of their creation. It combines with `_promiseCompleted` to provide high-fidelity diagnostics for debugging leaked
|
|
/// promises.
|
|
///
|
|
/// In release mode, this function will never be called.
|
|
///
|
|
/// It is valid for an `EventLoop` not to implement any of the two `_promise` functions. If either of them are implemented,
|
|
/// however, both of them should be implemented.
|
|
func _promiseCreated(futureIdentifier: _NIOEventLoopFutureIdentifier, file: StaticString, line: UInt)
|
|
|
|
/// Debug hook: complete a specific promise and return its creation location.
|
|
///
|
|
/// This debug hook is called by EventLoopFutures and EventLoopPromises when they are deinited, and removes the data from
|
|
/// the promise tracking map and, if available, provides that data as its return value. It combines with `_promiseCreated`
|
|
/// to provide high-fidelity diagnostics for debugging leaked promises.
|
|
///
|
|
/// In release mode, this function will never be called.
|
|
///
|
|
/// It is valid for an `EventLoop` not to implement any of the two `_promise` functions. If either of them are implemented,
|
|
/// however, both of them should be implemented.
|
|
func _promiseCompleted(futureIdentifier: _NIOEventLoopFutureIdentifier) -> (file: StaticString, line: UInt)?
|
|
|
|
/// Schedule a callback at a given time.
|
|
///
|
|
/// - Parameters:
|
|
/// - deadline: The instant in time before which the task will not execute.
|
|
/// - handler: The handler that defines the behavior of the callback when executed or canceled.
|
|
/// - Returns: A ``NIOScheduledCallback`` that can be used to cancel the scheduled callback.
|
|
///
|
|
/// - NOTE: Event loops that provide a custom scheduled callback implementation **must** also implement
|
|
/// `cancelScheduledCallback`. Failure to do so will result in a runtime error.
|
|
@preconcurrency
|
|
@discardableResult
|
|
func scheduleCallback(
|
|
at deadline: NIODeadline,
|
|
handler: some (NIOScheduledCallbackHandler & Sendable)
|
|
) throws -> NIOScheduledCallback
|
|
|
|
/// Schedule a callback after given time.
|
|
///
|
|
/// - Parameters:
|
|
/// - amount: The amount of time before which the task will not execute.
|
|
/// - handler: The handler that defines the behavior of the callback when executed or canceled.
|
|
/// - Returns: A ``NIOScheduledCallback`` that can be used to cancel the scheduled callback.
|
|
///
|
|
/// - NOTE: Event loops that provide a custom scheduled callback implementation **must** also implement
|
|
/// `cancelScheduledCallback`. Failure to do so will result in a runtime error.
|
|
@preconcurrency
|
|
@discardableResult
|
|
func scheduleCallback(
|
|
in amount: TimeAmount,
|
|
handler: some (NIOScheduledCallbackHandler & Sendable)
|
|
) throws -> NIOScheduledCallback
|
|
|
|
/// Cancel a scheduled callback.
|
|
///
|
|
/// - NOTE: Event loops only need to implemented this if they provide a custom scheduled callback implementation.
|
|
func cancelScheduledCallback(_ scheduledCallback: NIOScheduledCallback)
|
|
|
|
/// Submit a given task to be executed by the ``EventLoop``, from a context where the caller
|
|
/// statically knows that the context is isolated.
|
|
///
|
|
/// This is an optional performance hook. ``EventLoop`` implementers are not required to implement
|
|
/// this witness, but may choose to do so to enable better performance of the isolated EL views. If
|
|
/// they do so, ``EventLoop/Isolated/execute`` will perform better.
|
|
func _executeIsolatedUnsafeUnchecked(_ task: @escaping () -> Void)
|
|
|
|
/// Submit a given task to be executed by the ``EventLoop```, from a context where the caller
|
|
/// statically knows that the context is isolated.
|
|
///
|
|
/// Once the execution is complete the returned ``EventLoopFuture`` is notified.
|
|
///
|
|
/// This is an optional performance hook. ``EventLoop`` implementers are not required to implement
|
|
/// this witness, but may choose to do so to enable better performance of the isolated EL views. If
|
|
/// they do so, ``EventLoop/Isolated/submit`` will perform better.
|
|
///
|
|
/// - Parameters:
|
|
/// - task: The closure that will be submitted to the ``EventLoop`` for execution.
|
|
/// - Returns: ``EventLoopFuture`` that is notified once the task was executed.
|
|
func _submitIsolatedUnsafeUnchecked<T>(_ task: @escaping () throws -> T) -> EventLoopFuture<T>
|
|
|
|
/// Schedule a `task` that is executed by this ``EventLoop`` at the given time, from a context where the caller
|
|
/// statically knows that the context is isolated.
|
|
///
|
|
/// This is an optional performance hook. ``EventLoop`` implementers are not required to implement
|
|
/// this witness, but may choose to do so to enable better performance of the isolated EL views. If
|
|
/// they do so, ``EventLoop/Isolated/scheduleTask(deadline:_:)`` will perform better.
|
|
///
|
|
/// - Parameters:
|
|
/// - deadline: The instant in time before which the task will not execute.
|
|
/// - task: The synchronous task to run. As with everything that runs on the ``EventLoop```, it must not block.
|
|
/// - Returns: A ``Scheduled``` object which may be used to cancel the task if it has not yet run, or to wait
|
|
/// on the completion of the task.
|
|
///
|
|
/// - Note: You can only cancel a task before it has started executing.
|
|
@discardableResult
|
|
func _scheduleTaskIsolatedUnsafeUnchecked<T>(
|
|
deadline: NIODeadline,
|
|
_ task: @escaping () throws -> T
|
|
) -> Scheduled<T>
|
|
|
|
/// Schedule a `task` that is executed by this ``EventLoop`` after the given amount of time, from a context where the caller
|
|
/// statically knows that the context is isolated.
|
|
///
|
|
/// This is an optional performance hook. ``EventLoop`` implementers are not required to implement
|
|
/// this witness, but may choose to do so to enable better performance of the isolated EL views. If
|
|
/// they do so, ``EventLoop/Isolated/scheduleTask(in:_:)`` will perform better.
|
|
///
|
|
/// - Parameters:
|
|
/// - in: The amount of time before which the task will not execute.
|
|
/// - task: The synchronous task to run. As with everything that runs on the ``EventLoop``, it must not block.
|
|
/// - Returns: A ``Scheduled`` object which may be used to cancel the task if it has not yet run, or to wait
|
|
/// on the completion of the task.
|
|
///
|
|
/// - Note: You can only cancel a task before it has started executing.
|
|
/// - Note: The `in` value is clamped to a maximum when running on a Darwin-kernel.
|
|
@discardableResult
|
|
func _scheduleTaskIsolatedUnsafeUnchecked<T>(
|
|
in: TimeAmount,
|
|
_ task: @escaping () throws -> T
|
|
) -> Scheduled<T>
|
|
|
|
/// Schedule a callback that is executed by this ``EventLoop`` at a given time, from a context where the caller
|
|
/// statically knows that the context is isolated.
|
|
///
|
|
/// This is an optional performance hook. ``EventLoop`` implementers are not required to implement
|
|
/// this witness, but may choose to do so to enable better performance of the isolated EL views. If
|
|
/// they do so, ``EventLoop/Isolated/scheduleCallback(at:_:)`` will perform better.
|
|
///
|
|
/// - Parameters:
|
|
/// - at: The instant in time before which the task will not execute.
|
|
/// - handler: The handler that defines the behavior of the callback when executed or canceled.
|
|
/// - Returns: A ``NIOScheduledCallback`` that can be used to cancel the scheduled callback.
|
|
@discardableResult
|
|
func _scheduleCallbackIsolatedUnsafeUnchecked(
|
|
at deadline: NIODeadline,
|
|
handler: some NIOScheduledCallbackHandler
|
|
) throws -> NIOScheduledCallback
|
|
|
|
/// Schedule a callback that is executed by this ``EventLoop`` after a given time, from a context where the caller
|
|
/// statically knows that the context is isolated.
|
|
///
|
|
/// This is an optional performance hook. ``EventLoop`` implementers are not required to implement
|
|
/// this witness, but may choose to do so to enable better performance of the isolated EL views. If
|
|
/// they do so, ``EventLoop/Isolated/scheduleCallback(in:_:)`` will perform better.
|
|
///
|
|
/// - Parameters:
|
|
/// - in: The amount of time before which the task will not execute.
|
|
/// - handler: The handler that defines the behavior of the callback when executed or canceled.
|
|
/// - Returns: A ``NIOScheduledCallback`` that can be used to cancel the scheduled callback.
|
|
@discardableResult
|
|
func _scheduleCallbackIsolatedUnsafeUnchecked(
|
|
in amount: TimeAmount,
|
|
handler: some NIOScheduledCallbackHandler
|
|
) throws -> NIOScheduledCallback
|
|
}
|
|
|
|
extension EventLoop {
|
|
/// Default implementation of `now`: Returns `NIODeadline.now()`.
|
|
public var now: NIODeadline { .now() }
|
|
}
|
|
|
|
extension EventLoop {
|
|
/// Default implementation of `makeSucceededVoidFuture`: Return a fresh future (which will allocate).
|
|
public func makeSucceededVoidFuture() -> EventLoopFuture<Void> {
|
|
EventLoopFuture(eventLoop: self, value: ())
|
|
}
|
|
|
|
public func _preconditionSafeToWait(file: StaticString, line: UInt) {
|
|
self.preconditionNotInEventLoop(file: file, line: line)
|
|
}
|
|
|
|
/// Default implementation of `_promiseCreated`: does nothing.
|
|
public func _promiseCreated(futureIdentifier: _NIOEventLoopFutureIdentifier, file: StaticString, line: UInt) {
|
|
return
|
|
}
|
|
|
|
/// Default implementation of `_promiseCompleted`: does nothing.
|
|
public func _promiseCompleted(futureIdentifier: _NIOEventLoopFutureIdentifier) -> (file: StaticString, line: UInt)?
|
|
{
|
|
nil
|
|
}
|
|
|
|
/// Default implementation: wraps the task in an UnsafeTransfer.
|
|
@inlinable
|
|
public func _executeIsolatedUnsafeUnchecked(_ task: @escaping () -> Void) {
|
|
self.assertInEventLoop()
|
|
let unsafeTransfer = UnsafeTransfer(task)
|
|
self.execute {
|
|
unsafeTransfer.wrappedValue()
|
|
}
|
|
}
|
|
|
|
/// Default implementation: wraps the task in an UnsafeTransfer.
|
|
@inlinable
|
|
public func _submitIsolatedUnsafeUnchecked<T>(_ task: @escaping () throws -> T) -> EventLoopFuture<T> {
|
|
self.assertInEventLoop()
|
|
let unsafeTransfer = UnsafeTransfer(task)
|
|
return self.submit {
|
|
try unsafeTransfer.wrappedValue()
|
|
}
|
|
}
|
|
|
|
/// Default implementation: wraps the task in an UnsafeTransfer.
|
|
@inlinable
|
|
@discardableResult
|
|
public func _scheduleTaskIsolatedUnsafeUnchecked<T>(
|
|
deadline: NIODeadline,
|
|
_ task: @escaping () throws -> T
|
|
) -> Scheduled<T> {
|
|
self.assertInEventLoop()
|
|
let unsafeTransfer = UnsafeTransfer(task)
|
|
return self.scheduleTask(deadline: deadline) {
|
|
try unsafeTransfer.wrappedValue()
|
|
}
|
|
}
|
|
|
|
/// Default implementation: wraps the task in an UnsafeTransfer.
|
|
@inlinable
|
|
@discardableResult
|
|
public func _scheduleTaskIsolatedUnsafeUnchecked<T>(
|
|
in delay: TimeAmount,
|
|
_ task: @escaping () throws -> T
|
|
) -> Scheduled<T> {
|
|
self.assertInEventLoop()
|
|
let unsafeTransfer = UnsafeTransfer(task)
|
|
return self.scheduleTask(in: delay) {
|
|
try unsafeTransfer.wrappedValue()
|
|
}
|
|
}
|
|
|
|
@inlinable
|
|
@discardableResult
|
|
public func _scheduleCallbackIsolatedUnsafeUnchecked(
|
|
at deadline: NIODeadline,
|
|
handler: some NIOScheduledCallbackHandler
|
|
) throws -> NIOScheduledCallback {
|
|
let unsafeHandlerWrapper = LoopBoundScheduledCallbackHandlerWrapper(wrapping: handler, eventLoop: self)
|
|
return try self.scheduleCallback(at: deadline, handler: unsafeHandlerWrapper)
|
|
}
|
|
|
|
@inlinable
|
|
@discardableResult
|
|
public func _scheduleCallbackIsolatedUnsafeUnchecked(
|
|
in amount: TimeAmount,
|
|
handler: some NIOScheduledCallbackHandler
|
|
) throws -> NIOScheduledCallback {
|
|
let unsafeHandlerWrapper = LoopBoundScheduledCallbackHandlerWrapper(wrapping: handler, eventLoop: self)
|
|
return try self.scheduleCallback(in: amount, handler: unsafeHandlerWrapper)
|
|
}
|
|
}
|
|
|
|
extension EventLoop {
|
|
@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
|
|
public var executor: any SerialExecutor {
|
|
NIODefaultSerialEventLoopExecutor(self)
|
|
}
|
|
|
|
@inlinable
|
|
@available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *)
|
|
public func enqueue(_ job: consuming ExecutorJob) {
|
|
// By default we are just going to use execute to run the job
|
|
// this is quite heavy since it allocates the closure for
|
|
// every single job.
|
|
let unownedJob = UnownedJob(job)
|
|
self.execute {
|
|
unownedJob.runSynchronously(on: self.executor.asUnownedSerialExecutor())
|
|
}
|
|
}
|
|
}
|
|
|
|
extension EventLoopGroup {
|
|
public var description: String {
|
|
String(describing: self)
|
|
}
|
|
}
|
|
|
|
/// Represents a time _interval_.
|
|
///
|
|
/// - Note: `TimeAmount` should not be used to represent a point in time.
|
|
public struct TimeAmount: Hashable, Sendable {
|
|
@available(*, deprecated, message: "This typealias doesn't serve any purpose. Please use Int64 directly.")
|
|
public typealias Value = Int64
|
|
|
|
/// The nanoseconds representation of the `TimeAmount`.
|
|
public let nanoseconds: Int64
|
|
|
|
@inlinable
|
|
init(_ nanoseconds: Int64) {
|
|
self.nanoseconds = nanoseconds
|
|
}
|
|
|
|
/// Creates a new `TimeAmount` for the given amount of nanoseconds.
|
|
///
|
|
/// - Parameters:
|
|
/// - amount: the amount of nanoseconds this `TimeAmount` represents.
|
|
/// - Returns: the `TimeAmount` for the given amount.
|
|
@inlinable
|
|
public static func nanoseconds(_ amount: Int64) -> TimeAmount {
|
|
TimeAmount(amount)
|
|
}
|
|
|
|
/// Creates a new `TimeAmount` for the given amount of microseconds.
|
|
///
|
|
/// - Parameters:
|
|
/// - amount: the amount of microseconds this `TimeAmount` represents.
|
|
/// - Returns: the `TimeAmount` for the given amount.
|
|
///
|
|
/// - Note: returns `TimeAmount(.max)` if the amount overflows when converted to nanoseconds and `TimeAmount(.min)` if it underflows.
|
|
@inlinable
|
|
public static func microseconds(_ amount: Int64) -> TimeAmount {
|
|
TimeAmount(_cappedNanoseconds(amount: amount, multiplier: 1000))
|
|
}
|
|
|
|
/// Creates a new `TimeAmount` for the given amount of milliseconds.
|
|
///
|
|
/// - Parameters:
|
|
/// - amount: the amount of milliseconds this `TimeAmount` represents.
|
|
/// - Returns: the `TimeAmount` for the given amount.
|
|
///
|
|
/// - Note: returns `TimeAmount(.max)` if the amount overflows when converted to nanoseconds and `TimeAmount(.min)` if it underflows.
|
|
@inlinable
|
|
public static func milliseconds(_ amount: Int64) -> TimeAmount {
|
|
TimeAmount(_cappedNanoseconds(amount: amount, multiplier: 1000 * 1000))
|
|
}
|
|
|
|
/// Creates a new `TimeAmount` for the given amount of seconds.
|
|
///
|
|
/// - Parameters:
|
|
/// - amount: the amount of seconds this `TimeAmount` represents.
|
|
/// - Returns: the `TimeAmount` for the given amount.
|
|
///
|
|
/// - Note: returns `TimeAmount(.max)` if the amount overflows when converted to nanoseconds and `TimeAmount(.min)` if it underflows.
|
|
@inlinable
|
|
public static func seconds(_ amount: Int64) -> TimeAmount {
|
|
TimeAmount(_cappedNanoseconds(amount: amount, multiplier: 1000 * 1000 * 1000))
|
|
}
|
|
|
|
/// Creates a new `TimeAmount` for the given amount of minutes.
|
|
///
|
|
/// - Parameters:
|
|
/// - amount: the amount of minutes this `TimeAmount` represents.
|
|
/// - Returns: the `TimeAmount` for the given amount.
|
|
///
|
|
/// - Note: returns `TimeAmount(.max)` if the amount overflows when converted to nanoseconds and `TimeAmount(.min)` if it underflows.
|
|
@inlinable
|
|
public static func minutes(_ amount: Int64) -> TimeAmount {
|
|
TimeAmount(_cappedNanoseconds(amount: amount, multiplier: 1000 * 1000 * 1000 * 60))
|
|
}
|
|
|
|
/// Creates a new `TimeAmount` for the given amount of hours.
|
|
///
|
|
/// - Parameters:
|
|
/// - amount: the amount of hours this `TimeAmount` represents.
|
|
/// - Returns: the `TimeAmount` for the given amount.
|
|
///
|
|
/// - Note: returns `TimeAmount(.max)` if the amount overflows when converted to nanoseconds and `TimeAmount(.min)` if it underflows.
|
|
@inlinable
|
|
public static func hours(_ amount: Int64) -> TimeAmount {
|
|
TimeAmount(_cappedNanoseconds(amount: amount, multiplier: 1000 * 1000 * 1000 * 60 * 60))
|
|
}
|
|
|
|
/// Converts `amount` to nanoseconds multiplying it by `multiplier`. The return value is capped to `Int64.max` if the multiplication overflows and `Int64.min` if it underflows.
|
|
///
|
|
/// - parameters:
|
|
/// - amount: the amount to be converted to nanoseconds.
|
|
/// - multiplier: the multiplier that converts the given amount to nanoseconds.
|
|
/// - returns: the amount converted to nanoseconds within [Int64.min, Int64.max].
|
|
@inlinable
|
|
static func _cappedNanoseconds(amount: Int64, multiplier: Int64) -> Int64 {
|
|
let nanosecondsMultiplication = amount.multipliedReportingOverflow(by: multiplier)
|
|
if nanosecondsMultiplication.overflow {
|
|
return amount >= 0 ? .max : .min
|
|
} else {
|
|
return nanosecondsMultiplication.partialValue
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Contains the logic for parsing time amounts from strings,
|
|
/// and printing pretty strings to represent time amounts.
|
|
extension TimeAmount: CustomStringConvertible {
|
|
|
|
/// Errors thrown when parsint a TimeAmount from a string
|
|
internal enum ValidationError: Error, Equatable {
|
|
/// Can't parse the provided unit
|
|
case unsupportedUnit(String)
|
|
|
|
/// Can't parse the number into a Double
|
|
case invalidNumber(String)
|
|
}
|
|
|
|
/// Creates a TimeAmount from a string representation with an optional default unit.
|
|
///
|
|
/// Supports formats like:
|
|
/// - "5s" (5 seconds)
|
|
/// - "100ms" (100 milliseconds)
|
|
/// - "42" (42 of default unit)
|
|
/// - "1 hr" (1 hour)
|
|
///
|
|
/// This function only supports one pair of the number and units, i.e. "5s" or "100ms" but not "5s 100ms".
|
|
///
|
|
/// Supported units:
|
|
/// - h, hr, hrs (hours)
|
|
/// - m, min (minutes)
|
|
/// - s, sec, secs (seconds)
|
|
/// - ms, millis (milliseconds)
|
|
/// - us, µs, micros (microseconds)
|
|
/// - ns, nanos (nanoseconds)
|
|
///
|
|
/// - Parameters:
|
|
/// - userProvidedString: The string to parse
|
|
///
|
|
/// - Throws: ValidationError if the string cannot be parsed
|
|
public init(_ userProvidedString: String) throws {
|
|
let lowercased = String(userProvidedString.filter { !$0.isWhitespace }).lowercased()
|
|
let parsedNumbers = lowercased.prefix(while: { $0.isWholeNumber || $0 == "," || $0 == "." })
|
|
let parsedUnit = String(lowercased.dropFirst(parsedNumbers.count))
|
|
|
|
guard let numbers = Int64(parsedNumbers) else {
|
|
throw ValidationError.invalidNumber("'\(userProvidedString)' cannot be parsed as number and unit")
|
|
}
|
|
|
|
switch parsedUnit {
|
|
case "h", "hr", "hrs":
|
|
self = .hours(numbers)
|
|
case "m", "min":
|
|
self = .minutes(numbers)
|
|
case "s", "sec", "secs":
|
|
self = .seconds(numbers)
|
|
case "ms", "millis":
|
|
self = .milliseconds(numbers)
|
|
case "us", "µs", "micros":
|
|
self = .microseconds(numbers)
|
|
case "ns", "nanos":
|
|
self = .nanoseconds(numbers)
|
|
default:
|
|
throw ValidationError.unsupportedUnit("Unknown unit '\(parsedUnit)' in '\(userProvidedString)'")
|
|
}
|
|
}
|
|
|
|
/// Returns a human-readable string representation of the time amount
|
|
/// using the most appropriate unit
|
|
public var description: String {
|
|
let fullNS = self.nanoseconds
|
|
let (fullUS, remUS) = fullNS.quotientAndRemainder(dividingBy: 1_000)
|
|
let (fullMS, remMS) = fullNS.quotientAndRemainder(dividingBy: 1_000_000)
|
|
let (fullS, remS) = fullNS.quotientAndRemainder(dividingBy: 1_000_000_000)
|
|
|
|
if remS == 0 {
|
|
return "\(fullS) s"
|
|
} else if remMS == 0 {
|
|
return "\(fullMS) ms"
|
|
} else if remUS == 0 {
|
|
return "\(fullUS) us"
|
|
} else {
|
|
return "\(fullNS) ns"
|
|
}
|
|
}
|
|
}
|
|
|
|
extension TimeAmount: Comparable {
|
|
@inlinable
|
|
public static func < (lhs: TimeAmount, rhs: TimeAmount) -> Bool {
|
|
lhs.nanoseconds < rhs.nanoseconds
|
|
}
|
|
}
|
|
|
|
extension TimeAmount: AdditiveArithmetic {
|
|
/// The zero value for `TimeAmount`.
|
|
@inlinable
|
|
public static var zero: TimeAmount {
|
|
TimeAmount.nanoseconds(0)
|
|
}
|
|
|
|
@inlinable
|
|
public static func + (lhs: TimeAmount, rhs: TimeAmount) -> TimeAmount {
|
|
TimeAmount(lhs.nanoseconds + rhs.nanoseconds)
|
|
}
|
|
|
|
@inlinable
|
|
public static func += (lhs: inout TimeAmount, rhs: TimeAmount) {
|
|
lhs = lhs + rhs
|
|
}
|
|
|
|
@inlinable
|
|
public static func - (lhs: TimeAmount, rhs: TimeAmount) -> TimeAmount {
|
|
TimeAmount(lhs.nanoseconds - rhs.nanoseconds)
|
|
}
|
|
|
|
@inlinable
|
|
public static func -= (lhs: inout TimeAmount, rhs: TimeAmount) {
|
|
lhs = lhs - rhs
|
|
}
|
|
|
|
@inlinable
|
|
public static func * <T: BinaryInteger>(lhs: T, rhs: TimeAmount) -> TimeAmount {
|
|
TimeAmount(Int64(lhs) * rhs.nanoseconds)
|
|
}
|
|
|
|
@inlinable
|
|
public static func * <T: BinaryInteger>(lhs: TimeAmount, rhs: T) -> TimeAmount {
|
|
TimeAmount(lhs.nanoseconds * Int64(rhs))
|
|
}
|
|
}
|
|
|
|
/// Represents a point in time.
|
|
///
|
|
/// Stores the time in nanoseconds as returned by `DispatchTime.now().uptimeNanoseconds`
|
|
///
|
|
/// `NIODeadline` allow chaining multiple tasks with the same deadline without needing to
|
|
/// compute new timeouts for each step
|
|
///
|
|
/// ```
|
|
/// func doSomething(deadline: NIODeadline) -> EventLoopFuture<Void> {
|
|
/// return step1(deadline: deadline).flatMap {
|
|
/// step2(deadline: deadline)
|
|
/// }
|
|
/// }
|
|
/// doSomething(deadline: .now() + .seconds(5))
|
|
/// ```
|
|
///
|
|
/// - Note: `NIODeadline` should not be used to represent a time interval
|
|
public struct NIODeadline: Equatable, Hashable, Sendable {
|
|
@available(*, deprecated, message: "This typealias doesn't serve any purpose, please use UInt64 directly.")
|
|
public typealias Value = UInt64
|
|
|
|
// This really should be an UInt63 but we model it as Int64 with >=0 assert
|
|
@usableFromInline var _uptimeNanoseconds: Int64 {
|
|
didSet {
|
|
assert(self._uptimeNanoseconds >= 0)
|
|
}
|
|
}
|
|
|
|
/// The nanoseconds since boot representation of the `NIODeadline`.
|
|
@inlinable
|
|
public var uptimeNanoseconds: UInt64 {
|
|
.init(self._uptimeNanoseconds)
|
|
}
|
|
|
|
@inlinable
|
|
public static var distantPast: NIODeadline { NIODeadline(0) }
|
|
|
|
@inlinable
|
|
public static var distantFuture: NIODeadline { NIODeadline(.init(Int64.max)) }
|
|
|
|
@inlinable init(_ nanoseconds: Int64) {
|
|
precondition(nanoseconds >= 0)
|
|
self._uptimeNanoseconds = nanoseconds
|
|
}
|
|
|
|
/// Getting the time is a very common operation so it warrants optimization.
|
|
///
|
|
/// Prior to this function, NIO relied on `DispatchTime.now()`, on all platforms. In addition to
|
|
/// the overhead of making a library call, the underlying implementation has a lot of branches
|
|
/// because `libdispatch` supports many more usecases than we are making use of here.
|
|
///
|
|
/// On Linux, `DispachTime.now()` _always_ results in a simple call to `clock_gettime(3)` and so
|
|
/// we make that call here, directly from NIO.
|
|
///
|
|
/// - TODO: Investigate optimizing the call to `DispatchTime.now()` away on other platforms too.
|
|
@inlinable
|
|
static func timeNow() -> UInt64 {
|
|
#if os(Linux)
|
|
var ts = timespec()
|
|
clock_gettime(CLOCK_MONOTONIC, &ts)
|
|
/// We use unsafe arithmetic here because `UInt64.max` nanoseconds is more than 580 years,
|
|
/// and the odds that this code will still be running 530 years from now is very, very low,
|
|
/// so as a practical matter this will never overflow.
|
|
return UInt64(ts.tv_sec) &* 1_000_000_000 &+ UInt64(ts.tv_nsec)
|
|
#elseif os(WASI)
|
|
var ts = timespec()
|
|
CNIOWASI_gettime(&ts)
|
|
/// We use unsafe arithmetic here because `UInt64.max` nanoseconds is more than 580 years,
|
|
/// and the odds that this code will still be running 530 years from now is very, very low,
|
|
/// so as a practical matter this will never overflow.
|
|
return UInt64(ts.tv_sec) &* 1_000_000_000 &+ UInt64(ts.tv_nsec)
|
|
#else
|
|
return DispatchTime.now().uptimeNanoseconds
|
|
#endif // os(Linux)
|
|
}
|
|
|
|
@inlinable
|
|
public static func now() -> NIODeadline {
|
|
NIODeadline.uptimeNanoseconds(timeNow())
|
|
}
|
|
|
|
@inlinable
|
|
public static func uptimeNanoseconds(_ nanoseconds: UInt64) -> NIODeadline {
|
|
NIODeadline(Int64(min(UInt64(Int64.max), nanoseconds)))
|
|
}
|
|
|
|
@inlinable
|
|
public static func == (lhs: NIODeadline, rhs: NIODeadline) -> Bool {
|
|
lhs.uptimeNanoseconds == rhs.uptimeNanoseconds
|
|
}
|
|
|
|
@inlinable
|
|
public func hash(into hasher: inout Hasher) {
|
|
hasher.combine(self.uptimeNanoseconds)
|
|
}
|
|
}
|
|
|
|
extension NIODeadline: Comparable {
|
|
@inlinable
|
|
public static func < (lhs: NIODeadline, rhs: NIODeadline) -> Bool {
|
|
lhs.uptimeNanoseconds < rhs.uptimeNanoseconds
|
|
}
|
|
|
|
@inlinable
|
|
public static func > (lhs: NIODeadline, rhs: NIODeadline) -> Bool {
|
|
lhs.uptimeNanoseconds > rhs.uptimeNanoseconds
|
|
}
|
|
}
|
|
|
|
extension NIODeadline: CustomStringConvertible {
|
|
@inlinable
|
|
public var description: String {
|
|
self.uptimeNanoseconds.description
|
|
}
|
|
}
|
|
|
|
extension NIODeadline {
|
|
@inlinable
|
|
public static func - (lhs: NIODeadline, rhs: NIODeadline) -> TimeAmount {
|
|
// This won't ever crash, NIODeadlines are guaranteed to be within 0 ..< 2^63-1 nanoseconds so the result can
|
|
// definitely be stored in a TimeAmount (which is an Int64).
|
|
.nanoseconds(Int64(lhs.uptimeNanoseconds) - Int64(rhs.uptimeNanoseconds))
|
|
}
|
|
|
|
@inlinable
|
|
public static func + (lhs: NIODeadline, rhs: TimeAmount) -> NIODeadline {
|
|
let partial: Int64
|
|
let overflow: Bool
|
|
(partial, overflow) = Int64(lhs.uptimeNanoseconds).addingReportingOverflow(rhs.nanoseconds)
|
|
if overflow {
|
|
assert(rhs.nanoseconds > 0) // this certainly must have overflowed towards +infinity
|
|
return NIODeadline.distantFuture
|
|
}
|
|
guard partial >= 0 else {
|
|
return NIODeadline.uptimeNanoseconds(0)
|
|
}
|
|
return NIODeadline(partial)
|
|
}
|
|
|
|
@inlinable
|
|
public static func - (lhs: NIODeadline, rhs: TimeAmount) -> NIODeadline {
|
|
if rhs.nanoseconds < 0 {
|
|
// The addition won't crash because the worst that could happen is `UInt64(Int64.max) + UInt64(Int64.max)`
|
|
// which fits into an UInt64 (and will then be capped to Int64.max == distantFuture by `uptimeNanoseconds`).
|
|
return NIODeadline.uptimeNanoseconds(lhs.uptimeNanoseconds + rhs.nanoseconds.magnitude)
|
|
} else if rhs.nanoseconds > lhs.uptimeNanoseconds {
|
|
// Cap it at `0` because otherwise this would be negative.
|
|
return NIODeadline.init(0)
|
|
} else {
|
|
// This will be positive but still fix in an Int64.
|
|
let result = Int64(lhs.uptimeNanoseconds) - rhs.nanoseconds
|
|
assert(result >= 0)
|
|
return NIODeadline(result)
|
|
}
|
|
}
|
|
}
|
|
|
|
extension EventLoop {
|
|
/// Submit `task` to be run on this `EventLoop`.
|
|
///
|
|
/// The returned `EventLoopFuture` will be completed when `task` has finished running. It will be succeeded with
|
|
/// `task`'s return value or failed if the execution of `task` threw an error.
|
|
///
|
|
/// - Parameters:
|
|
/// - task: The synchronous task to run. As everything that runs on the `EventLoop`, it must not block.
|
|
/// - Returns: An `EventLoopFuture` containing the result of `task`'s execution.
|
|
@inlinable
|
|
@preconcurrency
|
|
public func submit<T>(_ task: @escaping @Sendable () throws -> T) -> EventLoopFuture<T> {
|
|
let promise: EventLoopPromise<T> = makePromise(file: #fileID, line: #line)
|
|
|
|
self.execute {
|
|
do {
|
|
// UnsafeUnchecked is allowed here because we know we are on the EL.
|
|
promise.assumeIsolatedUnsafeUnchecked().succeed(try task())
|
|
} catch let err {
|
|
promise.fail(err)
|
|
}
|
|
}
|
|
|
|
return promise.futureResult
|
|
}
|
|
|
|
/// Submit `task` to be run on this `EventLoop`.
|
|
///
|
|
/// The returned `EventLoopFuture` will be completed when `task` has finished running. It will be identical to
|
|
/// the `EventLoopFuture` returned by `task`.
|
|
///
|
|
/// - Parameters:
|
|
/// - task: The asynchronous task to run. As with everything that runs on the `EventLoop`, it must not block.
|
|
/// - Returns: An `EventLoopFuture` identical to the `EventLoopFuture` returned from `task`.
|
|
@inlinable
|
|
@preconcurrency
|
|
public func flatSubmit<T: Sendable>(_ task: @escaping @Sendable () -> EventLoopFuture<T>) -> EventLoopFuture<T> {
|
|
self.submit(task).flatMap { $0 }
|
|
}
|
|
|
|
/// Schedule a `task` that is executed by this `EventLoop` at the given time.
|
|
///
|
|
/// - Note: The `T` must be `Sendable` since the isolation domains of the event loop future returned from `task` and
|
|
/// this event loop might differ.
|
|
///
|
|
/// - Parameters:
|
|
/// - deadline: The instant in time before which the task will not execute.
|
|
/// - file: The file this function was called in, for debugging purposes.
|
|
/// - line: The line this function was called on, for debugging purposes.
|
|
/// - task: The asynchronous task to run. As with everything that runs on the `EventLoop`, it must not block.
|
|
/// - Returns: A `Scheduled` object which may be used to cancel the task if it has not yet run, or to wait
|
|
/// on the full execution of the task, including its returned `EventLoopFuture`.
|
|
///
|
|
/// - Note: You can only cancel a task before it has started executing.
|
|
@discardableResult
|
|
@inlinable
|
|
@preconcurrency
|
|
public func flatScheduleTask<T: Sendable>(
|
|
deadline: NIODeadline,
|
|
file: StaticString = #fileID,
|
|
line: UInt = #line,
|
|
_ task: @escaping @Sendable () throws -> EventLoopFuture<T>
|
|
) -> Scheduled<T> {
|
|
let promise: EventLoopPromise<T> = self.makePromise(file: file, line: line)
|
|
let scheduled = self.scheduleTask(deadline: deadline, task)
|
|
|
|
scheduled.futureResult.whenComplete { result in
|
|
switch result {
|
|
case .success(let futureResult):
|
|
promise.completeWith(futureResult)
|
|
case .failure(let error):
|
|
promise.fail(error)
|
|
}
|
|
}
|
|
|
|
return .init(promise: promise, cancellationTask: { scheduled.cancel() })
|
|
}
|
|
|
|
/// Schedule a `task` that is executed by this `EventLoop` after the given amount of time.
|
|
///
|
|
/// - Note: The `T` must be `Sendable` since the isolation domains of the event loop future returned from `task` and
|
|
/// this event loop might differ.
|
|
///
|
|
/// - Parameters:
|
|
/// - delay: The amount of time before which the task will not execute.
|
|
/// - file: The file this function was called in, for debugging purposes.
|
|
/// - line: The line this function was called on, for debugging purposes.
|
|
/// - task: The asynchronous task to run. As everything that runs on the `EventLoop`, it must not block.
|
|
/// - Returns: A `Scheduled` object which may be used to cancel the task if it has not yet run, or to wait
|
|
/// on the full execution of the task, including its returned `EventLoopFuture`.
|
|
///
|
|
/// - Note: You can only cancel a task before it has started executing.
|
|
@discardableResult
|
|
@inlinable
|
|
@preconcurrency
|
|
public func flatScheduleTask<T: Sendable>(
|
|
in delay: TimeAmount,
|
|
file: StaticString = #fileID,
|
|
line: UInt = #line,
|
|
_ task: @escaping @Sendable () throws -> EventLoopFuture<T>
|
|
) -> Scheduled<T> {
|
|
self._flatScheduleTask(in: delay, file: file, line: line, task)
|
|
}
|
|
|
|
@usableFromInline typealias FlatScheduleTaskDelayCallback<T> = @Sendable () throws -> EventLoopFuture<T>
|
|
|
|
@inlinable
|
|
func _flatScheduleTask<T: Sendable>(
|
|
in delay: TimeAmount,
|
|
file: StaticString,
|
|
line: UInt,
|
|
_ task: @escaping FlatScheduleTaskDelayCallback<T>
|
|
) -> Scheduled<T> {
|
|
let promise: EventLoopPromise<T> = self.makePromise(file: file, line: line)
|
|
let scheduled = self.scheduleTask(in: delay, task)
|
|
|
|
scheduled.futureResult.whenComplete { result in
|
|
switch result {
|
|
case .success(let futureResult):
|
|
promise.completeWith(futureResult)
|
|
case .failure(let error):
|
|
promise.fail(error)
|
|
}
|
|
}
|
|
|
|
return .init(promise: promise, cancellationTask: { scheduled.cancel() })
|
|
}
|
|
|
|
/// Creates and returns a new `EventLoopPromise` that will be notified using this `EventLoop` as execution `NIOThread`.
|
|
@inlinable
|
|
public func makePromise<T>(
|
|
of type: T.Type = T.self,
|
|
file: StaticString = #fileID,
|
|
line: UInt = #line
|
|
) -> EventLoopPromise<T> {
|
|
EventLoopPromise<T>(eventLoop: self, file: file, line: line)
|
|
}
|
|
|
|
/// Creates and returns a new `EventLoopFuture` that is already marked as failed. Notifications will be done using this `EventLoop` as execution `NIOThread`.
|
|
///
|
|
/// - Parameters:
|
|
/// - error: the `Error` that is used by the `EventLoopFuture`.
|
|
/// - Returns: a failed `EventLoopFuture`.
|
|
@inlinable
|
|
public func makeFailedFuture<T>(_ error: Error) -> EventLoopFuture<T> {
|
|
EventLoopFuture<T>(eventLoop: self, error: error)
|
|
}
|
|
|
|
/// Creates and returns a new `EventLoopFuture` that is already marked as success. Notifications will be done using this `EventLoop` as execution `NIOThread`.
|
|
///
|
|
/// - Parameters:
|
|
/// - value: the value that is used by the `EventLoopFuture`.
|
|
/// - Returns: a succeeded `EventLoopFuture`.
|
|
@preconcurrency
|
|
@inlinable
|
|
public func makeSucceededFuture<Success: Sendable>(_ value: Success) -> EventLoopFuture<Success> {
|
|
if Success.self == Void.self {
|
|
// The as! will always succeed because we previously checked that Success.self == Void.self.
|
|
return self.makeSucceededVoidFuture() as! EventLoopFuture<Success>
|
|
} else {
|
|
return EventLoopFuture<Success>(eventLoop: self, value: value)
|
|
}
|
|
}
|
|
|
|
/// Creates and returns a new isolated `EventLoopFuture` that is already marked as success. Notifications will be done using this `EventLoop.
|
|
///
|
|
/// - Parameters:
|
|
/// - value: the value that is used by the `EventLoopFuture.Isolated`.
|
|
/// - Returns: a succeeded `EventLoopFuture.Isolated`.
|
|
@inlinable
|
|
@available(*, noasync)
|
|
public func makeSucceededIsolatedFuture<Success>(_ value: Success) -> EventLoopFuture<Success>.Isolated {
|
|
if Success.self == Void.self {
|
|
// The as! will always succeed because we previously checked that Success.self == Void.self.
|
|
return self.makeSucceededVoidFuture().assumeIsolated() as! EventLoopFuture<Success>.Isolated
|
|
} else {
|
|
return EventLoopFuture.Isolated(_wrapped: EventLoopFuture(eventLoop: self, isolatedValue: value))
|
|
}
|
|
}
|
|
|
|
/// Creates and returns a new `EventLoopFuture` that is marked as succeeded or failed with the value held by `result`.
|
|
///
|
|
/// - Parameters:
|
|
/// - result: The value that is used by the `EventLoopFuture`
|
|
/// - Returns: A completed `EventLoopFuture`.
|
|
@preconcurrency
|
|
@inlinable
|
|
public func makeCompletedFuture<Success: Sendable>(_ result: Result<Success, Error>) -> EventLoopFuture<Success> {
|
|
switch result {
|
|
case .success(let value):
|
|
return self.makeSucceededFuture(value)
|
|
case .failure(let error):
|
|
return self.makeFailedFuture(error)
|
|
}
|
|
}
|
|
|
|
/// Creates and returns a new `EventLoopFuture` that is marked as succeeded or failed with the value returned by `body`.
|
|
///
|
|
/// - Parameters:
|
|
/// - body: The function that is used to complete the `EventLoopFuture`
|
|
/// - Returns: A completed `EventLoopFuture`.
|
|
@preconcurrency
|
|
@inlinable
|
|
public func makeCompletedFuture<Success: Sendable>(
|
|
withResultOf body: () throws -> Success
|
|
) -> EventLoopFuture<Success> {
|
|
let trans = Result(catching: body)
|
|
return self.makeCompletedFuture(trans)
|
|
}
|
|
|
|
/// An `EventLoop` forms a singular `EventLoopGroup`, returning itself as the 'next' `EventLoop`.
|
|
///
|
|
/// - Returns: Itself, because an `EventLoop` forms a singular `EventLoopGroup`.
|
|
public func next() -> EventLoop {
|
|
self
|
|
}
|
|
|
|
/// An `EventLoop` forms a singular `EventLoopGroup`, returning itself as 'any' `EventLoop`.
|
|
///
|
|
/// - Returns: Itself, because an `EventLoop` forms a singular `EventLoopGroup`.
|
|
public func any() -> EventLoop {
|
|
self
|
|
}
|
|
|
|
/// Close this `EventLoop`.
|
|
public func close() throws {
|
|
// Do nothing
|
|
}
|
|
|
|
/// Schedule a repeated task to be executed by the `EventLoop` with a fixed delay between the end and start of each
|
|
/// task.
|
|
///
|
|
/// - Parameters:
|
|
/// - initialDelay: The delay after which the first task is executed.
|
|
/// - delay: The delay between the end of one task and the start of the next.
|
|
/// - promise: If non-nil, a promise to fulfill when the task is cancelled and all execution is complete.
|
|
/// - task: The closure that will be executed.
|
|
/// - return: `RepeatedTask`
|
|
@discardableResult
|
|
@preconcurrency
|
|
public func scheduleRepeatedTask(
|
|
initialDelay: TimeAmount,
|
|
delay: TimeAmount,
|
|
notifying promise: EventLoopPromise<Void>? = nil,
|
|
_ task: @escaping @Sendable (RepeatedTask) throws -> Void
|
|
) -> RepeatedTask {
|
|
self._scheduleRepeatedTask(initialDelay: initialDelay, delay: delay, notifying: promise, task)
|
|
}
|
|
|
|
/// Schedule a repeated task to be executed by the `EventLoop` with a fixed delay between the end and start of each
|
|
/// task.
|
|
///
|
|
/// - Parameters:
|
|
/// - initialDelay: The delay after which the first task is executed.
|
|
/// - delay: The delay between the end of one task and the start of the next.
|
|
/// - maximumAllowableJitter: Exclusive upper bound of jitter range added to the `delay` parameter.
|
|
/// - promise: If non-nil, a promise to fulfill when the task is cancelled and all execution is complete.
|
|
/// - task: The closure that will be executed.
|
|
/// - return: `RepeatedTask`
|
|
@discardableResult
|
|
public func scheduleRepeatedTask(
|
|
initialDelay: TimeAmount,
|
|
delay: TimeAmount,
|
|
maximumAllowableJitter: TimeAmount,
|
|
notifying promise: EventLoopPromise<Void>? = nil,
|
|
_ task: @escaping @Sendable (RepeatedTask) throws -> Void
|
|
) -> RepeatedTask {
|
|
let jitteredInitialDelay = Self._getJitteredDelay(
|
|
delay: initialDelay,
|
|
maximumAllowableJitter: maximumAllowableJitter
|
|
)
|
|
let jitteredDelay = Self._getJitteredDelay(delay: delay, maximumAllowableJitter: maximumAllowableJitter)
|
|
return self.scheduleRepeatedTask(
|
|
initialDelay: jitteredInitialDelay,
|
|
delay: jitteredDelay,
|
|
notifying: promise,
|
|
task
|
|
)
|
|
}
|
|
typealias ScheduleRepeatedTaskCallback = @Sendable (RepeatedTask) throws -> Void
|
|
|
|
func _scheduleRepeatedTask(
|
|
initialDelay: TimeAmount,
|
|
delay: TimeAmount,
|
|
notifying promise: EventLoopPromise<Void>?,
|
|
_ task: @escaping ScheduleRepeatedTaskCallback
|
|
) -> RepeatedTask {
|
|
let futureTask: @Sendable (RepeatedTask) -> EventLoopFuture<Void> = { repeatedTask in
|
|
do {
|
|
try task(repeatedTask)
|
|
return self.makeSucceededFuture(())
|
|
} catch {
|
|
return self.makeFailedFuture(error)
|
|
}
|
|
}
|
|
return self.scheduleRepeatedAsyncTask(initialDelay: initialDelay, delay: delay, notifying: promise, futureTask)
|
|
}
|
|
|
|
/// Schedule a repeated asynchronous task to be executed by the `EventLoop` with a fixed delay between the end and
|
|
/// start of each task.
|
|
///
|
|
/// - Note: The delay is measured from the completion of one run's returned future to the start of the execution of
|
|
/// the next run. For example: If you schedule a task once per second but your task takes two seconds to
|
|
/// complete, the time interval between two subsequent runs will actually be three seconds (2s run time plus
|
|
/// the 1s delay.)
|
|
///
|
|
/// - Parameters:
|
|
/// - initialDelay: The delay after which the first task is executed.
|
|
/// - delay: The delay between the end of one task and the start of the next.
|
|
/// - promise: If non-nil, a promise to fulfill when the task is cancelled and all execution is complete.
|
|
/// - task: The closure that will be executed. Task will keep repeating regardless of whether the future
|
|
/// gets fulfilled with success or error.
|
|
///
|
|
/// - return: `RepeatedTask`
|
|
@discardableResult
|
|
@preconcurrency
|
|
public func scheduleRepeatedAsyncTask(
|
|
initialDelay: TimeAmount,
|
|
delay: TimeAmount,
|
|
notifying promise: EventLoopPromise<Void>? = nil,
|
|
_ task: @escaping @Sendable (RepeatedTask) -> EventLoopFuture<Void>
|
|
) -> RepeatedTask {
|
|
self._scheduleRepeatedAsyncTask(initialDelay: initialDelay, delay: delay, notifying: promise, task)
|
|
}
|
|
|
|
/// Schedule a repeated asynchronous task to be executed by the `EventLoop` with a fixed delay between the end and
|
|
/// start of each task.
|
|
///
|
|
/// - Note: The delay is measured from the completion of one run's returned future to the start of the execution of
|
|
/// the next run. For example: If you schedule a task once per second but your task takes two seconds to
|
|
/// complete, the time interval between two subsequent runs will actually be three seconds (2s run time plus
|
|
/// the 1s delay.)
|
|
///
|
|
/// - Parameters:
|
|
/// - initialDelay: The delay after which the first task is executed.
|
|
/// - delay: The delay between the end of one task and the start of the next.
|
|
/// - maximumAllowableJitter: Exclusive upper bound of jitter range added to the `delay` parameter.
|
|
/// - promise: If non-nil, a promise to fulfill when the task is cancelled and all execution is complete.
|
|
/// - task: The closure that will be executed. Task will keep repeating regardless of whether the future
|
|
/// gets fulfilled with success or error.
|
|
///
|
|
/// - return: `RepeatedTask`
|
|
@discardableResult
|
|
public func scheduleRepeatedAsyncTask(
|
|
initialDelay: TimeAmount,
|
|
delay: TimeAmount,
|
|
maximumAllowableJitter: TimeAmount,
|
|
notifying promise: EventLoopPromise<Void>? = nil,
|
|
_ task: @escaping @Sendable (RepeatedTask) -> EventLoopFuture<Void>
|
|
) -> RepeatedTask {
|
|
let jitteredInitialDelay = Self._getJitteredDelay(
|
|
delay: initialDelay,
|
|
maximumAllowableJitter: maximumAllowableJitter
|
|
)
|
|
let jitteredDelay = Self._getJitteredDelay(delay: delay, maximumAllowableJitter: maximumAllowableJitter)
|
|
return self._scheduleRepeatedAsyncTask(
|
|
initialDelay: jitteredInitialDelay,
|
|
delay: jitteredDelay,
|
|
notifying: promise,
|
|
task
|
|
)
|
|
}
|
|
typealias ScheduleRepeatedAsyncTaskCallback = @Sendable (RepeatedTask) -> EventLoopFuture<Void>
|
|
|
|
func _scheduleRepeatedAsyncTask(
|
|
initialDelay: TimeAmount,
|
|
delay: TimeAmount,
|
|
notifying promise: EventLoopPromise<Void>?,
|
|
_ task: @escaping ScheduleRepeatedAsyncTaskCallback
|
|
) -> RepeatedTask {
|
|
let repeated = RepeatedTask(interval: delay, eventLoop: self, cancellationPromise: promise, task: task)
|
|
repeated.begin(in: initialDelay)
|
|
return repeated
|
|
}
|
|
|
|
/// Adds a random amount of `.nanoseconds` (within `.zero..<maximumAllowableJitter`) to the delay.
|
|
///
|
|
/// - Parameters:
|
|
/// - delay: the `TimeAmount` delay to jitter.
|
|
/// - maximumAllowableJitter: Exclusive upper bound of jitter range added to the `delay` parameter.
|
|
/// - Returns: The jittered delay.
|
|
@inlinable
|
|
static func _getJitteredDelay(
|
|
delay: TimeAmount,
|
|
maximumAllowableJitter: TimeAmount
|
|
) -> TimeAmount {
|
|
let jitter = TimeAmount.nanoseconds(Int64.random(in: .zero..<maximumAllowableJitter.nanoseconds))
|
|
return delay + jitter
|
|
}
|
|
|
|
/// Returns an `EventLoopIterator` over this `EventLoop`.
|
|
///
|
|
/// - Returns: `EventLoopIterator`
|
|
public func makeIterator() -> EventLoopIterator {
|
|
EventLoopIterator([self])
|
|
}
|
|
|
|
/// Asserts that the current thread is the one tied to this `EventLoop`.
|
|
/// Otherwise, if running in debug mode, the process will be abnormally terminated as per the semantics of
|
|
/// `preconditionFailure(_:file:line:)`. Never has any effect in release mode.
|
|
///
|
|
/// - Note: This is not a customization point so calls to this function can be fully optimized out in release mode.
|
|
@inlinable
|
|
public func assertInEventLoop(file: StaticString = #fileID, line: UInt = #line) {
|
|
debugOnly {
|
|
self.preconditionInEventLoop(file: file, line: line)
|
|
}
|
|
}
|
|
|
|
/// Asserts that the current thread is _not_ the one tied to this `EventLoop`.
|
|
/// Otherwise, if running in debug mode, the process will be abnormally terminated as per the semantics of
|
|
/// `preconditionFailure(_:file:line:)`. Never has any effect in release mode.
|
|
///
|
|
/// - Note: This is not a customization point so calls to this function can be fully optimized out in release mode.
|
|
@inlinable
|
|
public func assertNotInEventLoop(file: StaticString = #fileID, line: UInt = #line) {
|
|
debugOnly {
|
|
self.preconditionNotInEventLoop(file: file, line: line)
|
|
}
|
|
}
|
|
|
|
/// Checks the necessary condition of currently running on the called `EventLoop` for making forward progress.
|
|
@inlinable
|
|
public func preconditionInEventLoop(file: StaticString = #fileID, line: UInt = #line) {
|
|
precondition(self.inEventLoop, file: file, line: line)
|
|
}
|
|
|
|
/// Checks the necessary condition of currently _not_ running on the called `EventLoop` for making forward progress.
|
|
@inlinable
|
|
public func preconditionNotInEventLoop(file: StaticString = #fileID, line: UInt = #line) {
|
|
precondition(!self.inEventLoop, file: file, line: line)
|
|
}
|
|
}
|
|
|
|
/// Provides an endless stream of `EventLoop`s to use.
|
|
public protocol EventLoopGroup: AnyObject, _NIOPreconcurrencySendable {
|
|
/// Returns the next `EventLoop` to use, this is useful for load balancing.
|
|
///
|
|
/// The algorithm that is used to select the next `EventLoop` is specific to each `EventLoopGroup`. A common choice
|
|
/// is _round robin_.
|
|
///
|
|
/// Please note that you should only be using `next()` if you want to load balance over all `EventLoop`s of the
|
|
/// `EventLoopGroup`. If the actual `EventLoop` does not matter much, `any()` should be preferred because it can
|
|
/// try to return you the _current_ `EventLoop` which usually is faster because the number of thread switches can
|
|
/// be reduced.
|
|
///
|
|
/// The rule of thumb is: If you are trying to do _load balancing_, use `next()`. If you just want to create a new
|
|
/// future or kick off some operation, use `any()`.
|
|
|
|
func next() -> EventLoop
|
|
|
|
/// Returns any `EventLoop` from the `EventLoopGroup`, a common choice is the current `EventLoop`.
|
|
///
|
|
/// - warning: You cannot rely on the returned `EventLoop` being the current one, not all `EventLoopGroup`s support
|
|
/// choosing the current one. Use this method only if you are truly happy with _any_ `EventLoop` of this
|
|
/// `EventLoopGroup` instance.
|
|
///
|
|
/// - Note: You will only receive the current `EventLoop` here iff the current `EventLoop` belongs to the
|
|
/// `EventLoopGroup` you call `any()` on.
|
|
///
|
|
/// This method is useful having access to an `EventLoopGroup` without the knowledge of which `EventLoop` would be
|
|
/// the best one to select to create a new `EventLoopFuture`. This commonly happens in libraries where the user
|
|
/// cannot indicate what `EventLoop` they would like their futures on.
|
|
///
|
|
/// Typically, it is faster to kick off a new operation on the _current_ `EventLoop` because that minimised thread
|
|
/// switches. Hence, if situations where you don't need precise knowledge of what `EventLoop` some code is running
|
|
/// on, use `any()` to indicate this.
|
|
///
|
|
/// The rule of thumb is: If you are trying to do _load balancing_, use `next()`. If you just want to create a new
|
|
/// future or kick off some operation, use `any()`.
|
|
func any() -> EventLoop
|
|
|
|
#if canImport(Dispatch)
|
|
/// Shuts down the eventloop gracefully. This function is clearly an outlier in that it uses a completion
|
|
/// callback instead of an EventLoopFuture. The reason for that is that NIO's EventLoopFutures will call back on an event loop.
|
|
/// The virtue of this function is to shut the event loop down. To work around that we call back on a DispatchQueue
|
|
/// instead.
|
|
@preconcurrency func shutdownGracefully(queue: DispatchQueue, _ callback: @Sendable @escaping (Error?) -> Void)
|
|
#endif
|
|
|
|
/// Returns an `EventLoopIterator` over the `EventLoop`s in this `EventLoopGroup`.
|
|
///
|
|
/// - Returns: `EventLoopIterator`
|
|
func makeIterator() -> EventLoopIterator
|
|
|
|
/// Must crash if it's not safe to call `syncShutdownGracefully` in the current context.
|
|
///
|
|
/// This method is a debug hook that can be used to override the behaviour of `syncShutdownGracefully`
|
|
/// when called. By default it does nothing.
|
|
func _preconditionSafeToSyncShutdown(file: StaticString, line: UInt)
|
|
}
|
|
|
|
extension EventLoopGroup {
|
|
/// The default implementation of `any()` just returns the `next()` EventLoop but it's highly recommended to
|
|
/// override this and return the current `EventLoop` if possible.
|
|
public func any() -> EventLoop {
|
|
self.next()
|
|
}
|
|
}
|
|
|
|
#if canImport(Dispatch)
|
|
extension EventLoopGroup {
|
|
@preconcurrency public func shutdownGracefully(_ callback: @escaping @Sendable (Error?) -> Void) {
|
|
self.shutdownGracefully(queue: .global(), callback)
|
|
}
|
|
|
|
@available(*, noasync, message: "this can end up blocking the calling thread", renamed: "shutdownGracefully()")
|
|
public func syncShutdownGracefully() throws {
|
|
try self._syncShutdownGracefully()
|
|
}
|
|
|
|
private func _syncShutdownGracefully() throws {
|
|
self._preconditionSafeToSyncShutdown(file: #fileID, line: #line)
|
|
let error = NIOLockedValueBox<Error?>(nil)
|
|
let semaphore = DispatchSemaphore(value: 0)
|
|
self.shutdownGracefully { shutdownError in
|
|
if let shutdownError = shutdownError {
|
|
error.withLockedValue {
|
|
$0 = shutdownError
|
|
}
|
|
}
|
|
semaphore.signal()
|
|
}
|
|
semaphore.wait()
|
|
try error.withLockedValue { error in
|
|
if let error = error {
|
|
throw error
|
|
}
|
|
}
|
|
}
|
|
|
|
public func _preconditionSafeToSyncShutdown(file: StaticString, line: UInt) {
|
|
return
|
|
}
|
|
}
|
|
#endif
|
|
|
|
/// Deprecated.
|
|
///
|
|
/// This type was intended to be used by libraries which use NIO, and offer their users either the option
|
|
/// to `.share` an existing event loop group or create (and manage) a new one (`.createNew`) and let it be
|
|
/// managed by given library and its lifecycle.
|
|
///
|
|
/// Please use a `group: any EventLoopGroup` parameter instead. If you want to default to a global
|
|
/// singleton group instead, consider group: any EventLoopGroup = MultiThreadedEventLoopGroup.singleton` or
|
|
/// similar.
|
|
///
|
|
/// - See also: https://github.com/apple/swift-nio/issues/2142
|
|
public enum NIOEventLoopGroupProvider {
|
|
/// Use an `EventLoopGroup` provided by the user.
|
|
/// The owner of this group is responsible for its lifecycle.
|
|
case shared(EventLoopGroup)
|
|
|
|
/// Deprecated. Create a new `EventLoopGroup` when necessary.
|
|
/// The library which accepts this provider takes ownership of the created event loop group,
|
|
/// and must ensure its proper shutdown when the library is being shut down.
|
|
@available(
|
|
*,
|
|
deprecated,
|
|
message: """
|
|
Please use `.shared(existingGroup)` or use the singleton via \
|
|
`.shared(MultiThreadedEventLoopGroup.singleton)` or similar
|
|
"""
|
|
)
|
|
case createNew
|
|
}
|
|
|
|
extension NIOEventLoopGroupProvider: Sendable {}
|
|
|
|
/// Different `Error`s that are specific to `EventLoop` operations / implementations.
|
|
public enum EventLoopError: Error {
|
|
/// An operation was executed that is not supported by the `EventLoop`
|
|
case unsupportedOperation
|
|
|
|
/// An scheduled task was cancelled.
|
|
case cancelled
|
|
|
|
/// The `EventLoop` was shutdown already.
|
|
case shutdown
|
|
|
|
/// Shutting down the `EventLoop` failed.
|
|
case shutdownFailed
|
|
}
|
|
|
|
extension EventLoopError {
|
|
@usableFromInline
|
|
static let _cancelled: any Error = EventLoopError.cancelled
|
|
}
|
|
|
|
extension EventLoopError: CustomStringConvertible {
|
|
public var description: String {
|
|
switch self {
|
|
case .unsupportedOperation:
|
|
return "EventLoopError: the executed operation is not supported by the event loop"
|
|
case .cancelled:
|
|
return "EventLoopError: the scheduled task was cancelled"
|
|
case .shutdown:
|
|
return "EventLoopError: the event loop is shutdown"
|
|
case .shutdownFailed:
|
|
return "EventLoopError: failed to shutdown the event loop"
|
|
}
|
|
}
|
|
}
|