markbaseengine/Sources/MarkBaseServer/ConcurrentSupport.swift

import Foundation
import os

// ─────────────────────────────────────────────────────────────
// Concurrent Request Handling with Dynamic Concurrency
// ─────────────────────────────────────────────────────────────

/// Request queue for concurrent processing with dynamic concurrency
public actor RequestQueue {
    private let controller: DynamicConcurrencyController
    private var semaphore: AsyncSemaphore
    
    public init(controller: DynamicConcurrencyController) {
        self.controller = controller
        self.semaphore = AsyncSemaphore(value: 4)
    }
    
    public func initialize() async {
        let maxConcurrency = await controller.getCurrentMax()
        semaphore = AsyncSemaphore(value: maxConcurrency)
    }
    
    /// Execute request with concurrency limit
    public func execute<T: Sendable>(_ operation: @escaping @Sendable () async throws -> T) async throws -> T {
        await semaphore.wait()
        defer { semaphore.signal() }
        
        return try await operation()
    }
    
    /// Update semaphore when concurrency changes
    public func updateSemaphore(newMax: Int) {
        semaphore = AsyncSemaphore(value: newMax)
    }
}

/// Async semaphore for concurrency control
public final class AsyncSemaphore: @unchecked Sendable {
    private var value: Int
    private let lock = OSAllocatedUnfairLock()
    private var waiters: [CheckedContinuation<Void, Never>] = []
    
    public init(value: Int) {
        self.value = value
    }
    
    public func wait() async {
        let shouldWait = lock.withLock {
            if value > 0 {
                value -= 1
                return false
            }
            return true
        }
        guard shouldWait else { return }
        
        return await withCheckedContinuation { continuation in
            lock.withLock {
                waiters.append(continuation)
            }
        }
    }
    
    public func signal() {
        lock.withLock {
            if !waiters.isEmpty {
                let waiter = waiters.removeFirst()
                waiter.resume()
            } else {
                value += 1
            }
        }
    }
}

/// Batch request processor with dynamic concurrency
public actor BatchProcessor {
    private let requestQueue: RequestQueue
    
    public init(controller: DynamicConcurrencyController) {
        self.requestQueue = RequestQueue(controller: controller)
    }
    
    /// Process multiple requests concurrently
    public func processBatch<T: Sendable>(
        _ requests: [@Sendable () async throws -> T]
    ) async throws -> [T] {
        try await withThrowingTaskGroup(of: (Int, T).self) { group in
            var results: [T?] = Array(repeating: nil, count: requests.count)
            
            for (index, request) in requests.enumerated() {
                let req = request
                group.addTask {
                    let result = try await self.requestQueue.execute(req)
                    return (index, result)
                }
            }
            
            for try await (index, result) in group {
                results[index] = result
            }
            
            return results.compactMap { $0 }
        }
    }
}