LoopExecutor: time jobs and report slow ones by task (0.7.3)

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

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() {

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<Int, Never>? = 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)
+    }
 }