Swift Threading and Concurrency: Parallel Processing

import Foundation

class ThreadManager {
    func performOnBackgroundThread() {
        DispatchQueue.global(qos: .background).async {
            print("Background thread: \(Thread.current)")
        }
    }

    func performOnMainThread() {
        DispatchQueue.main.async {
            print("Main thread: \(Thread.current)")
        }
    }

    func performAfterDelay(_ delay: TimeInterval, block: @escaping () -> Void) {
        DispatchQueue.main.asyncAfter(deadline: .now() + delay) {
            block()
        }
    }
}

class QueueManager {
    private let serialQueue = DispatchQueue(label: "com.app.serial")
    private let concurrentQueue = DispatchQueue(label: "com.app.concurrent", attributes: .concurrent)

    func performTasksSerially() {
        for i in 1...5 {
            serialQueue.async {
                print("Task \(i) on serial queue")
            }
        }
    }

    func performTasksConcurrently() {
        for i in 1...5 {
            concurrentQueue.async {
                print("Task \(i) on concurrent queue")
            }
        }
    }

    func waitForTasks() {
        let dispatchGroup = DispatchGroup()

        for i in 1...3 {
            dispatchGroup.enter()
            DispatchQueue.global().async {
                print("Task \(i)")
                dispatchGroup.leave()
            }
        }

        dispatchGroup.notify(queue: .main) {
            print("All tasks completed")
        }
    }
}

class ThreadSafeCounter {
    private var count = 0
    private let queue = DispatchQueue(label: "com.app.counter")

    func increment() {
        queue.sync {
            count += 1
        }
    }

    func getCount() -> Int {
        return queue.sync {
            count
        }
    }

    func reset() {
        queue.sync {
            count = 0
        }
    }
}

class OperationManager {
    private let operationQueue: OperationQueue

    init() {
        operationQueue = OperationQueue()
        operationQueue.maxConcurrentOperationCount = 2
        operationQueue.qualityOfService = .userInitiated
    }

    func addOperations() {
        for i in 1...10 {
            let operation = BlockOperation {
                print("Operation \(i) started")

                Thread.sleep(forTimeInterval: 1)

                print("Operation \(i) completed")
            }

            operation.queuePriority = .medium
            operationQueue.addOperation(operation)
        }
    }

    func cancelAllOperations() {
        operationQueue.cancelAllOperations()
    }
}

import ThreadSanitizer

class ThreadSafeArray {
    private var array: [Int] = []
    private let queue = DispatchQueue(label: "com.app.array")

    func append(_ element: Int) {
        queue.sync {
            array.append(element)
        }
    }

    func getElements() -> [Int] {
        return queue.sync {
            array
        }
    }
}

import Combine

class ReactiveProcessor {
    private let subject = PassthroughSubject<Data, Never>()
    private var cancellables = Set<AnyCancellable>()

    func processData() {
        subject
            .receive(on: DispatchQueue.global(qos: .userInitiated))
            .flatMap { data in
                self.parseData(data)
            }
            .receive(on: DispatchQueue.main)
            .sink { result in
                print("Result: \(result)")
            }
            .store(in: &cancellables)
    }

    private func parseData(_ data: Data) -> AnyPublisher<String, Never> {
        // Parse data on background thread
        return Just("Parsed data")
    }
}

class AsyncStreamManager {
    func processAsyncStream() async throws {
        let stream = AsyncStream<Int> { continuation in
            for i in 1...10 {
                continuation.yield(i)
                try? await Task.sleep(nanoseconds: 1_000_000_000)
            }
            continuation.finish()
        }

        for await number in stream {
            print("Number: \(number)")
        }
    }

    func createAsyncStream() -> AsyncStream<String> {
        return AsyncStream { continuation in
            continuation.yield("Message 1")
            try? await Task.sleep(nanoseconds: 1_000_000_000)
            continuation.yield("Message 2")
            continuation.finish()
        }
    }
}

class LockedResource {
    private var resource: String = ""
    private let lock = NSLock()

    func updateResource(_ newValue: String) {
        lock.lock()
        resource = newValue
        lock.unlock()
    }

    func readResource() -> String {
        lock.lock()
        defer { lock.unlock() }
        return resource
    }
}

class ThreadPoolManager {
    private var threadPool: [Thread] = []
    private let maxThreads = 4

    init() {
        createThreadPool()
    }

    private func createThreadPool() {
        for i in 0..<maxThreads {
            let thread = Thread {
                while true {
                    // Wait for tasks
                    Thread.sleep(forTimeInterval: 1)
                }
            }
            thread.start()
            threadPool.append(thread)
        }
    }

    func submitTask(_ task: @escaping () -> Void) {
        // Submit task to thread
        Task {
            task()
        }.start()
    }
}