I am trying to create a class that executes data loading once and returns the data to all callers of the method while the data was loading to not perform the data loading for the same item (identifier) more than once. The issue I am having is that it seems to crash on the first initialization of CurrentValueSubject for an identifier. This only happens if the downloadStuff returns an Error I have no idea what's wrong. Here is a reproduction of the issue.

Class that does the synchronization:

class FetchSynchronizer<T, ItemIdentifier: Hashable> {

typealias CustomParams = (isFirstLoad: Bool, result: Result<T, Error>)

    enum FetchCondition {
        // executes data fetching only once
        case executeFetchOnlyOnce
        // re-executes fetch if request failed
        case retryOnlyIfFailure
        // always executes fetch even if response is cached
        case noDataCache
        // custom condition
        case custom((CustomParams) -> Bool)
    }
    
    struct LoadingState<T> {
        let result: Result<T, Error>
        let isLoading: Bool
        
        init(result: Result<T, Error>? = nil, isLoading: Bool = false) {
            self.result = result ?? .failure(NoResultsError())
            self.isLoading = isLoading
        }
    }
    
    private var cancellables = Set<AnyCancellable>()
    private var isLoading: [ItemIdentifier: CurrentValueSubject<LoadingState<T>, Never>] = [:]
    
    func startLoading(identifier: ItemIdentifier,
                      fetchCondition: FetchCondition = .executeFetchOnlyOnce,
                      loaderMethod: @escaping () async -> Result<T, Error>) async -> Result<T, Error> {
        
        // initialize loading tracker for identifier on first execution
        var isFirstExecution = false
        if isLoading[identifier] == nil {
            print("----0")
            isLoading[identifier] = CurrentValueSubject<LoadingState<T>, Never>(LoadingState<T>())
            isFirstExecution = true
        }
        
        guard let currentIsLoading = isLoading[identifier] else {
            assertionFailure("Should never be nil because it's set above")
            return .failure(NoResultsError())
        }
        
        if currentIsLoading.value.isLoading {
            // loading in progress, wait for finish and call pending callbacks
            return await withCheckedContinuation { continuation in
                currentIsLoading.filter { !$0.isLoading }.sink { currentIsLoading in
                    continuation.resume(returning: currentIsLoading.result)
                }.store(in: &cancellables)
            }
        } else {
            // no fetching in progress, check if it can be executed
            let shouldFetchData: Bool
            switch fetchCondition {
            case .executeFetchOnlyOnce:
                // first execution -> fetch data
                shouldFetchData = isFirstExecution
            case .retryOnlyIfFailure:
                // no cached data -> fetch data
                switch currentIsLoading.value.result {
                case .success:
                    shouldFetchData = false
                case .failure:
                    shouldFetchData = true
                }
            case .noDataCache:
                // always fetch
                shouldFetchData = true
            case .custom(let completion):
                shouldFetchData = completion((isFirstLoad: isFirstExecution,
                                              result: currentIsLoading.value.result))
            }
            
            if shouldFetchData {
                
                currentIsLoading.send(LoadingState(isLoading: true))
                // fetch data
                return await withCheckedContinuation { continuation in
                    Task {
                        // execute loader method
                        let result = await loaderMethod()
                        let state = LoadingState(result: result,
                                                 isLoading: false)
                        currentIsLoading.send(state)
                        continuation.resume(returning: result)
                    }
                }
            } else {
                // use existing data
                return currentIsLoading.value.result
            }
        }
    }
}

Example usage:

class Executer {
    
    let fetchSynchronizer = FetchSynchronizer<Data?, String>()
    
    func downloadStuff() async -> Result<Data?, Error> {
        await fetchSynchronizer.startLoading(identifier: "1") {
            return await withCheckedContinuation { continuation in
                sleep(UInt32.random(in: 1...3))
                print("-------request")
                continuation.resume(returning: .failure(NSError() as Error))
            }
        }
    }
    
    init() {
        start()
    }
    
    func start() {
        Task {
            await downloadStuff()
            print("-----3")
        }
        DispatchQueue.global(qos: .utility).async {
            Task {
                await self.downloadStuff()
                print("-----2")
            }
        }
        
        DispatchQueue.global(qos: .background).async {
            Task {
                await self.downloadStuff()
                print("-----1")
            }
        }
    }
}

Start the execution:

Executer()

Crashes at

isLoading[identifier] = CurrentValueSubject<LoadingState<T>, Never>(LoadingState<T>())

Any guidance would be appreciated.

CodePudding user response：

Swift Dictionary is not thread-safe. You need to make sure it is being accessed from only one thread (i.e queue) or using locks.

EDIT - another solution suggested by @Bogdan the question writer is to make the class an actor class which the concurrency safety is taken care of by the compiler!

By dispatching to a global queue, you increase the chance that two threads will try and write into the dictionary “at the same time” which probably causes the crash

Take a look at these examples. How to implement a Thread Safe HashTable (PhoneBook) Data Structure in Swift?

https://github.com/iThink32/Thread-Safe-Dictionary/blob/main/ThreadSafeDictionary.swift