//===----------------------------------------------------------------------===// // // 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 { @usableFromInline typealias CancelationCallback = @Sendable () -> Void @usableFromInline let _promise: EventLoopPromise @usableFromInline let _cancellationTask: CancelationCallback @inlinable @preconcurrency public init(promise: EventLoopPromise, 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 { 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 private let delay: TimeAmount private let eventLoop: EventLoop private let cancellationPromise: EventLoopPromise? private var scheduled: Optional>> private var task: Optional internal init( interval: TimeAmount, eventLoop: EventLoop, cancellationPromise: EventLoopPromise? = 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? = nil) { if self.eventLoop.inEventLoop { self.cancel0(localCancellationPromise: promise) } else { self.eventLoop.execute { self.cancel0(localCancellationPromise: promise) } } } private func cancel0(localCancellationPromise: EventLoopPromise?) { 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) 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(_ task: @escaping @Sendable () throws -> T) -> EventLoopFuture /// 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(deadline: NIODeadline, _ task: @escaping @Sendable () throws -> T) -> Scheduled /// 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(in: TimeAmount, _ task: @escaping @Sendable () throws -> T) -> Scheduled /// 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 /// 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(_ task: @escaping () throws -> T) -> EventLoopFuture /// 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( deadline: NIODeadline, _ task: @escaping () throws -> T ) -> Scheduled /// 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( in: TimeAmount, _ task: @escaping () throws -> T ) -> Scheduled /// 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 { 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(_ task: @escaping () throws -> T) -> EventLoopFuture { 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( deadline: NIODeadline, _ task: @escaping () throws -> T ) -> Scheduled { 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( in delay: TimeAmount, _ task: @escaping () throws -> T ) -> Scheduled { 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 * (lhs: T, rhs: TimeAmount) -> TimeAmount { TimeAmount(Int64(lhs) * rhs.nanoseconds) } @inlinable public static func * (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 { /// 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(_ task: @escaping @Sendable () throws -> T) -> EventLoopFuture { let promise: EventLoopPromise = 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(_ task: @escaping @Sendable () -> EventLoopFuture) -> EventLoopFuture { 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( deadline: NIODeadline, file: StaticString = #fileID, line: UInt = #line, _ task: @escaping @Sendable () throws -> EventLoopFuture ) -> Scheduled { let promise: EventLoopPromise = 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( in delay: TimeAmount, file: StaticString = #fileID, line: UInt = #line, _ task: @escaping @Sendable () throws -> EventLoopFuture ) -> Scheduled { self._flatScheduleTask(in: delay, file: file, line: line, task) } @usableFromInline typealias FlatScheduleTaskDelayCallback = @Sendable () throws -> EventLoopFuture @inlinable func _flatScheduleTask( in delay: TimeAmount, file: StaticString, line: UInt, _ task: @escaping FlatScheduleTaskDelayCallback ) -> Scheduled { let promise: EventLoopPromise = 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( of type: T.Type = T.self, file: StaticString = #fileID, line: UInt = #line ) -> EventLoopPromise { EventLoopPromise(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(_ error: Error) -> EventLoopFuture { EventLoopFuture(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(_ value: Success) -> EventLoopFuture { if Success.self == Void.self { // The as! will always succeed because we previously checked that Success.self == Void.self. return self.makeSucceededVoidFuture() as! EventLoopFuture } else { return EventLoopFuture(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(_ value: Success) -> EventLoopFuture.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.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(_ result: Result) -> EventLoopFuture { 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( withResultOf body: () throws -> Success ) -> EventLoopFuture { 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? = 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? = 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?, _ task: @escaping ScheduleRepeatedTaskCallback ) -> RepeatedTask { let futureTask: @Sendable (RepeatedTask) -> EventLoopFuture = { 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? = nil, _ task: @escaping @Sendable (RepeatedTask) -> EventLoopFuture ) -> 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? = nil, _ task: @escaping @Sendable (RepeatedTask) -> EventLoopFuture ) -> 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 func _scheduleRepeatedAsyncTask( initialDelay: TimeAmount, delay: TimeAmount, notifying promise: EventLoopPromise?, _ 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.. TimeAmount { let jitter = TimeAmount.nanoseconds(Int64.random(in: .zero.. 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(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" } } }