// EXERCISE 06 — BROKEN STARTING POINT.  Do not edit this file; copy it.
//
// ============================ UNSAFE CODE WARNING ============================
// This program deliberately blocks many work items on an overcommitting global
// queue. It is bounded — a fixed item count, a fixed block duration, and a
// watchdog — but it will briefly create dozens of kernel threads. Never
// dispatch blocking work this way in a real app.
// =============================================================================
//
// THE SCENARIO
//   An importer reads 128 records through a synchronous, blocking API. Someone
//   noticed the thread count exploding under Activity Monitor and added a
//   DispatchSemaphore "to limit concurrency to 4". The thread count did not
//   change. The semaphore is in the code, the limit is right there, and it does
//   nothing at all.
//
// WHAT THIS PROGRAM MEASURES
//   Live kernel threads in this process, sampled every 10 ms from
//   task_threads() — the same number Activity Monitor's "Threads" column shows.
//   The reported figure is the high-water mark over the run.
//
// THE MECHANISM TO EXPLAIN
//   A Dispatch global queue is OVERCOMMITTING. When a work item blocks,
//   libdispatch cannot distinguish "blocked" from "slow", so to keep the
//   queue's width occupied it brings up another thread. Blocking work items
//   therefore convert directly into threads, each with its own stack and its
//   own share of the scheduler.
//
// Build: swiftc -swift-version 6 -O ingest.swift -o ingest_broken
// Run:   ./ingest_broken
//
// Expected: peakThreads several times the core count, and permitHolders never
// above 4 — the permit really is limiting something, just not the thing that
// matters. 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 DEFECT
//
// The permit is taken INSIDE the work item. All 128 items are handed to the
// queue immediately; libdispatch starts them; each one then blocks — first on
// the semaphore, later on the work itself. A blocked work item still owns a
// thread, so the pool grows to cover them all.
//
// 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 {
        DispatchQueue.global(qos: .utility).async(group: group) {
            gate.wait()
            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)
}

armWatchdog(WATCHDOG_S, "EX06 build=broken")

let r = runImport()

print("EX06 build=broken 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)")