Merge branch 'diag/ae445-growth-fingerprint'

Author: superuser404notfound

Sources/AetherEngine/AetherEngine+Diagnostics.swift

@@ -17,10 +17,17 @@ extension AetherEngine {
     /// #220 turned out to be, so the counter is polled at `triggerPollHz` and the zone walk runs once
     /// it climbs `triggerThresholdMB` above its running high-water. Pass `triggerPollHz: 0` for the
     /// plain 30 s census with no watcher.
+    ///
+    /// `triggerCaptureCap` bounds how many of those walks are logged (`0` = uncapped). The default of
+    /// twelve keeps a runaway from turning the log into a slideshow, but a session that climbs at a
+    /// steady mux rate spends one capture per threshold climbed and reaches the cap minutes before
+    /// the kill (AE#445, where the decisive final step survived only in the 30 s grid). Lift it when
+    /// the shape being hunted is a steady climb rather than a single step.
     public nonisolated static func setLargeAllocationCensusEnabled(
         _ enabled: Bool,
         triggerThresholdMB: Int = 64,
-        triggerPollHz: Double = 8
+        triggerPollHz: Double = 8,
+        triggerCaptureCap: Int = 12   // MallocBlockCensus.defaultTriggerCaptureCap, spelled out because that type is internal
     ) {
         MallocBlockCensus.isEnabled = enabled
         // AE#445: the same switch, because they answer halves of one question. The malloc census
@@ -29,7 +36,8 @@ extension AetherEngine {
         VMRegionCensus.isEnabled = enabled
         if !enabled { VMRegionCensus.clearBaseline() }
         if enabled {
-            MallocBlockCensus.startTriggerWatch(thresholdMB: triggerThresholdMB, pollHz: triggerPollHz)
+            MallocBlockCensus.startTriggerWatch(thresholdMB: triggerThresholdMB, pollHz: triggerPollHz,
+                                                captureCap: triggerCaptureCap)
         } else {
             MallocBlockCensus.stopTriggerWatch()
         }

Sources/AetherEngine/Diagnostics/MallocBlockCensus.swift

@@ -101,6 +101,73 @@ enum MallocBlockCensus {
         return Result(count: total, bytes: totalBytes, buckets: buckets, largest: largest)
     }
 
+    /// AE#445: the growth POLICY of the largest block, which names the allocator family holding it.
+    ///
+    /// `bigExact` says one block is growing. Deciding WHOSE it is took that issue three rounds, and
+    /// the ratio between two consecutive walks answers it more sharply than any size does, because
+    /// each family grows by its own fixed factor and nothing else lands on one:
+    ///
+    ///   1.25   Foundation `Data` (`__DataStorage._grow` adds `newLength >> 2` above 128 KB, rounds
+    ///          through `malloc_good_size`, and grows via `realloc`, so the region also carries the
+    ///          REALLOC tag)
+    ///   1.5    FFmpeg's AVIO dynamic buffer (`dyn_buf_write`: `size += size / 2 + 1`)
+    ///   1.0625 `av_fast_realloc` / `av_fast_malloc` (`min_size + min_size / 16 + 32`)
+    ///   2      Swift `Array` / `ContiguousArray` / `String`
+    ///
+    /// A block can climb several rungs between two walks, so a ratio is matched against the first
+    /// four powers of each factor and the fewest-rung match wins. The census tracks sizes and not
+    /// identities: if the top block changes hands the ratio belongs to no family, which is why an
+    /// unrecognised value is printed bare instead of being rounded into the nearest story.
+    static let growthFamilies: [(name: String, factor: Double)] = [
+        ("Data", 1.25),
+        ("dyn_buf", 1.5),
+        ("Array", 2.0),
+        ("av_fast_realloc", 1.0625),
+    ]
+
+    /// Relative tolerance per rung. The reporter's own ladder measured 1.2500 to 1.2502, i.e. 0.02%,
+    /// and the nearest pair of neighbouring rungs across the table sits 2.4% apart, so this
+    /// separates the families without needing them to be exact.
+    static let growthTolerance = 0.004
+
+    /// Classify one growth step. Nil when the block did not grow.
+    static func growthFamily(previousBytes: Int, currentBytes: Int) -> (ratio: Double, family: String?)? {
+        guard previousBytes > 0, currentBytes > previousBytes else { return nil }
+        let ratio = Double(currentBytes) / Double(previousBytes)
+        for rungs in 1...4 {
+            for family in growthFamilies {
+                let step = pow(family.factor, Double(rungs))
+                if abs(ratio - step) <= growthTolerance * step { return (ratio, family.name) }
+            }
+        }
+        return (ratio, nil)
+    }
+
+    /// Per-caller growth state. The 30 s memprobe walk and the 8 Hz trigger walk must not share a
+    /// previous value: each would then report the ratio the OTHER's interval produced, which is a
+    /// number about the sampler rather than about the buffer.
+    final class GrowthTracker: @unchecked Sendable {
+        private let lock = NSLock()
+        private var previousLargest = 0
+
+        /// `bigGrowth=` fragment for this caller's cadence. "flat" is emitted deliberately: a line
+        /// that only speaks while something grows cannot say that nothing did.
+        func fragment(largest: Int) -> String {
+            lock.lock()
+            let previous = previousLargest
+            previousLargest = largest
+            lock.unlock()
+            guard let step = MallocBlockCensus.growthFamily(previousBytes: previous, currentBytes: largest) else {
+                return "bigGrowth=flat "
+            }
+            let family = step.family ?? "?"
+            return String(format: "bigGrowth=%.4fx(%@) ", step.ratio, family)
+        }
+    }
+
+    static let memprobeGrowth = GrowthTracker()
+    static let triggerGrowth = GrowthTracker()
+
     /// memprobe fragment: total, summed size, and the largest buckets.
     /// Empty string when disabled, so the line shape is unchanged for normal sessions.
     static func probeFragment(topBuckets: Int = 4) -> String {
@@ -112,6 +179,7 @@ enum MallocBlockCensus {
         return "bigBlocks=\(result.count) bigMB=\(result.bytes / (1 << 20)) "
             + "bigTop=\(top.isEmpty ? "none" : top) "
             + "bigExact=\(exact.isEmpty ? "none" : exact) "
+            + memprobeGrowth.fragment(largest: result.largest.first ?? 0)
     }
 }
 

Sources/AetherEngine/Diagnostics/MallocBlockCensusTrigger.swift

@@ -25,7 +25,14 @@ extension MallocBlockCensus {
     /// Emitted lines are capped: the escalation itself is informative (a step, then a bigger step),
     /// but a sustained explosion must not turn the log into a slideshow of zone walks while the
     /// device is already under pressure.
-    static let maxTriggerCaptures = 12
+    ///
+    /// AE#445 bought the exception the cap needed. A session that climbs at a steady mux rate spends
+    /// one capture per threshold climbed, so twelve of them are gone long before the kill: the
+    /// reporter's run hit the cap 4.4 minutes early and the decisive final step survived only in the
+    /// 30 s grid. The cap is therefore configurable through `setLargeAllocationCensusEnabled`, and
+    /// `0` means uncapped for exactly that shape.
+    static let defaultTriggerCaptureCap = 12
+    nonisolated(unsafe) private(set) static var triggerCaptureCap = defaultTriggerCaptureCap
 
     nonisolated(unsafe) private static var watchQueue: DispatchQueue?
     nonisolated(unsafe) private static var watchTimer: DispatchSourceTimer?
@@ -53,10 +60,12 @@ extension MallocBlockCensus {
     /// HIGH-WATER to arm a capture, so a session that climbs gently spends one capture per threshold
     /// climbed while ordinary oscillation spends none. It should sit well above that gentle climb and
     /// far below the steps being hunted, which run from hundreds of MB to gigabytes.
-    static func startTriggerWatch(thresholdMB: Int, pollHz: Double) {
+    static func startTriggerWatch(thresholdMB: Int, pollHz: Double,
+                                  captureCap: Int = defaultTriggerCaptureCap) {
         stopTriggerWatch()
         guard isEnabled, thresholdMB > 0, pollHz > 0 else { return }
         thresholdBytes = thresholdMB << 20
+        triggerCaptureCap = max(0, captureCap)
         highWater = sizeInUse()
         triggerCaptures = 0
 
@@ -71,7 +80,8 @@ extension MallocBlockCensus {
 
         EngineLog.emit(
             "[AetherEngine] census trigger armed: threshold=\(thresholdMB)MB/poll "
-            + "poll=\(String(format: "%.1f", pollHz))Hz baseline=\(highWater >> 20)MB",
+            + "poll=\(String(format: "%.1f", pollHz))Hz baseline=\(highWater >> 20)MB "
+            + "cap=\(triggerCaptureCap == 0 ? "none" : String(triggerCaptureCap))",
             category: .engine
         )
     }
@@ -95,7 +105,7 @@ extension MallocBlockCensus {
         let previousHighWater = highWater
         defer { if now > highWater { highWater = now } }
         guard now >= previousHighWater + thresholdBytes,
-              triggerCaptures < maxTriggerCaptures
+              triggerCaptureCap == 0 || triggerCaptures < triggerCaptureCap
         else { return }
         triggerCaptures += 1
         let delta = now - previousHighWater
@@ -113,14 +123,15 @@ extension MallocBlockCensus {
             let exact = result.largest.map(String.init).joined(separator: ",")
             line += "bigBlocks=\(result.count) bigMB=\(result.bytes >> 20) "
                 + "bigTop=\(top.isEmpty ? "none" : top) "
-                + "bigExact=\(exact.isEmpty ? "none" : exact)"
+                + "bigExact=\(exact.isEmpty ? "none" : exact) "
+                + triggerGrowth.fragment(largest: result.largest.first ?? 0)
         } else {
             line += "census=unavailable"
         }
         EngineLog.emit(line, category: .engine)
-        if triggerCaptures == maxTriggerCaptures {
+        if triggerCaptureCap > 0, triggerCaptures == triggerCaptureCap {
             EngineLog.emit(
-                "[AetherEngine] census trigger cap reached (\(maxTriggerCaptures)); "
+                "[AetherEngine] census trigger cap reached (\(triggerCaptureCap)); "
                 + "further jumps are not captured, peak keeps updating",
                 category: .engine
             )

Sources/aetherctl/LiveSpoolIO.swift

@@ -28,6 +28,24 @@ import AetherEngine
 /// buffer and allocates nothing, so that run measured the HARNESS rather than his case. The default
 /// arm is therefore the allocation-free one: what it measures is the engine. `--foundation-reader`
 /// restores the allocating arm, which is now a control for the pool rather than the subject.
+/// AE#445 round 3: the ingest-side carry, as a POSITIVE control.
+///
+/// The reporter's census named his growing block precisely: one `REALLOC`-tagged allocation on an
+/// exact x1.25 ladder whose content is every byte the session consumed. That factor is Foundation's,
+/// not libav's (`Data.__DataStorage._grow` adds `newLength >> 2` above 128 KB; `av_fast_realloc`
+/// adds a sixteenth, the AVIO dynamic buffer a half), so the block is a Swift `Data`, and the one
+/// `Data` shape that grows like that while its `count` stays small is a parse carry consumed from
+/// the front with `removeFirst`: that only advances the slice's lower bound, so the backing store
+/// keeps every byte below it and reallocs to fit the ever-rising upper bound. The engine paid for
+/// this lesson twice on its own readers (70430de, `ByteFIFO`) and re-bases with `subdata` in both.
+///
+/// `--host-carry removeFirst` puts that shape back into the harness on purpose, so the tool that
+/// measures the engine at ratio 0.00 can also produce the reporter's ratio 1.00 on demand and name
+/// the cause. `--host-carry subdata` is the same carry re-based, which is the fix.
+enum HostCarryTrim: String {
+    case none, removeFirst, subdata
+}
+
 final class PacedLiveSpoolIOReader: IOReader, @unchecked Sendable {
     private let path: String
     /// Foundation arm only. The POSIX arm reads through `fd` and never builds an object.
@@ -56,9 +74,19 @@ final class PacedLiveSpoolIOReader: IOReader, @unchecked Sendable {
     private(set) var maxLookbackBytes: Int64 = 0
     private(set) var seekCount: Int = 0
 
+    /// Host-side parse carry (see `HostCarryTrim`). Bounded in `count` by construction: everything
+    /// but the partial trailing TS packet is consumed on every fill.
+    private let carryTrim: HostCarryTrim
+    private var carry = Data()
+    private(set) var carryCount = 0
+    /// The slice's lower bound, which is the whole tell: for a re-based carry it stays 0, for a
+    /// `removeFirst` one it equals every byte ever consumed, and the backing store is that large.
+    private(set) var carryStartIndex = 0
+
     init(path: String, rateKbps: Int, reportsSize: Bool, wraps: Bool,
-         foundationRead: Bool = false) throws {
+         foundationRead: Bool = false, carryTrim: HostCarryTrim = .none) throws {
         self.path = path
+        self.carryTrim = carryTrim
         let attrs = try FileManager.default.attributesOfItem(atPath: path)
         self.fileSize = (attrs[.size] as? NSNumber)?.int64Value ?? 0
         if foundationRead {
@@ -127,6 +155,7 @@ final class PacedLiveSpoolIOReader: IOReader, @unchecked Sendable {
                 }
                 position += Int64(got)
                 bytesRead += Int64(got)
+                feedCarryLocked(buffer, count: got)
                 lock.unlock()
                 return Int32(got)
             }
@@ -136,6 +165,23 @@ final class PacedLiveSpoolIOReader: IOReader, @unchecked Sendable {
         }
     }
 
+    /// Push the delivered bytes through the carry and consume whole TS packets, which is what a
+    /// PCR indexer on the ingest side does. Called under `lock`.
+    private func feedCarryLocked(_ buffer: UnsafeMutablePointer<UInt8>, count: Int) {
+        guard carryTrim != .none, count > 0 else { return }
+        carry.append(buffer, count: count)
+        let consumable = (carry.count / 188) * 188
+        if consumable > 0 {
+            switch carryTrim {
+            case .removeFirst: carry.removeFirst(consumable)
+            case .subdata:     carry = carry.subdata(in: consumable..<carry.count)
+            case .none:        break
+            }
+        }
+        carryCount = carry.count
+        carryStartIndex = carry.startIndex
+    }
+
     /// One upstream burst. 32 KB is the size the engine's own file reader is measured in (#243) and
     /// is large enough that the read cadence is set by the source, not by the pacer's resolution.
     private static let releaseChunk: Int64 = 32 * 1024
@@ -165,7 +211,7 @@ final class PacedLiveSpoolIOReader: IOReader, @unchecked Sendable {
     func makeIndependentReader() -> IOReader? {
         try? PacedLiveSpoolIOReader(path: path, rateKbps: Int(rateBytesPerSecond * 8.0 / 1000.0),
                                     reportsSize: reportsSize, wraps: wraps,
-                                    foundationRead: handle != nil)
+                                    foundationRead: handle != nil, carryTrim: carryTrim)
     }
 
     var discImageProbeEnabled: Bool { false }
@@ -184,20 +230,23 @@ final class PacedLiveSpoolIOReader: IOReader, @unchecked Sendable {
 /// graph. macOS has no jetsam, so the run cannot be killed here: the slope IS the finding.
 func runCustomLiveSpool(path: String, seconds: Double, rateKbps: Int, dvrWindow: Double?,
                         reportsSize: Bool, wraps: Bool, mallocCensus: Bool,
-                        foundationReader: Bool = false) -> Int32 {
+                        foundationReader: Bool = false, carryTrim: HostCarryTrim = .none) -> Int32 {
     EngineLog.handler = { print($0) }
     if mallocCensus {
-        AetherEngine.setLargeAllocationCensusEnabled(true, triggerThresholdMB: 32, triggerPollHz: 8)
+        // Uncapped captures: a steady mux-rate climb spends one capture per threshold climbed, so the
+        // default twelve are gone long before a long run ends (AE#445 hit the cap 4.4 min early).
+        AetherEngine.setLargeAllocationCensusEnabled(true, triggerThresholdMB: 32, triggerPollHz: 8,
+                                                     triggerCaptureCap: 0)
     }
     print("aetherctl customio --live: \(path) (rate=\(rateKbps) kbit/s seconds=\(seconds) "
           + "dvrWindow=\(dvrWindow.map { String($0) } ?? "nil") size=\(reportsSize ? "reported" : "unknown") "
           + "wrap=\(wraps) census=\(mallocCensus) "
-          + "reader=\(foundationReader ? "foundation" : "posix"))")
+          + "reader=\(foundationReader ? "foundation" : "posix") hostCarry=\(carryTrim.rawValue))")
     let box = UncheckedBox<Int32?>(nil)
     Task { @MainActor in
         box.value = await customLiveSpoolRun(path: path, seconds: seconds, rateKbps: rateKbps,
                                              dvrWindow: dvrWindow, reportsSize: reportsSize, wraps: wraps,
-                                             foundationReader: foundationReader)
+                                             foundationReader: foundationReader, carryTrim: carryTrim)
         CFRunLoopStop(CFRunLoopGetMain())
     }
     CFRunLoopRun()
@@ -206,12 +255,13 @@ func runCustomLiveSpool(path: String, seconds: Double, rateKbps: Int, dvrWindow:
 
 @MainActor
 private func customLiveSpoolRun(path: String, seconds: Double, rateKbps: Int, dvrWindow: Double?,
-                                reportsSize: Bool, wraps: Bool, foundationReader: Bool) async -> Int32 {
+                                reportsSize: Bool, wraps: Bool, foundationReader: Bool,
+                                carryTrim: HostCarryTrim) async -> Int32 {
     let reader: PacedLiveSpoolIOReader
     do {
         reader = try PacedLiveSpoolIOReader(path: path, rateKbps: rateKbps,
                                             reportsSize: reportsSize, wraps: wraps,
-                                            foundationRead: foundationReader)
+                                            foundationRead: foundationReader, carryTrim: carryTrim)
     } catch {
         print("VERDICT: reader init failed: \(error.localizedDescription)")
         return 1
@@ -255,8 +305,11 @@ private func customLiveSpoolRun(path: String, seconds: Double, rateKbps: Int, dv
             guard dt > 1 else { return "n/a" }
             return String(format: "%.2f", Double(fp - f.footprint) / dt)
         } ?? "n/a"
-        print(String(format: "  t=%.0fs state=%@ pos=%.2fs physFP=%dMB srcMB=%.1f growthMBps=%@",
-                     elapsed, "\(engine.state)", engine.currentTime, fp, srcMB, slope))
+        let carryLine = carryTrim == .none ? "" : String(
+            format: " carryCount=%dB carryStart=%.1fMB",
+            reader.carryCount, Double(reader.carryStartIndex) / 1_048_576.0)
+        print(String(format: "  t=%.0fs state=%@ pos=%.2fs physFP=%dMB srcMB=%.1f growthMBps=%@%@",
+                     elapsed, "\(engine.state)", engine.currentTime, fp, srcMB, slope, carryLine))
         if case .error(let msg) = engine.state {
             print("VERDICT: session errored: \(msg)")
             engine.stop()
@@ -269,6 +322,16 @@ private func customLiveSpoolRun(path: String, seconds: Double, rateKbps: Int, dv
     print(String(format: "LOOKBACK: %d seeks, deepest reach-back %.1f MB behind the live edge "
                  + "(%.0f s of source at this rate)",
                  reader.seekCount, lookbackMB, lookbackMB / srcMBps))
+    if carryTrim != .none {
+        // The verdict is the lower bound, not the count: a carry that starts at 0 owns exactly its
+        // count, and one whose start tracks the consumed stream owns all of it.
+        let startMB = Double(reader.carryStartIndex) / 1_048_576.0
+        let reading = reader.carryStartIndex > (1 << 20)
+            ? "riding a backing store that large"
+            : "re-based, so the allocation is the count"
+        print(String(format: "HOST CARRY (%@): count=%dB, slice lower bound %.1f MB: %@.",
+                     carryTrim.rawValue, reader.carryCount, startMB, reading))
+    }
     if let f = firstSample, let l = lastSample, l.t - f.t > 30 {
         let growth = Double(l.footprint - f.footprint) / (l.t - f.t)
         print(String(format: "VERDICT: physFP %d -> %d MB over %.0fs = %.2f MB/s "