Dictionary of semaphores

[smithi01]

Hi. I am finding Semaphore useful in so many ways in my apps. Thank you Gwendal. In one part of my app I need to download a large number of images from URLs supplied externally. URLs in the list can often be duplicates and I cache the results. I use a dictionary of AsyncSemaphore to avoid downloading from the same URL simultaneously. All the code here is within an image service actor.

private var urlSemaphores = [URL: AsyncSemaphore]()

I have reviewed the code in the package, which is well commented (thank you). I want to be sure that I don't deallocate the semaphore while the suspensions array is not empty. If this happens, it looks like my production code would crash.

I use code as follows, which I think is safe. Note I use another semaphore to limit total simultaneous downloads to 8.

    private let downloadSemaphore = AsyncSemaphore(value: 8)

    func fetchImage(from url: URL) async -> ImageData? {
        
        // See if there is a task in progress to download from the URL
        let semaphore = urlSemaphores[url] ?? AsyncSemaphore(value: 1)
        urlSemaphores[url] = semaphore
        await semaphore.wait()
        defer {
            let otherFetchesForThisUrlWereWaiting = semaphore.signal()
            if !otherFetchesForThisUrlWereWaiting {
                urlSemaphores[url] = nil
            }
        }
        if Task.isCancelled { return nil }
        
        // Check the cache
        if let encodedData = await urlCache.value(forUrl: url), let imageData = ImageData(fromEncodedData: encodedData) {
            guard !imageData.isNoImage else { return nil }
            return imageData
        }
        
        // Download the image
        await downloadSemaphore.wait()
        defer { downloadSemaphore.signal() }
        if Task.isCancelled { return nil }
        
        do {
            let (data, mime, _) = try await  downloadData(from: url)
        // Rest of function

Assuming this is ok, my question comes because I want to improve handling fetchImage being cancelled. However, I am unclear about using waitUnlessCancelled() for a semaphore in the dictionary. The semaphore seems to effectively get signalled (value += 1) when cancelled, but I don't see where in the code the next suspension is popped and told to resume in this circumstance. I also have no return value like I do with signal(), so I don't know if I can set urlSemaphores[url] to nil.

Sometimes when I have these sort of issues, it means I am heading down the wrong track with my implementation. I would be most grateful for any comments by the community.

Cheers,
Ian

[groue]

Hello @smithi01,

I want to be sure that I don't deallocate the semaphore while the suspensions array is not empty. If this happens, it looks like my production code would crash.

Yes, it is a programmer error to deallocate a semaphore that a task is waiting for. Nothing could make it continue, and this would create a memory leak.

The semaphore seems to effectively get signalled (value += 1) when cancelled, but I don't see where in the code the next suspension is popped and told to resume in this circumstance.

Cancellation does not resume anything. Signals do. Imagine two tasks that are actually blocked on waitUnlessCancelled(). They both are waiting for signal(). When the first task is cancelled, waitUnlessCancelled() throws CancellationError, and... the second task is still waiting for a signal. When the signal eventually occurs, the second task is resumed.

I also have no return value like I do with signal(), so I don't know if I can set urlSemaphores[url] to nil.

If waitUnlessCancelled() throws CancellationError, it means that it is was blocked waiting for some other task to finish downloading the image, or that the current task is already cancelled and that the semaphore did not even attempt at waiting.

In the first case, you do not want to destroy the semaphore, right? In the second case, it is ambiguous: you don't know if other tasks are waiting or not, and if you can destroy the semaphore.

Checking Task.isCancelled before waiting is racy: maybe the task will be cancelled between this check and the call to waitUnlessCancelled().

I don't have any solution that pops out of my mind at the present time. Probably some kind of counter is necessary? What do you think?

[smithi01]

Hi Gwendal. I think your answer has rearranged things in my mind, so I can understand better now. I was thrown off by the value +=1 in the waitUnlessCancelled code, but I see now that this is not a signal.

I see that it is essential that if a task calls either wait() or waitUnlessCancelled(), that it signals at some point after that. Hence the example code (and my code) that sets up a defer to do this.

In the case of my code using a dictionary of semaphores, the defer clock also checks the result of the signal and if no other tasks were waiting, sets the dictionary entry to nil, which will usually cause deallocation. I now see that this code has a race risk if another task starts waiting on that semaphore between the call to signal and me deallocating. I need to prevent a new wait while the defer block runs. (Perhaps I could use another AsyncSemaphore for this - haha)

If this race risk is dealt with, I think the code would work fine as it is. The only inefficiency is that if a task is cancelled it will remain waiting unnecessarily. When it finally resumes when another task signals, the next line "if Task.isCanceleld { return }" will cause the defer block to run.

However, a bunch of tasks may be waiting unnecessarily when they are cancelled, consuming memory. If I change sempaphore.wait() to try? semaphore.waitUnlessCancelled(), then if cancelled the await will finish and the next line will cause the defer block to run signalling and deallocating the semaphore in the dictionary if appropriate. ( A try catch block could do the same thing but I always check Task.isCancelled after a wait anyway.)

I probably haven't explained my thoughts well, my mind works in mysterious ways. However, your thoughts have helped me a lot, thank you.

Cheers,
Ian

[smithi01]

Again, thanks for the thoughts. I have implemented as follows, which seems to work well. I can't use defer because the defer block doesn't allow async code. I have to make sure I call await signalDictionarySemaphore at every return point.

    private let downloadSemaphore = AsyncSemaphore(value: 8)
    
    let semaphoreDictionaryLock = AsyncSemaphore(value: 1)
    func getDictionarySemaphore(for url: URL) async -> AsyncSemaphore {
        await semaphoreDictionaryLock.wait()
        defer { semaphoreDictionaryLock.signal() }
        if let semaphore = urlSemaphores[url] {
            return semaphore
        } else {
            let semaphore = AsyncSemaphore(value: 1)
            urlSemaphores[url] = semaphore
            return semaphore
        }
    }
    
    func signalDictionarySemaphore(_ semaphore: AsyncSemaphore, for url: URL) async {
        await semaphoreDictionaryLock.wait()
        defer { semaphoreDictionaryLock.signal() }
        let othersWaiting = semaphore.signal()
        if !othersWaiting {
            urlSemaphores[url] = nil
        } else {
        }
    }
    
    private func fetchData(from url: URL, checking cache: DataCache, transform: (DownloadResponse) -> Data?) async -> Data? {
        
        // See if there is a task in progress to download from the URL
        let semaphore = await getDictionarySemaphore(for: url)
        await semaphore.wait()
        if Task.isCancelled {
            await signalDictionarySemaphore(semaphore, for: url)
            return nil
        }
        
        // Check the cache
        if let data = await cache.value(forUrl: url) {
            guard data != DataCache.noData else {
                await signalDictionarySemaphore(semaphore, for: url)
                return nil
            }
            await signalDictionarySemaphore(semaphore, for: url)
            return data
        }
        if Task.isCancelled {
            await signalDictionarySemaphore(semaphore, for: url)
            return nil
        }
        
        // Try to download
        
        // Limit concurrent downloads
        await downloadSemaphore.wait()
        defer {
            downloadSemaphore.signal()
        }
        if Task.isCancelled {
            await signalDictionarySemaphore(semaphore, for: url)
            return nil
        }
        
        do {
            let response = try await downloadData(from: url)

       // Rest of function
'''