a simple semaphore solution with cancel support (with sendable and without sendable)

custom_semaphore_cancel_support.py

import Foundation

// A custom error type for cancellation
enum TaskError: Error {
    case canceled
}

// A struct to manage tasks and handle cancellation
class ConcurrentTaskManager {
    
    // Semaphore to limit concurrent tasks
    private let semaphore: DispatchSemaphore
    private var isCancelled = false
    private let cancellationLock = NSLock()
    
    // Initializer takes a concurrency limit
    init(concurrencyLimit: Int) {
        self.semaphore = DispatchSemaphore(value: concurrencyLimit)
    }
    
    // Method to cancel all tasks
    func cancelAllTasks() {
        cancellationLock.lock()
        isCancelled = true
        cancellationLock.unlock()
    }
    
    // Run a block of code with concurrency control and cancellation option
    func runTask(name: String, task: @escaping () -> Void, completion: @escaping (Result<Void, Error>) -> Void) {
        // Check cancellation state before starting the task
        cancellationLock.lock()
        if isCancelled {
            cancellationLock.unlock()
            completion(.failure(TaskError.canceled))
            return
        }
        cancellationLock.unlock()
        
        // Wait for the semaphore (concurrency control)
        semaphore.wait()
        
        // Run the task asynchronously
        DispatchQueue.global(qos: .userInitiated).async {
            // Check for cancellation again before execution
            self.cancellationLock.lock()
            if self.isCancelled {
                self.cancellationLock.unlock()
                self.semaphore.signal()  // Signal the semaphore to release it
                completion(.failure(TaskError.canceled))
                return
            }
            self.cancellationLock.unlock()
            
            // Perform the task
            print("Starting task: \(name)")
            task()
            
            // Once the task is completed, signal the semaphore
            self.semaphore.signal()
            completion(.success(()))
            print("Task completed: \(name)")
        }
    }
}

// Usage example

// Create a task manager with a concurrency limit of 3
let taskManager = ConcurrentTaskManager(concurrencyLimit: 3)

// Example task function
func exampleTask(id: Int) {
    sleep(2)  // Simulate some work being done
    print("Task \(id) is complete.")
}

// Add tasks to the manager
taskManager.runTask(name: "Task 1", task: {
    exampleTask(id: 1)
}) { result in
    switch result {
    case .success:
        print("Task 1 finished successfully.")
    case .failure(let error):
        print("Task 1 failed with error: \(error)")
    }
}

taskManager.runTask(name: "Task 2", task: {
    exampleTask(id: 2)
}) { result in
    switch result {
    case .success:
        print("Task 2 finished successfully.")
    case .failure(let error):
        print("Task 2 failed with error: \(error)")
    }
}

taskManager.runTask(name: "Task 3", task: {
    exampleTask(id: 3)
}) { result in
    switch result {
    case .success:
        print("Task 3 finished successfully.")
    case .failure(let error):
        print("Task 3 failed with error: \(error)")
    }
}

// Cancel all tasks after 1 second
DispatchQueue.global().asyncAfter(deadline: .now() + 1) {
    print("Cancelling all tasks.")
    taskManager.cancelAllTasks()
}


//with Sendable Protocol

import Foundation

// Custom error type for task cancellations
enum TaskError: Error {
    case canceled
}

// A struct to manage tasks with concurrency control, and cancellation using Sendable
actor ConcurrentTaskManager {
    
    private var semaphore: DispatchSemaphore
    private var isCancelled = false
    private let cancellationLock = NSLock()
    
    // Initializer takes a concurrency limit
    init(concurrencyLimit: Int) {
        self.semaphore = DispatchSemaphore(value: concurrencyLimit)
    }
    
    // Method to cancel all tasks
    func cancelAllTasks() {
        cancellationLock.lock()
        isCancelled = true
        cancellationLock.unlock()
    }
    
    // Run a block of code with concurrency control and cancellation
    func runTask(name: String, task: @escaping @Sendable () async -> Void) async -> Result<Void, Error> {
        // Check cancellation state before starting the task
        cancellationLock.lock()
        if isCancelled {
            cancellationLock.unlock()
            return .failure(TaskError.canceled)
        }
        cancellationLock.unlock()
        
        // Wait for the semaphore (concurrency control)
        semaphore.wait()
        
        // Run the task asynchronously
        await Task.detached(priority: .userInitiated) {
            // Check for cancellation again before execution
            self.cancellationLock.lock()
            if self.isCancelled {
                self.cancellationLock.unlock()
                self.semaphore.signal()  // Release semaphore
                return
            }
            self.cancellationLock.unlock()
            
            // Perform the task
            print("Starting task: \(name)")
            await task()
            
            // Once the task is completed, signal the semaphore
            self.semaphore.signal()
            print("Task completed: \(name)")
        }
        
        return .success(())
    }
}

// Example task function
func exampleTask(id: Int) async {
    // Simulate some work being done
    await Task.sleep(2 * 1_000_000_000)  // Sleep for 2 seconds
    print("Task \(id) is complete.")
}

// Usage example

// Create a task manager with a concurrency limit of 3
let taskManager = ConcurrentTaskManager(concurrencyLimit: 3)

// Add tasks to the manager
Task {
    await taskManager.runTask(name: "Task 1", task: {
        await exampleTask(id: 1)
    })
    
    await taskManager.runTask(name: "Task 2", task: {
        await exampleTask(id: 2)
    })
    
    await taskManager.runTask(name: "Task 3", task: {
        await exampleTask(id: 3)
    })
    
    // Cancel all tasks after 1 second
    await Task.sleep(1 * 1_000_000_000)  // Sleep for 1 second
    print("Cancelling all tasks.")
    taskManager.cancelAllTasks()
}