async Task 를 Serial 하게 실행하는 방법

[HoJongE]

Task(priority: .userInitiated) {
    for _ in 0...100_0000 {
        print("First")
    }
}

Task(priority: .userInitiated) {
    for _ in 0...100_0000 {
        print("Second")
    }
}

RunLoop.current.run(until: .now.addingTimeInterval(3000))

• 당연히 First 와 Second 가 섞여서 나오게 된다.

queue.async {
//    Task(priority: .userInitiated) {
        for _ in 0...100_00 {
            print("First")
        }
//    }
}

queue.async {
//    Task(priority: .userInitiated) {
        for _ in 0...100_00 {
            print("Second")
        }
//    }

}

• 위 코드에서 Task 로 async 한 context 를 제공해주지 않으면 당연히 첫 번째 print 가 끝난 뒤 두 번째 print 가 실행된다!
• 그러나 Task 로 실행하면 serial queue 에 들어가있더라도 병렬적으로 실행됨

References

https://stackoverflow.com/questions/70540541/swift-5-5-concurrency-how-to-serialize-async-tasks-to-replace-an-operationqueue

actor SerialTasks<Success> {
    private var previousTask: Task<Success, Error>?

    func add(block: @Sendable @escaping () async throws -> Success) {
        previousTask = Task { [previousTask] in
            let _ = await previousTask?.result
            return try await block()
        }
    }
}

async task 를 serial 하게 실행하는 방법으로, actor 를 활용해 previous Task 가 끝날 때 까지 기다리고 다음에 추가된 task 를 실행하는 방법인듯! 조금 더 공부를 해봐야 할 것 같음.

• 9b2f89d 커밋에서 실제로 serial 하게 동작하는지에 대한 테스트를 추가하였다!

• SerialTasksTests 파일에서 arr 에 순차적으로 0부터 1000 까지 데이터가 추가되는지에 대한 테스트를 작성함 -> 0부터 499, 500부터 1000까지 추가하는 async Task 를 두 개 만들고, 이것을 전부 serial Task Queue 에다가 추가함 -> 0부터 1000까지 정상적으로 출력됨

• 0부터 1000까지 데이터가 순차적으로 추가되지않는 상황에 대해 테스트를 작성함 -> 0부터 499, 500부터 1000까지 추가하는 async task 를 두 개 만들고, 이것을 그대로 병렬적으로 실행시킴으로써 0부터 1000까지 순차적으로 추가되지 않는 테스트를 작성함 -> actor 를 활용해 shared mutable state 인 arr 에 대한 race condition 을 방지함

---- 14:57 -----

• 그러나 위의 방법도 완벽한 sync task 를 지원하는 것은 아닌듯 하다. 왜냐하면 serial Task 에 task 를 추가하기 위해선 또 다른 async context 가 필요한데, 이에 Task 를 생성해야함. 그러나 Task 자체가 synchronous 하게 생성되지 않기 때문에, 원하는 결과를 얻지 못했다....

---- 15:48 -----

final class SerialTasksDispatchQueue {
    private let queue: DispatchQueue
    private let semaphore: DispatchSemaphore = .init(value: 1)

    init(_ qos: DispatchQoS = .userInteractive) {
        queue = DispatchQueue(label: "Serial Queue", qos: qos)
    }

    func addTask(_ block: @Sendable @escaping () async throws -> Void) {
        queue.async { [weak self] in
            guard let self else { return }
            self.semaphore.wait()
            Task(priority: .userInitiated) {
                defer { self.semaphore.signal() }
                try await block()
            }
        }
    }
}

• Serial Queue 를 만들어서 그 안에 async task 를 집어 넣음
• 그러나 Task 생성 및 Task 실행 코드는 또 다른 Thread 에서 실행되기 때문에, semaphore 를 사용해서 이전 task 가 끝나기 전까진 다음 task 를 추가할 수 없도록 구현하였다.