Concurrency Fixes (#283)

* Review places where MainActor used * Adapt to MainActor changes, avoid redundant task detachments * Move Rust client operations into a dedicated concurrent queue, make init async * Add changelog * Fixed small working for awaiting non-async method * Remove some redundant tasks * Put back some main actors * Fix tests * Fixed timeline updates when the number of items doesn't change. Keeps more previous home screen room data between reloads Co-authored-by: Stefan Ceriu <stefanc@matrix.org>
2026-11-04 01:20:12 +00:00 · 2022-11-02 18:25:46 +03:00
parent 965569aae2
commit f6a599a51b
19 changed files with 243 additions and 221 deletions
@@ -27,7 +27,6 @@ enum AuthenticationServiceError: Error {
    case failedLoggingIn
 }

-@MainActor
 protocol AuthenticationServiceProxyProtocol {
    var homeserver: LoginHomeserver { get }
    
@@ -16,6 +16,7 @@

 import Foundation

+@MainActor
 protocol BackgroundTaskServiceProtocol {
    func startBackgroundTask(withName name: String,
                             isReusable: Bool,
@@ -31,23 +31,17 @@ private class WeakClientProxyWrapper: ClientDelegate, SlidingSyncObserver {
    func didReceiveSyncUpdate() { }

    func didReceiveAuthError(isSoftLogout: Bool) {
-        Task {
-            await clientProxy?.didReceiveAuthError(isSoftLogout: isSoftLogout)
-        }
+        clientProxy?.didReceiveAuthError(isSoftLogout: isSoftLogout)
    }

    func didUpdateRestoreToken() {
-        Task {
-            await clientProxy?.didUpdateRestoreToken()
-        }
+        clientProxy?.didUpdateRestoreToken()
    }
    
    // MARK: - SlidingSyncDelegate
    
    func didReceiveSyncUpdate(summary: UpdateSummary) {
-        Task {
-            await self.clientProxy?.didReceiveSlidingSyncUpdate(summary: summary)
-        }
+        clientProxy?.didReceiveSlidingSyncUpdate(summary: summary)
    }
 }

@@ -58,11 +52,18 @@ class ClientProxy: ClientProxyProtocol {
    private let client: ClientProtocol
    private let backgroundTaskService: BackgroundTaskServiceProtocol
    private var sessionVerificationControllerProxy: SessionVerificationControllerProxy?
+    private let clientQueue: DispatchQueue
    
    private var slidingSyncObserverToken: StoppableSpawn?
-    private let slidingSync: SlidingSync
+    private var slidingSync: SlidingSync!
    
-    let roomSummaryProvider: RoomSummaryProviderProtocol
+    var roomSummaryProviderInternal: RoomSummaryProviderProtocol!
+    var roomSummaryProvider: RoomSummaryProviderProtocol {
+        guard let roomSummaryProviderInternal else {
+            fatalError("There is an issue with ClientProxy object initialization")
+        }
+        return roomSummaryProviderInternal
+    }
    
    deinit {
        // These need to be inlined instead of using stopSync()
@@ -75,33 +76,37 @@ class ClientProxy: ClientProxyProtocol {
    let callbacks = PassthroughSubject<ClientProxyCallback, Never>()
    
    init(client: ClientProtocol,
-         backgroundTaskService: BackgroundTaskServiceProtocol) {
+         backgroundTaskService: BackgroundTaskServiceProtocol) async {
        self.client = client
        self.backgroundTaskService = backgroundTaskService
-        
-        do {
-            let slidingSyncBuilder = try client.slidingSync().homeserver(url: BuildSettings.slidingSyncProxyBaseURL.absoluteString)
-            
-            let slidingSyncView = try SlidingSyncViewBuilder()
-                .timelineLimit(limit: 10)
-                .requiredState(requiredState: [RequiredState(key: "m.room.avatar", value: ""),
-                                               RequiredState(key: "m.room.encryption", value: "")])
-                .name(name: "HomeScreenView")
-                .syncMode(mode: .fullSync)
-                .build()
-            
-            slidingSync = try slidingSyncBuilder
-                .addView(view: slidingSyncView)
-                .withCommonExtensions()
-                .build()
-            
-            roomSummaryProvider = RoomSummaryProvider(slidingSyncController: slidingSync,
-                                                      slidingSyncView: slidingSyncView,
-                                                      roomMessageFactory: RoomMessageFactory())
-        } catch {
-            fatalError("Failed configuring sliding sync")
+        clientQueue = .init(label: "ClientProxyQueue",
+                            attributes: .concurrent)
+
+        await Task.dispatch(on: clientQueue) {
+            do {
+                let slidingSyncBuilder = try client.slidingSync().homeserver(url: BuildSettings.slidingSyncProxyBaseURL.absoluteString)
+
+                let slidingSyncView = try SlidingSyncViewBuilder()
+                    .timelineLimit(limit: 10)
+                    .requiredState(requiredState: [RequiredState(key: "m.room.avatar", value: ""),
+                                                   RequiredState(key: "m.room.encryption", value: "")])
+                    .name(name: "HomeScreenView")
+                    .syncMode(mode: .fullSync)
+                    .build()
+
+                self.slidingSync = try slidingSyncBuilder
+                    .addView(view: slidingSyncView)
+                    .withCommonExtensions()
+                    .build()
+
+                self.roomSummaryProviderInternal = RoomSummaryProvider(slidingSyncController: self.slidingSync,
+                                                                       slidingSyncView: slidingSyncView,
+                                                                       roomMessageFactory: RoomMessageFactory())
+            } catch {
+                fatalError("Failed configuring sliding sync")
+            }
        }
-        
+
        client.setDelegate(delegate: WeakClientProxyWrapper(clientProxy: self))
    }
    
@@ -156,27 +161,22 @@ class ClientProxy: ClientProxyProtocol {
        slidingSync.setObserver(observer: nil)
    }
    
-    func roomForIdentifier(_ identifier: String) -> RoomProxyProtocol? {
-        do {
-            guard let slidingSyncRoom = try slidingSync.getRoom(roomId: identifier),
-                  let room = slidingSyncRoom.fullRoom() else {
-                MXLog.error("Failed retrieving room with identifier: \(identifier)")
-                return nil
-            }
-            
-            let roomProxy = RoomProxy(slidingSyncRoom: slidingSyncRoom,
-                                      room: room,
-                                      backgroundTaskService: backgroundTaskService)
-            
-            return roomProxy
-        } catch {
-            MXLog.error("Failed retrieving room with identifier: \(identifier)")
+    func roomForIdentifier(_ identifier: String) async -> RoomProxyProtocol? {
+        let (slidingSyncRoom, room) = await Task.dispatch(on: clientQueue) {
+            self.roomTupleForIdentifier(identifier)
+        }
+
+        guard let slidingSyncRoom, let room else {
            return nil
        }
+
+        return await RoomProxy(slidingSyncRoom: slidingSyncRoom,
+                               room: room,
+                               backgroundTaskService: backgroundTaskService)
    }

    func loadUserDisplayName() async -> Result<String, ClientProxyError> {
-        await Task.dispatch(on: .global()) {
+        await Task.dispatch(on: clientQueue) {
            do {
                let displayName = try self.client.displayName()
                return .success(displayName)
@@ -187,7 +187,7 @@ class ClientProxy: ClientProxyProtocol {
    }
        
    func loadUserAvatarURLString() async -> Result<String, ClientProxyError> {
-        await Task.dispatch(on: .global()) {
+        await Task.dispatch(on: clientQueue) {
            do {
                let avatarURL = try self.client.avatarUrl()
                return .success(avatarURL)
@@ -198,11 +198,15 @@ class ClientProxy: ClientProxyProtocol {
    }
    
    func accountDataEvent<Content>(type: String) async -> Result<Content?, ClientProxyError> where Content: Decodable {
-        .failure(.failedRetrievingAccountData)
+        await Task.dispatch(on: clientQueue) {
+            .failure(.failedRetrievingAccountData)
+        }
    }
    
    func setAccountData<Content: Encodable>(content: Content, type: String) async -> Result<Void, ClientProxyError> {
-        .failure(.failedSettingAccountData)
+        await Task.dispatch(on: clientQueue) {
+            .failure(.failedSettingAccountData)
+        }
    }
    
    func mediaSourceForURLString(_ urlString: String) -> MatrixRustSDK.MediaSource {
@@ -210,21 +214,21 @@ class ClientProxy: ClientProxyProtocol {
    }
    
    func loadMediaContentForSource(_ source: MatrixRustSDK.MediaSource) async throws -> Data {
-        try await Task.dispatch(on: .global()) {
+        try await Task.dispatch(on: clientQueue) {
            let bytes = try self.client.getMediaContent(source: source)
            return Data(bytes: bytes, count: bytes.count)
        }
    }
    
    func loadMediaThumbnailForSource(_ source: MatrixRustSDK.MediaSource, width: UInt, height: UInt) async throws -> Data {
-        try await Task.dispatch(on: .global()) {
+        try await Task.dispatch(on: clientQueue) {
            let bytes = try self.client.getMediaThumbnail(source: source, width: UInt64(width), height: UInt64(height))
            return Data(bytes: bytes, count: bytes.count)
        }
    }
    
    func sessionVerificationControllerProxy() async -> Result<SessionVerificationControllerProxyProtocol, ClientProxyError> {
-        await Task.dispatch(on: .global()) {
+        await Task.dispatch(on: clientQueue) {
            do {
                let sessionVerificationController = try self.client.getSessionVerificationController()
                return .success(SessionVerificationControllerProxy(sessionVerificationController: sessionVerificationController))
@@ -235,14 +239,28 @@ class ClientProxy: ClientProxyProtocol {
    }

    func logout() async {
-        do {
-            try client.logout()
-        } catch {
-            MXLog.error("Failed logging out with error: \(error)")
+        await Task.dispatch(on: clientQueue) {
+            do {
+                try self.client.logout()
+            } catch {
+                MXLog.error("Failed logging out with error: \(error)")
+            }
        }
    }
    
    // MARK: Private
+
+    private func roomTupleForIdentifier(_ identifier: String) -> (SlidingSyncRoom?, Room?) {
+        do {
+            let slidingSyncRoom = try slidingSync.getRoom(roomId: identifier)
+            let fullRoom = slidingSyncRoom?.fullRoom()
+
+            return (slidingSyncRoom, fullRoom)
+        } catch {
+            MXLog.error("Failed retrieving room with identifier: \(identifier)")
+            return (nil, nil)
+        }
+    }
    
    fileprivate func didReceiveAuthError(isSoftLogout: Bool) {
        callbacks.send(.receivedAuthError(isSoftLogout: isSoftLogout))
@@ -33,7 +33,6 @@ enum ClientProxyError: Error {
    case failedLoadingMedia
 }

-@MainActor
 protocol ClientProxyProtocol {
    var callbacks: PassthroughSubject<ClientProxyCallback, Never> { get }
    
@@ -53,7 +52,7 @@ protocol ClientProxyProtocol {
    
    func stopSync()
    
-    func roomForIdentifier(_ identifier: String) -> RoomProxyProtocol?
+    func roomForIdentifier(_ identifier: String) async -> RoomProxyProtocol?
    
    func loadUserDisplayName() async -> Result<String, ClientProxyError>
        
@@ -32,7 +32,7 @@ struct MockClientProxy: ClientProxyProtocol {
    
    func stopSync() { }
    
-    func roomForIdentifier(_ identifier: String) -> RoomProxyProtocol? {
+    func roomForIdentifier(_ identifier: String) async -> RoomProxyProtocol? {
        nil
    }
    
@@ -55,43 +55,38 @@ struct MediaProvider: MediaProviderProtocol {
            return .success(image)
        }

-        let loadImageBgTask = backgroundTaskService.startBackgroundTask(withName: "LoadImage: \(source.underlyingSource.url().hashValue)")
+        let loadImageBgTask = await backgroundTaskService.startBackgroundTask(withName: "LoadImage: \(source.url.hashValue)")
        defer {
            loadImageBgTask?.stop()
        }
        
-        let cacheKey = cacheKeyForURLString(source.underlyingSource.url(), avatarSize: avatarSize)
-        
-        return await Task.detached { () -> Result<UIImage, MediaProviderError> in
-            if case let .success(cacheResult) = await imageCache.retrieveImage(forKey: cacheKey),
-               let image = cacheResult.image {
-                return .success(image)
-            }
-            
-            do {
-                let imageData = try await Task.detached { () -> Data in
-                    if let avatarSize {
-                        return try await clientProxy.loadMediaThumbnailForSource(source.underlyingSource, width: UInt(avatarSize.scaledValue), height: UInt(avatarSize.scaledValue))
-                    } else {
-                        return try await clientProxy.loadMediaContentForSource(source.underlyingSource)
-                    }
-                    
-                }.value
-                
-                guard let image = UIImage(data: imageData) else {
-                    MXLog.error("Invalid image data")
-                    return .failure(.invalidImageData)
-                }
-                
-                imageCache.store(image, forKey: cacheKey)
-                
-                return .success(image)
-            } catch {
-                MXLog.error("Failed retrieving image with error: \(error)")
-                return .failure(.failedRetrievingImage)
-            }
+        let cacheKey = cacheKeyForURLString(source.url, avatarSize: avatarSize)
+
+        if case let .success(cacheResult) = await imageCache.retrieveImage(forKey: cacheKey),
+           let image = cacheResult.image {
+            return .success(image)
+        }
+
+        do {
+            let imageData: Data
+            if let avatarSize {
+                imageData = try await clientProxy.loadMediaThumbnailForSource(source.underlyingSource, width: UInt(avatarSize.scaledValue), height: UInt(avatarSize.scaledValue))
+            } else {
+                imageData = try await clientProxy.loadMediaContentForSource(source.underlyingSource)
+            }
+
+            guard let image = UIImage(data: imageData) else {
+                MXLog.error("Invalid image data")
+                return .failure(.invalidImageData)
+            }
+
+            imageCache.store(image, forKey: cacheKey)
+
+            return .success(image)
+        } catch {
+            MXLog.error("Failed retrieving image with error: \(error)")
+            return .failure(.failedRetrievingImage)
        }
-        .value
    }
    
    // MARK: - Private
@@ -22,7 +22,6 @@ enum MediaProviderError: Error {
    case invalidImageData
 }

-@MainActor
 protocol MediaProviderProtocol {
    func imageFromSource(_ source: MediaSource?, avatarSize: AvatarSize?) -> UIImage?
    
@@ -19,6 +19,10 @@ import MatrixRustSDK

 struct MediaSource: Equatable {
    let underlyingSource: MatrixRustSDK.MediaSource
+
+    var url: String {
+        underlyingSource.url()
+    }
    
    init(source: MatrixRustSDK.MediaSource) {
        underlyingSource = source
@@ -101,36 +101,34 @@ class RoomSummaryProvider: RoomSummaryProviderProtocol {
    }
    
    func updateRoomsWithIdentifiers(_ identifiers: [String]) {
-        Task.detached {
-            guard self.statePublisher.value == .live else {
-                return
-            }
-            
-            var changes = [CollectionDifference<RoomSummary>.Change]()
-            for identifier in identifiers {
-                guard let index = self.rooms.firstIndex(where: { $0.id == identifier }) else {
-                    continue
-                }
-                
-                let oldRoom = self.rooms[index]
-                let newRoom = self.buildRoomSummaryForIdentifier(identifier)
-                
-                changes.append(.remove(offset: index, element: oldRoom, associatedWith: nil))
-                changes.append(.insert(offset: index, element: newRoom, associatedWith: nil))
-            }
-            
-            guard let diff = CollectionDifference(changes) else {
-                MXLog.error("Failed creating diff from changes: \(changes)")
-                return
-            }
-            
-            guard let newSummaries = self.rooms.applying(diff) else {
-                MXLog.error("Failed applying diff: \(diff)")
-                return
-            }
-            
-            self.rooms = newSummaries
+        guard statePublisher.value == .live else {
+            return
        }
+
+        var changes = [CollectionDifference<RoomSummary>.Change]()
+        for identifier in identifiers {
+            guard let index = rooms.firstIndex(where: { $0.id == identifier }) else {
+                continue
+            }
+
+            let oldRoom = rooms[index]
+            let newRoom = buildRoomSummaryForIdentifier(identifier)
+
+            changes.append(.remove(offset: index, element: oldRoom, associatedWith: nil))
+            changes.append(.insert(offset: index, element: newRoom, associatedWith: nil))
+        }
+
+        guard let diff = CollectionDifference(changes) else {
+            MXLog.error("Failed creating diff from changes: \(changes)")
+            return
+        }
+
+        guard let newSummaries = rooms.applying(diff) else {
+            MXLog.error("Failed applying diff: \(diff)")
+            return
+        }
+
+        rooms = newSummaries
    }
    
    // MARK: - Private
@@ -24,7 +24,6 @@ enum UserSessionCallback {
    case updateRestoreTokenNeeded
 }

-@MainActor
 protocol UserSessionProtocol {
    var userID: String { get }
    var isSoftLogout: Bool { get }
@@ -45,11 +45,11 @@ class RoomTimelineProvider: RoomTimelineProviderProtocol {
    init(roomProxy: RoomProxyProtocol) {
        self.roomProxy = roomProxy
        itemProxies = []
-        
+
        Task {
            let roomTimelineListener = RoomTimelineListener()
            await roomProxy.addTimelineListener(listener: roomTimelineListener)
-            
+
            roomTimelineListener
                .itemsUpdatePublisher
                .collect(.byTime(DispatchQueue.global(qos: .background), 0.5))
@@ -135,7 +135,7 @@ class UserSessionStore: UserSessionStoreProtocol {
            return .failure(.failedSettingUpSession)
        }
        
-        let clientProxy = ClientProxy(client: client, backgroundTaskService: backgroundTaskService)
+        let clientProxy = await ClientProxy(client: client, backgroundTaskService: backgroundTaskService)
        
        return .success(clientProxy)
    }
@@ -24,7 +24,6 @@ enum UserSessionStoreError: Error {
    case failedRefreshingRestoreToken
 }

-@MainActor
 protocol UserSessionStoreProtocol {
    /// Whether or not there are sessions in the store.
    var hasSessions: Bool { get }