https://stackoverflow.com/q/79876948/1271826

PlaygroundDemo.swift

import Foundation
import PlaygroundSupport

PlaygroundPage.current.needsIndefiniteExecution = true

class Demo {
    func getAsyncStream() -> AsyncStream<Int> {
        AsyncStream { continuation in
            let task = Task {
                for i in 1...5 {
                    try await Task.sleep(for: .seconds(1))
                    continuation.yield(i)
                }
                continuation.finish()
            } 
            
            continuation.onTermination = { _ in
                task.cancel()
            }
        }
    }
    
    init() { print("initialised") }
    deinit { print("deinitialised") }
}

do {
    let demo = Demo()
    for await value in demo.getAsyncStream() {
        print(value)
    }
}

PlaygroundPage.current.finishExecution() // see https://stackoverflow.com/questions/78556983/async-await-unexpected-error-in-playground

@Pramod-36

To download these images concurrently, we need to create a separate task for each image download.

Yes, we would want a separate task (lowercase “t”) for each. But we would avoid introducing unnecessary unstructured concurrency (a Task {…}), but rather remain within structured concurrency. It keeps life much simpler.

Assume we have an array of images—say 20 images—each with different sizes and in a random order. Rather than using sleep, each image would be downloaded from the network, and every download would take a different amount of time.

The idiomatic solution would be a “task group”:

1. It won’t complete until all the child tasks are done. There is no “counter” to keep track of.
2. It remains with structured concurrency (meaning that if you cancel the task group, all the child tasks will be canceled for you).

Thus, perhaps a simple solution would look like:

class SimpleImageDownloader {
    func images(for urls: [URL]) -> AsyncStream<UIImage> {
        AsyncStream { continuation in
            let task = Task {
                await withTaskGroup(of: UIImage.self) { group in
                    // create child tasks

                    for url in urls {
                        group.addTask { await self.image(for: url) }
                    }

                    // await and yield the results (in the order that they complete)
                    
                    for await image in group {
                        continuation.yield(image)
                    }

                    // finish the stream
                    
                    continuation.finish()
                }                
            } 
            
            continuation.onTermination = { state in
                if case .cancelled = state {
                    task.cancel()
                }
            }
        }
    }
    
    func image(for url: URL) async -> UIImage {…}
}

var urls: [URL] = …
let downloader = SimpleImageDownloader()
for await image in downloader.images(for: urls) {
    // do something with image
}

---

Now, that having been said, there are two limitations in the above:

1. You don’t know which image is associated with which URL (because, as you say, they may complete in any order). We might, for example, use a tuple, instead, that includes the original URL, so we can correlate the image yielded by the sequence with our original sequence of URLs.
2. We’re not handling errors. We would generally expect image(for:) to throw errors (e.g., a network request fails) and then the asynchronous sequence would yield a Result<UIImage, any Error>, instead, so the caller knows which images succeeded and which failed (while not stopping just because one of the images failed).

So, perhaps:

class BetterImageDownloader {
    func images(for urls: [URL]) -> AsyncStream<(URL, Result<UIImage, any Error>)> {
        AsyncStream { continuation in
            let task = Task {
                await withTaskGroup(of: (URL, Result<UIImage, any Error>).self) { group in
                    for url in urls {
                        group.addTask {
                            do {
                                let image = try await self.image(for: url)
                                return (url, .success(image))
                            } catch {
                                return (url, .failure(error))
                            }
                        }
                    }
                    
                    for await urlAndResult in group {
                        continuation.yield(urlAndResult)
                    }
                    
                    continuation.finish()
                }                
            } 
            
            continuation.onTermination = { state in
                if case .cancelled = state {
                    task.cancel()
                }
            }
        }
    }
    
    func image(for url: URL) async throws -> UIImage {…}
}

var urls: [URL] = …
let downloader = BetterImageDownloader()
for await (url, result) in downloader.images(for: urls) {
    switch result {
    case .success(let image):
        // do something with image
    case .failure(let error):
        // do something with error
    }
}

This enjoys automatic tracking of when the collection of asynchronous tasks is complete, while remaining in a pattern that supports cancellation of the whole sequence.

---

If you want to go nuts, I might suggest a few more refinements: Notably, the above is fine if you have a dozen images. But what if you have 1,000? You really don’t want to create an unbridled number of tasks, but rather constrain it to some reasonable number that both enjoys concurrency, but limits it to some reasonable number. People often use something like 6 or 10 concurrent requests, because beyond that, you don’t see discernible performance benefits, but do suffer memory issues or overloading the system with too many tasks.

So, you might end up with something like the following, where we will await group.next() above a certain number. This will constrain the degree of concurrency to no more than n requests at a time:

class ImageDownloader {
    func images(for urls: [URL], maxConcurrent: Int = 6) -> AsyncStream<(URL, Result<UIImage, any Error>)> {
        precondition(maxConcurrent > 0)

        return AsyncStream { continuation in
            let task = Task {
                await withTaskGroup(of: (URL, Result<UIImage, any Error>).self) { group in
                    for (index, url) in urls.enumerated() {
                        if index >= maxConcurrent, let urlAndResult = await group.next() {
                            continuation.yield(urlAndResult)
                        }
                        
                        group.addTaskUnlessCancelled {
                            do {
                                let image = try await self.image(for: url)
                                return (url, .success(image))
                            } catch {
                                return (url, .failure(error))
                            }
                        }
                    }
                    
                    for await urlAndResult in group {
                        continuation.yield(urlAndResult)
                    }
                    
                    continuation.finish()
                }                
            } 
            
            continuation.onTermination = { state in
                if case .cancelled = state {
                    task.cancel()
                }
            }
        }
    }
    
    func image(for url: URL) async throws -> UIImage {
        let (data, response) = try await URLSession.shared.data(from: url)
        
        if let httpResponse = response as? HTTPURLResponse, !((200..<300) ~= httpResponse.statusCode) {
            throw DownloaderError.httpError(httpResponse.statusCode)
        }
        
        guard let image = UIImage(data: data) else {
            throw DownloaderError.notImage
        }
        
        return image
    }
    
    enum DownloaderError: LocalizedError {
        case httpError(Int)
        case notImage
        
        var errorDescription: String {
            return switch self {
            case .httpError: NSLocalizedString("Web server error", comment: "Downloader error")
            case .notImage: NSLocalizedString("Not image", comment: "Downloader error")
            }
        }
    }
}

There are additional refinements one might add, but this strikes me as a good starting point. But the key is:

• Remain within structured concurrency so that you can gracefully cancel all pending child tasks if the overall sequence is cancelled.
• Use task group to simplify the process of knowing when a series of child tasks finish.

FWIW, I have not done exhaustive testing of the above, so I apologize for any errors herein. But hopefully it illustrates the idea.

Hi @robertmryan , thanks for your reply.

I now have a much better understanding of concurrency, especially the difference between structured and unstructured concurrency.

I have one small doubt regarding the final solution that limits the number of concurrent tasks. This may be a minor concern, but I wanted to clarify my understanding.

Let’s say we have 5 images and maxConcurrent = 2.
The first two images take 10 seconds and 7 seconds to download.
The remaining three images each take 1 second.

With the current logic:

if index >= maxConcurrent, let urlAndResult = await group.next() {
    continuation.yield(urlAndResult)
}

When processing the3 image (index == maxConcurrent), the code waits for one of the first two tasks to complete. In this example, that would be the second image (7 seconds). Only after that completes is the third image added to the task group. The same pattern repeats for the fourth image, and so on.

From a user-experience perspective, this means the user would not see the third image until 7 seconds, even though it only takes 1 second to download. In contrast, if we did not limit concurrency, the third image could be displayed almost immediately.

what might be taken into considration User experience or memory issue that may occur? (This would happen even with 1,000 tasks—some quick images could still end up waiting on slower ones.)

Sample code

import Foundation
import PlaygroundSupport
PlaygroundPage.current.needsIndefiniteExecution = true

class ImageInverter {
    func convertToInvertedImage(_ image: Image) async -> Image {
        try? await Task.sleep(for: .seconds(image.timeToDownload))
        return image
    }
    
    func process(maxConcurrentTasks: Int = 2, _ images: [Image]) -> AsyncStream<Image>{
        AsyncStream { continuation in
            let task = Task {
                await withTaskGroup(of: Image.self) { group in
                    for (index, image) in images.enumerated() {
                        if index >= maxConcurrentTasks, let invertedImage = await group.next(){
                            continuation.yield(invertedImage)
                        }
                        group.addTask { [weak self] in
                            let invertedImage = await self!.convertToInvertedImage(image)
                            return invertedImage
                        }
                    }
                    for await invertedImage in group {
                        continuation.yield(invertedImage)
                    }
                    continuation.finish()
                }
            }
            continuation.onTermination = { state in
                print("Stream termination state:", state)
                task.cancel()
            }
        }
    }
}

struct Image {
    let name: String
    let timeToDownload: Double
}
let images: [Image] = [
    .init(name: "1", timeToDownload: 6),
    .init(name: "2", timeToDownload: 9),
    .init(name: "3", timeToDownload: 1),
    .init(name: "4", timeToDownload: 2),
    .init(name: "5", timeToDownload: 3)
]
let imageInverter = ImageInverter()
let startDate = Date.now
for await invertedImage in imageInverter.process(maxConcurrentTasks: 2, images) {
    print("Recieved \(invertedImage) in in \(Date.now.timeIntervalSince(startDate)) seconds")
}
PlaygroundPage.current.finishExecution()

@Pramod-36 … Yep, constraining concurrency has trade-offs. And if you don’t want to limit the degree of concurrency, then don’t: It’s entirely up to you. But I’d suggest that when you start dealing with massive parallelism (like thousands of tasks), there are all sorts of practical considerations that start to crop up, and you might want to revisit whether it might be prudent to constrain the degree of something reasonable for your problem:

1. You give an example of an “image inverter”, which is likely to be CPU-bound, anyway. So, if you start 1,000 image inversions, the number of available CPU cores will constrain you, anyway. So I can create 1,000 tasks, but if my Mac has 20 cores, it can only do 20 CPU-bound tasks at a time, regardless, and the other 9,980 tasks will just be sitting there, taking up memory, waiting for a thread from the concurrency thread pool (a.k.a., the “cooperative thread pool”), anyway. So, even if the first 999 tasks take 10 seconds each and the last one takes 0.1 seconds, you simply won’t see that last one start until it is afforded one of the very limited CPU cores, even if you don’t constrain the degree of concurrency yourself.

   As an aside, when doing a vast number of parallel CPU-bound tasks, we often introduce “striding” (so that rather than 1,000 individual tasks, we might do 40 tasks, each processing 25 items); parallelism isn’t free and each task has a little overhead, so we often stride to maximize performance. It often takes some experimentation to empirically identify the ideal striding factor for the particular use case. Personally, I often start with something like 2 or 4 times the number of processors, and then benchmark the various alternatives.

2. You focus on how long it takes to finish the fastest task. Often, we focus on how long it takes to finish all of the tasks. E.g., if downloading images, affording 6 concurrent requests will dramatically improve the performance. Allowing 10 at a time will marginally improve it. But beyond that, you have diminishing returns while introducing new problems (like latter network requests failing due to timeouts, excessive app peak memory usage, excessive server load, etc.).

3. If you are showing results in the UI, you would generally want it to favor those tasks associated with images that are visible. We wouldn’t want to sacrifice the speed for the user-visible images simply to maximize the speed for which it processes the 1,000th image.

Given your hesitancy, feel free to defer this “constrained degree of concurrency” question until you start to experience memory and performance problems. But I wager you will eventually encounter situations where you might want to revisit this.

FWIW, see Beyond the basics of structured concurrency for a discussion on constraining concurrency.