// EXERCISE 06 — FIXED VARIANT.  The permit moves to the submission site, and a
// structured-concurrency version is measured beside it for comparison.
//
// THE SCENARIO
//   An importer reads 128 records through a synchronous, blocking API, with a
//   concurrency limit of 4.
//
// THE REPAIR — move the admission decision, do not change the primitive
//   The semaphore was never the problem and the limit was never wrong. The
//   PLACE was wrong.
//
//   Taking the permit inside the work item means all 128 items are already on
//   the queue, already started, already holding threads, and only then waiting.
//   Taking it before `async` means the SUBMITTING thread blocks instead, and
//   work that has not been admitted yet occupies nothing at all. At most
//   `limit` items are ever in flight, so at most `limit` threads are ever tied
//   up by this workload.
//
//   Note the asymmetry that makes this work: the permit is released by the work
//   item when it FINISHES, not by the submitter after it submits. A permit
//   returned at submission time would bound nothing.
//
//   The second half of this file measures the same bounded workload written
//   with a task group. That version is only available if the blocking call can
//   become `async`: `Task.sleep` SUSPENDS the task and releases the thread,
//   whereas `Thread.sleep` inside a Task would block a cooperative-pool thread
//   and is the forward-progress violation to avoid. The cooperative pool is
//   deliberately NOT overcommitting — it is sized to the core count — which is
//   why blocking inside a Task is a correctness problem rather than a style
//   preference.
//
// Build: swiftc -swift-version 6 -O ingest.swift -o ingest_fixed
// Run:   ./ingest_fixed
//
// Expected: peakThreads close to the core count rather than several times it,
// permitHolders still 4, and a structured-concurrency line showing the same
// bound reached without blocking a thread at all. Exit status 0.
//
// Bounded: fixed item count, fixed 120 ms block, and a watchdog that
// force-exits after WATCHDOG_S.

import Foundation
import Darwin
import Synchronization

let WATCHDOG_S: Double = 120
let EXIT_WATCHDOG: Int32 = 75

func armWatchdog(_ seconds: Double, _ label: String) {
    let t = Thread {
        Thread.sleep(forTimeInterval: seconds)
        fputs("\nWATCHDOG: \(label) exceeded \(seconds)s — forcing _exit(\(EXIT_WATCHDOG)).\n", stderr)
        fflush(stderr); _exit(EXIT_WATCHDOG)
    }
    t.stackSize = 512 * 1024; t.start()
}

/// Live kernel threads in this process, straight from Mach.
func liveThreadCount() -> Int {
    var list: thread_act_array_t?
    var count: mach_msg_type_number_t = 0
    guard task_threads(mach_task_self_, &list, &count) == KERN_SUCCESS, let list else { return -1 }
    defer {
        for i in 0..<Int(count) { mach_port_deallocate(mach_task_self_, list[i]) }
        vm_deallocate(mach_task_self_,
                      vm_address_t(UInt(bitPattern: list)),
                      vm_size_t(Int(count) * MemoryLayout<thread_t>.size))
    }
    return Int(count)
}

/// Samples the live thread count on a timer and keeps the high-water mark.
final class ThreadMeter: @unchecked Sendable {
    private let peak = Mutex<Int>(0)
    private let stopped = Mutex<Bool>(false)

    func start() {
        let t = Thread { [self] in
            while !(stopped.withLock { $0 }) {
                let n = liveThreadCount()
                peak.withLock { if n > $0 { $0 = n } }
                Thread.sleep(forTimeInterval: 0.01)
            }
        }
        t.stackSize = 512 * 1024
        t.start()
    }

    func finish() -> Int {
        stopped.withLock { $0 = true }
        Thread.sleep(forTimeInterval: 0.05)
        return peak.withLock { $0 }
    }
}

/// Tracks how many work items are inside the permit at once, so the run can
/// show that the semaphore IS working — it is just working in the wrong place.
final class ConcurrencyMeter: @unchecked Sendable {
    private let state = Mutex(State())
    struct State { var current = 0; var peak = 0 }
    func enter() { state.withLock { $0.current += 1; if $0.current > $0.peak { $0.peak = $0.current } } }
    func leave() { state.withLock { $0.current -= 1 } }
    var peak: Int { state.withLock { $0.peak } }
}

let cores = ProcessInfo.processInfo.activeProcessorCount
let items = 128
let blockMs = 120
let limit = 4

// THE REPAIR
//
// The permit is taken BEFORE the work is submitted, so it throttles this one
// submitting thread. Unadmitted work is not on the queue, has not started, and
// holds no thread. The permit is returned by the work item when it completes.
//
// This function is deliberately NOT async. Swift 6 refuses `DispatchSemaphore
// .wait()` in an asynchronous context outright — "unavailable from
// asynchronous contexts" — so blocking code of this shape has to live in a
// synchronous function whatever else is true of the program.
func runImport() -> (baseline: Int, peak: Int, holders: Int, elapsedMs: Int) {
    let meter = ThreadMeter(); meter.start()
    let holders = ConcurrencyMeter()
    let baseline = liveThreadCount()
    let start = Date()

    let gate = DispatchSemaphore(value: limit)
    let group = DispatchGroup()

    for _ in 0..<items {
        gate.wait()
        DispatchQueue.global(qos: .utility).async(group: group) {
            holders.enter()
            Thread.sleep(forTimeInterval: Double(blockMs) / 1000.0)   // the blocking API
            holders.leave()
            gate.signal()
        }
    }
    group.wait()

    let elapsedMs = Int(Date().timeIntervalSince(start) * 1000)
    return (baseline, meter.finish(), holders.peak, elapsedMs)
}

/// The same bounded workload, written with structured concurrency.
///
/// The group is kept at `limit` in flight by awaiting one completion before
/// adding the next task — the idiomatic bounded task-group shape. No semaphore,
/// no blocked thread, and the bound is visible in the control flow.
func structuredIngest() async -> (peak: Int, elapsedMs: Int) {
    let meter = ThreadMeter(); meter.start()
    let start = Date()
    await withTaskGroup(of: Void.self) { group in
        var submitted = 0
        for _ in 0..<min(limit, items) {
            group.addTask { try? await Task.sleep(for: .milliseconds(blockMs)) }
            submitted += 1
        }
        while submitted < items {
            await group.next()                    // wait for one to finish...
            group.addTask { try? await Task.sleep(for: .milliseconds(blockMs)) }
            submitted += 1
        }
    }
    let elapsedMs = Int(Date().timeIntervalSince(start) * 1000)
    return (meter.finish(), elapsedMs)
}

armWatchdog(WATCHDOG_S, "EX06 build=fixed")

let r = runImport()

print("EX06 build=fixed items=\(items) limit=\(limit) blockMs=\(blockMs) cores=\(cores) "
    + "baselineThreads=\(r.baseline) peakThreads=\(r.peak) "
    + "threadsPerCore=\(String(format: "%.1f", Double(r.peak) / Double(cores))) "
    + "permitHolders=\(r.holders) elapsedMs=\(r.elapsedMs)")

let structured = await structuredIngest()

print("EX06 build=fixed variant=structured items=\(items) limit=\(limit) "
    + "blockMs=\(blockMs) cores=\(cores) structuredPeakThreads=\(structured.peak) "
    + "structuredThreadsPerCore=\(String(format: "%.1f", Double(structured.peak) / Double(cores))) "
    + "structuredElapsedMs=\(structured.elapsedMs)")