v2: Initial clean branch with unit tests + CI/CD pipeline

- Started from ac75faa (initial E4B-MarkBase integration) - Kept Sources/ (all engine code) + Package.swift + .gitignore - Removed all ad-hoc tests, documentation, scripts, Python files - Added Tests/00_Unit/ (MathTest, TokenizerTest, SamplerTest) - Added .gitea/workflows/ci.yaml (build + unit tests + lint) - Added Scripts/check_resources.sh (memory-aware test runner) - Added Tests/Manifest.json (resource requirements for all tests) - Focus: 4-bit quantized models only
2026-07-05 13:29:25 +08:00
commit 8a66b9086a
90 changed files with 22252 additions and 0 deletions
@@ -0,0 +1,106 @@
+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 }
+        }
+    }
+}