CODEWITHSUNDEEP
swiftswiftuiasync-awaitconcurrency
Michael
Michael

Reputation: 327

Swiftui Tasks are not running in Parallel

I have a problem running multiply tasks in parallel in a SwiftUI view.

struct ModelsView: View {
    @StateObject var tasks = TasksViewModel()

    var body: some View {
        NavigationView{
            ScrollView {
                ForEach(Array(zip(tasks.tasks.indices, tasks.tasks)), id: \.0) { task in
                    NavigationLink(destination: ModelView()) {
                        ModelPreviewView(model_name: "3dobject.usdz")
                            .onAppear {
                                if task.0 == tasks.tasks.count - 2 {
                                    Task {
                                        print(tasks.tasks.count)
                                        await tasks.fetch_tasks(count: 4)
                                    }
                                }
                            }
                    }
                }
            }.navigationTitle("3D modelle")
        }.onAppear{
            Task {
                await tasks.fetch_tasks(count: 5)
                await tasks.watch_for_new_tasks()
            }
        }
    }
}

In my view, I spawn a task as soon as the View Appears which, first, fetches 5 tasks from the database (this works fine), and then it starts watching for new tasks.

In the Scroll View, right before the bottom is reached, I start loading new tasks. The problem is, the asynchronous function fetch_tasks(count: 4) only gets continued if the asynchronous function watch_for_new_tasks() stops blocking.

actor TasksViewModel: ObservableObject {
    @MainActor @Published private(set) var tasks : [Tasks.Task] = []

    private var last_fetched_id : String? = nil

    func fetch_tasks(count: UInt32) async {
        do {
            let tasks_data = try await RedisClient.shared.xrevrange(streamName: "tasks", end: last_fetched_id ?? "+" , start: "-", count: count)
            last_fetched_id = tasks_data.last?.id
            let fetched_tasks = tasks_data.compactMap { Tasks.Task(from: $0.data) }

            await MainActor.run {
                withAnimation(.easeInOut) {
                    self.tasks.append(contentsOf: fetched_tasks)
                }
            }
        } catch {
            print("Error fetching taskss \(error)")
        }
    }

    func watch_for_new_tasks() async {
        while !Task.isCancelled {
            do {

                let tasks_data = try await RedisClient.shared.xread(streams: "tasks", ids: "$")
                let new_tasks = tasks_data.compactMap { Tasks.Task(from: $0.data) }

                await MainActor.run {
                    for new_task in new_tasks.reversed() {
                        withAnimation {
                            self.tasks.insert(new_task, at: 0)
                        }
                    }
                }
            } catch {
                print(error)
            }
        }
    }
 ...
}

The asynchronous function watch_for_new_tasks() uses RedisClient.shared.xread(streams: "tasks", ids: "$") which blocks until at least one tasks is added to the Redis Stream.

This is my redis client:

class RedisClient {
    typealias Stream = Array<StreamElement>
    
    static let shared = RedisClient(host: "127.0.0.1", port: 6379)
    
    let connection: Redis
    
    let host: String
    let port: Int32
    
    init(host: String, port: Int32) {
        connection = Redis()
        self.host = host
        self.port = port
        
        connection.connect(host: host, port: port) {error in
            if let err = error {
                print(err)
            }
        }
    }
    
    func connect() {
        connection.connect(host: self.host, port: self.port) {error in
            if let err = error {
                print(err)
            }
        }
    }
    
    func xrevrange(streamName: String, end: String, start: String, count: UInt32 = 0) async throws -> Stream {
        try await withCheckedThrowingContinuation { continuation in
            connection.issueCommand("xrevrange", streamName, end, start, "COUNT", String(count)) { res in
                switch res {
                case .Array(let data):
                    continuation.resume(returning: data.compactMap { StreamElement(from: $0) } )
                case .Error(let error):
                    continuation.resume(throwing: ResponseError.RedisError(error))
                case _:
                    continuation.resume(throwing: ResponseError.WrongData("Expected Array"))
                }
            }
        }
    }
    
    func xread(streams: String..., ids: String..., block: UInt32 = 0, count: UInt32 = 0) async throws -> Stream {
        return try await withCheckedThrowingContinuation({ continuation in
            var args = ["xread", "BLOCK", String(block),"COUNT", String(count),"STREAMS"]
            args.append(contentsOf: streams)
            args.append(contentsOf: ids)
            connection.issueCommandInArray(args){ res in
                print(res)
                switch res.asArray?[safe: 0]?.asArray?[safe: 1] ?? .Error("Expected response to be an array") {
                case .Array(let data):
                    continuation.resume(returning: data.compactMap { StreamElement(from: $0) } )
                case .Error(let error):
                    continuation.resume(throwing: ResponseError.RedisError(error))
                case _:
                    continuation.resume(throwing: ResponseError.WrongData("Expected Array"))
                }
            }
        })
    }
    
    func xreadgroup(group: String, consumer: String, count: UInt32 = 0, block: UInt32 = 0, streams: String..., ids: String..., noAck: Bool = true) async throws -> Stream {
        try await withCheckedThrowingContinuation({ continuation in
            var args = ["xreadgroup", "GROUP", group, consumer, "COUNT", String(count), "BLOCK", noAck ? nil : "NOACK", String(block), "STREAMS"].compactMap{ $0 }
            args.append(contentsOf: streams)
            args.append(contentsOf: ids)
            connection.issueCommandInArray(args){ res in
                print(res)
                switch res.asArray?[safe: 0]?.asArray?[safe: 1] ?? .Error("Expected response to be an array") {
                case .Array(let data):
                    continuation.resume(returning: data.compactMap { StreamElement(from: $0) } )
                case .Error(let error):
                    continuation.resume(throwing: ResponseError.RedisError(error))
                case _:
                    continuation.resume(throwing: ResponseError.WrongData("Expected Array"))
                }
            }
        })
    }
    
    enum ResponseError: Error {
        case RedisError(String)
        case WrongData(String)
    }
    
    struct StreamElement {
        let id: String
        let data: [RedisResponse]
        
        init?(from value: RedisResponse) {
            guard
                case .Array(let values) = value,
                let id = values[0].asString,
                let data = values[1].asArray
            else { return nil }
            
            self.id = id.asString
            self.data = data
        }
    }
}

I tried running the watch_for_new_tasks() on a Task.detached tasks, but that also blocks.

To be honest, I have no idea why this blocks, and I could use your guy's help if you could.

Thank you in Advance,

Michael

Upvotes: 4

Views: 485

Answers (1)

cora
cora

Reputation: 2102

.onAppear {
            Task {
                await tasks.fetch_tasks(count: 5)
                await tasks.watch_for_new_tasks()
            }
        }
  1. This does not run tasks in parallel. For the 2nd await to execute, the 1st one has to finish.
  2. You can use .task modifier

You code can be refactored into this to run 2 async functions in parallel:

.task {
    async let fetchTask = tasks.fetch_tasks(count: 5)
    async let watchTask = tasks.watch_for_new_tasks()
}

You can do:

await [fetchTask, watchTask]

if you need to do something after both of them complete

Upvotes: 3

Related Questions