From aa347827fb7d979296c1bd9fbbc151954e0b33aa Mon Sep 17 00:00:00 2001 From: grepjava Date: Sat, 10 Oct 2026 16:13:01 +0800 Subject: [PATCH 1/2] Let a loop executor time its jobs and report the slow ones by task A job runs with everything else on the loop waiting, so a task that computes without awaiting holds up every socket and timer the loop has, and nothing said so. Set slowJobMicros and a drain reads the clock around each job and hands onSlowJob the address and running time of any that ran past it. A task's job is the task itself, so address(of:) lets the loop say which of its tasks it was. Off by default, when a drain pays one branch per job. --- RELEASE.md | 8 ++++ Sources/AvianCore/LoopExecutor.swift | 40 +++++++++++++++- Tests/AvianTests/LoopExecutorTests.swift | 58 +++++++++++++++++++++++- 3 files changed, 104 insertions(+), 2 deletions(-) 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..a967e22 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 = 5_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(20_000) + } + Task(executorPreference: executor) { + withUnsafeCurrentTask { trail.addresses.append(LoopExecutor.address(of: $0!)) } + _ = await withUnsafeContinuation { trail.parked = $0 } + spin(20_000) + } + executor.drain() + #expect(trail.addresses.count == 3) + #expect(trail.slow.map(\.address) == [trail.addresses[1]]) + #expect(trail.slow.allSatisfy { $0.micros >= 20_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) + } } From 1cf97ba8be6ffbd946dd438808569d8dfddce744 Mon Sep 17 00:00:00 2001 From: grepjava Date: Sat, 10 Oct 2026 16:55:55 +0800 Subject: [PATCH 2/2] Give the job-timing test room on a slow runner: 25 ms limit, 60 ms spins --- Tests/AvianTests/LoopExecutorTests.swift | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/Tests/AvianTests/LoopExecutorTests.swift b/Tests/AvianTests/LoopExecutorTests.swift index a967e22..e4f9791 100644 --- a/Tests/AvianTests/LoopExecutorTests.swift +++ b/Tests/AvianTests/LoopExecutorTests.swift @@ -214,7 +214,7 @@ private func chain(_ step: Int, upTo last: Int, on executor: LoopExecutor, _ tra defer { owner.deallocate() } let executor = LoopExecutor(owner: owner) let trail = Trail() - executor.slowJobMicros = 5_000 + 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. @@ -223,17 +223,17 @@ private func chain(_ step: Int, upTo last: Int, on executor: LoopExecutor, _ tra } Task(executorPreference: executor) { withUnsafeCurrentTask { trail.addresses.append(LoopExecutor.address(of: $0!)) } - spin(20_000) + spin(60_000) } Task(executorPreference: executor) { withUnsafeCurrentTask { trail.addresses.append(LoopExecutor.address(of: $0!)) } _ = await withUnsafeContinuation { trail.parked = $0 } - spin(20_000) + spin(60_000) } executor.drain() #expect(trail.addresses.count == 3) #expect(trail.slow.map(\.address) == [trail.addresses[1]]) - #expect(trail.slow.allSatisfy { $0.micros >= 20_000 }) + #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()