diff --git a/RELEASE.md b/RELEASE.md index d584cf1..52d20ed 100644 --- a/RELEASE.md +++ b/RELEASE.md @@ -10,6 +10,14 @@ tagging, run the unit and end-to-end suites of Garuda and Peregrine against the new version as well: what they exercise here is only what this package tests of itself. +## Unreleased + +- **`LoopExecutor` can time its jobs.** Set `slowJobMicros` and a drain + reports each job that ran at least that long to `onSlowJob`, with how long + it ran and the job's address. For a task that is the address + `LoopExecutor.address(of:)` gives for it, so a loop can tell which of its + tasks held it up. Off by default, when a drain costs what it did. + ## 0.7.2 — 2026-10-03 - **`LoopExecutor.drain(budget:)` runs at most `budget` jobs and says whether diff --git a/Sources/AvianCore/LoopExecutor.swift b/Sources/AvianCore/LoopExecutor.swift index 696e92b..30f7041 100644 --- a/Sources/AvianCore/LoopExecutor.swift +++ b/Sources/AvianCore/LoopExecutor.swift @@ -16,6 +16,12 @@ // // The owner is whatever `av_worker_set_current` was given on the loop's // thread. Nothing else about it is used. +// +// Everything else on the loop waits while a job runs, so a job that runs long +// without waiting -- a task doing CPU work inline -- holds up every socket +// and timer the loop has. Set `slowJobMicros` and a drain times each job and +// tells `onSlowJob` about those that ran past it, by the job's address: for a +// task, the task's own (`address(of:)`), so the loop can say whose it was. //===----------------------------------------------------------------------===// import CAvian @@ -47,6 +53,25 @@ public final class LoopExecutor: TaskExecutor, @unchecked Sendable { public let wakeFD: Int32 private let wakeWriteFD: Int32 + /// A job that runs at least this many microseconds is reported to + /// `onSlowJob`. 0, the default, times nothing: a drain then pays one + /// branch per job. Set on the owner's thread. + public var slowJobMicros: UInt64 = 0 + /// Called on the owner's thread, after a job that ran at least + /// `slowJobMicros`, with the job's address and how long it ran in + /// microseconds. A task's job has the task's own address, the one + /// `address(of:)` gives for it, and the task may have ended by now: the + /// address is something to look up, never to follow. + public var onSlowJob: ((UnsafeRawPointer, UInt64) -> Void)? = nil + + /// The address a job of `task` has when it is run here, to recognise it + /// by in `onSlowJob`. + public static func address(of task: UnsafeCurrentTask) -> UnsafeRawPointer { + // A task is the job that runs it: the runtime enqueues the task + // itself, and both types are a single reference to it. + unsafeBitCast(task, to: UnsafeRawPointer.self) + } + /// `owner` is the value `av_worker_current()` returns on the loop's /// thread. `capacity` is where the ring starts; it grows while jobs are /// queued faster than they run. @@ -133,12 +158,25 @@ public final class LoopExecutor: TaskExecutor, @unchecked Sendable { head = (head &+ 1) & (capacity &- 1) count -= 1 left -= 1 - job.runSynchronously(on: asUnownedTaskExecutor()) + if slowJobMicros == 0 { + job.runSynchronously(on: asUnownedTaskExecutor()) + } else { + runTimed(job) + } } draining = false return hasWork } + /// Runs a job, and reports it if it ran past `slowJobMicros`. + private func runTimed(_ job: UnownedJob) { + let address = unsafeBitCast(job, to: UnsafeRawPointer.self) + let started = av_monotonic_us() + job.runSynchronously(on: asUnownedTaskExecutor()) + let ran = av_monotonic_us() &- started + if ran >= slowJobMicros { onSlowJob?(address, ran) } + } + /// Empties the wake pipe. Call it before draining, so that a byte /// arriving while the jobs run is not read away with them. public func clearWake() { diff --git a/Tests/AvianTests/LoopExecutorTests.swift b/Tests/AvianTests/LoopExecutorTests.swift index 6af4f0b..e4f9791 100644 --- a/Tests/AvianTests/LoopExecutorTests.swift +++ b/Tests/AvianTests/LoopExecutorTests.swift @@ -1,7 +1,8 @@ //===----------------------------------------------------------------------===// // LoopExecutor: jobs run only when the loop drains, in order, on the owner's // thread; a job from another thread is handed over and wakes the pipe; a -// budgeted drain stops at its budget however many jobs the jobs queue. +// budgeted drain stops at its budget however many jobs the jobs queue; a +// timed drain reports the jobs that ran long, by their task's address. //===----------------------------------------------------------------------===// import Testing @@ -14,6 +15,14 @@ private final class Trail: @unchecked Sendable { var steps: [Int] = [] var threads: [UnsafeMutableRawPointer?] = [] var parked: UnsafeContinuation? = nil + var addresses: [UnsafeRawPointer] = [] + var slow: [(address: UnsafeRawPointer, micros: UInt64)] = [] +} + +/// Runs on the calling thread, without waiting, for `micros`. +private func spin(_ micros: UInt64) { + let until = av_monotonic_us() &+ micros + while av_monotonic_us() < until {} } /// Queues a task on `executor` that records `step` and queues the next, up to @@ -198,4 +207,51 @@ private func chain(_ step: Int, upTo last: Int, on executor: LoopExecutor, _ tra #expect(Set(trail.steps) == [1, 2]) #expect(!executor.hasWork) } + + @available(macOS 15, *) + @Test func aTimedDrainReportsTheJobsThatRanLongByTheirTask() { + let owner = UnsafeMutableRawPointer.allocate(byteCount: 1, alignment: 1) + defer { owner.deallocate() } + let executor = LoopExecutor(owner: owner) + let trail = Trail() + executor.slowJobMicros = 25_000 + executor.onSlowJob = { address, micros in trail.slow.append((address, micros)) } + asOwner(owner) { + // One quick, one slow, and one slow only after it has waited. + Task(executorPreference: executor) { + withUnsafeCurrentTask { trail.addresses.append(LoopExecutor.address(of: $0!)) } + } + Task(executorPreference: executor) { + withUnsafeCurrentTask { trail.addresses.append(LoopExecutor.address(of: $0!)) } + spin(60_000) + } + Task(executorPreference: executor) { + withUnsafeCurrentTask { trail.addresses.append(LoopExecutor.address(of: $0!)) } + _ = await withUnsafeContinuation { trail.parked = $0 } + spin(60_000) + } + executor.drain() + #expect(trail.addresses.count == 3) + #expect(trail.slow.map(\.address) == [trail.addresses[1]]) + #expect(trail.slow.allSatisfy { $0.micros >= 60_000 }) + // Its resumption is a job of the same task, at the same address. + trail.parked?.resume(returning: 0) + executor.drain() + } + #expect(trail.slow.map(\.address) == [trail.addresses[1], trail.addresses[2]]) + } + + @available(macOS 15, *) + @Test func anUntimedDrainReportsNothing() { + let owner = UnsafeMutableRawPointer.allocate(byteCount: 1, alignment: 1) + defer { owner.deallocate() } + let executor = LoopExecutor(owner: owner) + let trail = Trail() + executor.onSlowJob = { address, micros in trail.slow.append((address, micros)) } + asOwner(owner) { + Task(executorPreference: executor) { spin(10_000) } + executor.drain() + } + #expect(trail.slow.isEmpty) + } }