Replace detached tasks with Task.dispatch(on:) (#201)

* Use Task.detach(on:) everywhere that was blocking.
* Don't wrap the continuations in a Task 🤦‍♂️
* Add sendable requirement to closure.
This commit is contained in:
Doug
2022-09-21 17:50:15 +01:00
committed by GitHub
parent 60f0f1605b
commit 8462862afc
9 changed files with 125 additions and 130 deletions
+36 -23
View File
@@ -16,33 +16,46 @@
import Foundation
extension Task where Failure == Never {
static func dispatched(on queue: DispatchQueue,
priority: TaskPriority? = nil,
operation: @escaping @Sendable () -> Success) -> Task<Success, Failure> {
Task.detached(priority: priority) {
await withCheckedContinuation { continuation in
queue.async {
continuation.resume(returning: operation())
}
public extension Task where Success == Never, Failure == Never {
/// Dispatches the given closure onto the given queue, wrapped within
/// a continuation to make it non-blocking and awaitable.
///
/// Use this method to `await` blocking calls to the SDK from a `Task`.
///
/// - Parameters:
/// - queue: The queue to run the closure on.
/// - function: A string identifying the declaration that is the notional
/// source for the continuation, used to identify the continuation in
/// runtime diagnostics related to misuse of this continuation.
/// - body: A sendable closure. Use of sendable won't work as it isn't
/// async, but is added to enforce actor semantics.
static func dispatch<T>(on queue: DispatchQueue, function: String = #function, _ body: @escaping @Sendable () -> T) async -> T {
await withCheckedContinuation(function: function) { continuation in
queue.async {
continuation.resume(returning: body())
}
}
}
}
extension Task where Failure == Error {
static func dispatched(on queue: DispatchQueue,
priority: TaskPriority? = nil,
operation: @escaping @Sendable () throws -> Success) -> Task<Success, Failure> {
Task.detached(priority: priority) {
try await withCheckedThrowingContinuation { continuation in
queue.async {
do {
let result = try operation()
continuation.resume(returning: result)
} catch {
continuation.resume(throwing: error)
}
/// Dispatches the given throwing closure onto the given queue, wrapped within
/// a continuation to make it non-blocking and awaitable.
///
/// Use this method to `await` blocking calls to the SDK from a `Task`.
///
/// - Parameters:
/// - queue: The queue to run the closure on.
/// - function: A string identifying the declaration that is the notional
/// source for the continuation, used to identify the continuation in
/// runtime diagnostics related to misuse of this continuation.
/// - body: A sendable closure. Use of sendable won't work as it isn't
/// async, but is added to enforce actor semantics.
static func dispatch<T>(on queue: DispatchQueue, function: String = #function, _ body: @escaping @Sendable () throws -> T) async throws -> T {
try await withCheckedThrowingContinuation(function: function) { continuation in
queue.async {
do {
continuation.resume(returning: try body())
} catch {
continuation.resume(throwing: error)
}
}
}
@@ -23,9 +23,9 @@ struct AttributedStringBuilder: AttributedStringBuilderProtocol {
private let linkColor = UIColor.blue
func fromPlain(_ string: String?) async -> AttributedString? {
await Task.detached {
await Task.dispatch(on: .global()) {
fromPlain(string)
}.value
}
}
func fromPlain(_ string: String?) -> AttributedString? {
@@ -41,9 +41,9 @@ struct AttributedStringBuilder: AttributedStringBuilderProtocol {
}
func fromHTML(_ htmlString: String?) async -> AttributedString? {
await Task.detached {
await Task.dispatch(on: .global()) {
fromHTML(htmlString)
}.value
}
}
// Do not use the default HTML renderer of NSAttributedString because this method
@@ -40,29 +40,26 @@ class AuthenticationServiceProxy: AuthenticationServiceProxyProtocol {
// MARK: - Public
func configure(for homeserverAddress: String) async -> Result<Void, AuthenticationServiceError> {
let task = Task.detached { () -> LoginHomeserver in
do {
var homeserver = LoginHomeserver(address: homeserverAddress, loginMode: .unknown)
try self.authenticationService.configureHomeserver(serverName: homeserverAddress)
guard let details = self.authenticationService.homeserverDetails() else { return homeserver }
if let issuer = details.authenticationIssuer(), let issuerURL = URL(string: issuer) {
homeserver.loginMode = .oidc(issuerURL)
} else if details.supportsPasswordLogin() {
homeserver.loginMode = .password
} else {
homeserver.loginMode = .unsupported
try await Task.dispatch(on: .global()) {
try self.authenticationService.configureHomeserver(serverName: homeserverAddress)
}
if let details = authenticationService.homeserverDetails() {
if let issuer = details.authenticationIssuer(), let issuerURL = URL(string: issuer) {
homeserver.loginMode = .oidc(issuerURL)
} else if details.supportsPasswordLogin() {
homeserver.loginMode = .password
} else {
homeserver.loginMode = .unsupported
}
}
return homeserver
}
switch await task.result {
case .success(let homeserver):
self.homeserver = homeserver
return .success(())
case .failure(let error):
} catch {
MXLog.error("Failed configuring a server: \(error)")
return .failure(.invalidHomeserverAddress)
}
@@ -105,20 +102,19 @@ class AuthenticationServiceProxy: AuthenticationServiceProxyProtocol {
}
func login(username: String, password: String, initialDeviceName: String?, deviceId: String?) async -> Result<UserSessionProtocol, AuthenticationServiceError> {
Benchmark.startTrackingForIdentifier("Login", message: "Started new login")
do {
Benchmark.startTrackingForIdentifier("Login", message: "Started new login")
let loginTask: Task<Client, Error> = Task.detached {
try self.authenticationService.login(username: username,
password: password,
initialDeviceName: initialDeviceName,
deviceId: deviceId)
}
switch await loginTask.result {
case .success(let client):
let client = try await Task.dispatch(on: .global()) {
try self.authenticationService.login(username: username,
password: password,
initialDeviceName: initialDeviceName,
deviceId: deviceId)
}
Benchmark.endTrackingForIdentifier("Login", message: "Finished login")
return await userSession(for: client)
case .failure(let error):
} catch {
Benchmark.endTrackingForIdentifier("Login", message: "Login failed")
MXLog.error("Failed logging in with error: \(error)")
@@ -182,7 +182,7 @@ class ClientProxy: ClientProxyProtocol {
}
func loadUserDisplayName() async -> Result<String, ClientProxyError> {
await Task.detached {
await Task.dispatch(on: .global()) {
do {
let displayName = try self.client.displayName()
return .success(displayName)
@@ -190,11 +190,10 @@ class ClientProxy: ClientProxyProtocol {
return .failure(.failedRetrievingDisplayName)
}
}
.value
}
func loadUserAvatarURLString() async -> Result<String, ClientProxyError> {
await Task.detached {
await Task.dispatch(on: .global()) {
do {
let avatarURL = try self.client.avatarUrl()
return .success(avatarURL)
@@ -202,7 +201,6 @@ class ClientProxy: ClientProxyProtocol {
return .failure(.failedRetrievingAvatarURL)
}
}
.value
}
func accountDataEvent<Content>(type: String) async -> Result<Content?, ClientProxyError> where Content: Decodable {
@@ -218,21 +216,21 @@ class ClientProxy: ClientProxyProtocol {
}
func loadMediaContentForSource(_ source: MatrixRustSDK.MediaSource) async throws -> Data {
try await Task.detached {
try await Task.dispatch(on: .global()) {
let bytes = try self.client.getMediaContent(source: source)
return Data(bytes: bytes, count: bytes.count)
}.value
}
}
func loadMediaThumbnailForSource(_ source: MatrixRustSDK.MediaSource, width: UInt, height: UInt) async throws -> Data {
try await Task.detached {
try await Task.dispatch(on: .global()) {
let bytes = try self.client.getMediaThumbnail(source: source, width: UInt64(width), height: UInt64(height))
return Data(bytes: bytes, count: bytes.count)
}.value
}
}
func sessionVerificationControllerProxy() async -> Result<SessionVerificationControllerProxyProtocol, ClientProxyError> {
await Task.detached {
await Task.dispatch(on: .global()) {
do {
let sessionVerificationController = try self.client.getSessionVerificationController()
return .success(SessionVerificationControllerProxy(sessionVerificationController: sessionVerificationController))
@@ -240,7 +238,6 @@ class ClientProxy: ClientProxyProtocol {
return .failure(.failedRetrievingSessionVerificationController)
}
}
.value
}
func logout() async {
+36 -42
View File
@@ -105,16 +105,15 @@ class RoomProxy: RoomProxyProtocol {
}
func loadAvatarURLForUserId(_ userId: String) async -> Result<String?, RoomProxyError> {
await Task.detached { () -> Result<String?, RoomProxyError> in
do {
let avatarURL = try self.room.memberAvatarUrl(userId: userId)
await self.update(avatarURL: avatarURL, forUserId: userId)
return .success(avatarURL)
} catch {
return .failure(.failedRetrievingMemberAvatarURL)
do {
let avatarURL = try await Task.dispatch(on: .global()) {
try self.room.memberAvatarUrl(userId: userId)
}
update(avatarURL: avatarURL, forUserId: userId)
return .success(avatarURL)
} catch {
return .failure(.failedRetrievingMemberAvatarURL)
}
.value
}
func displayNameForUserId(_ userId: String) -> String? {
@@ -122,33 +121,29 @@ class RoomProxy: RoomProxyProtocol {
}
func loadDisplayNameForUserId(_ userId: String) async -> Result<String?, RoomProxyError> {
await Task.detached { () -> Result<String?, RoomProxyError> in
do {
let displayName = try self.room.memberDisplayName(userId: userId)
await self.update(displayName: displayName, forUserId: userId)
return .success(displayName)
} catch {
return .failure(.failedRetrievingMemberDisplayName)
do {
let displayName = try await Task.dispatch(on: .global()) {
try self.room.memberDisplayName(userId: userId)
}
update(displayName: displayName, forUserId: userId)
return .success(displayName)
} catch {
return .failure(.failedRetrievingMemberDisplayName)
}
.value
}
func loadDisplayName() async -> Result<String, RoomProxyError> {
await Task.detached { () -> Result<String, RoomProxyError> in
if let displayName = await self.displayName {
return .success(displayName)
}
do {
let displayName = try self.room.displayName()
await self.update(displayName: displayName)
return .success(displayName)
} catch {
return .failure(.failedRetrievingDisplayName)
if let displayName = displayName { return .success(displayName) }
do {
let displayName = try await Task.dispatch(on: .global()) {
try self.room.displayName()
}
update(displayName: displayName)
return .success(displayName)
} catch {
return .failure(.failedRetrievingDisplayName)
}
.value
}
private func addTimelineListener(listener: TimelineListener) {
@@ -160,20 +155,20 @@ class RoomProxy: RoomProxyProtocol {
return .failure(.noMoreMessagesToBackPaginate)
}
MXLog.debug("BackPagination")
return await Task.detached {
do {
let id = await self.id
let id = id // Copy the ID due to @Sendable requirement.
do {
let outcome: PaginationOutcome = try await Task.dispatch(on: .global()) {
Benchmark.startTrackingForIdentifier("BackPagination \(id)", message: "Backpaginating \(count) message(s) in room \(id)")
await self.update(backPaginationOutcome: try self.room.paginateBackwards(limit: UInt16(count)))
let outcome = try self.room.paginateBackwards(limit: UInt16(count))
Benchmark.endTrackingForIdentifier("BackPagination \(id)", message: "Finished backpaginating \(count) message(s) in room \(id)")
return .success(())
} catch {
return .failure(.failedPaginatingBackwards)
return outcome
}
update(backPaginationOutcome: outcome)
return .success(())
} catch {
return .failure(.failedPaginatingBackwards)
}
.value
}
func sendMessage(_ message: String, inReplyToEventId: String? = nil) async -> Result<Void, RoomProxyError> {
@@ -184,7 +179,7 @@ class RoomProxy: RoomProxyProtocol {
let transactionId = genTransactionId()
return await Task.dispatched(on: concurrentDispatchQueue, operation: {
return await Task.dispatch(on: .global()) {
do {
if let inReplyToEventId = inReplyToEventId {
try self.room.sendReply(msg: message, inReplyToEventId: inReplyToEventId, txnId: transactionId)
@@ -196,8 +191,7 @@ class RoomProxy: RoomProxyProtocol {
} catch {
return .failure(.failedSendingMessage)
}
})
.value
}
}
func redact(_ eventID: String) async -> Result<Void, RoomProxyError> {
@@ -63,7 +63,7 @@ class SessionVerificationControllerProxy: SessionVerificationControllerProxyProt
}
func requestVerification() async -> Result<Void, SessionVerificationControllerProxyError> {
await Task.detached {
await Task.dispatch(on: .global()) {
do {
try self.sessionVerificationController.requestVerification()
return .success(())
@@ -71,11 +71,10 @@ class SessionVerificationControllerProxy: SessionVerificationControllerProxyProt
return .failure(.failedRequestingVerification)
}
}
.value
}
func approveVerification() async -> Result<Void, SessionVerificationControllerProxyError> {
await Task.detached {
await Task.dispatch(on: .global()) {
do {
try self.sessionVerificationController.approveVerification()
return .success(())
@@ -83,11 +82,10 @@ class SessionVerificationControllerProxy: SessionVerificationControllerProxyProt
return .failure(.failedApprovingVerification)
}
}
.value
}
func declineVerification() async -> Result<Void, SessionVerificationControllerProxyError> {
await Task.detached {
await Task.dispatch(on: .global()) {
do {
try self.sessionVerificationController.declineVerification()
return .success(())
@@ -95,11 +93,10 @@ class SessionVerificationControllerProxy: SessionVerificationControllerProxyProt
return .failure(.failedDecliningVerification)
}
}
.value
}
func cancelVerification() async -> Result<Void, SessionVerificationControllerProxyError> {
await Task.detached {
await Task.dispatch(on: .global()) {
do {
try self.sessionVerificationController.cancelVerification()
return .success(())
@@ -107,7 +104,6 @@ class SessionVerificationControllerProxy: SessionVerificationControllerProxyProt
return .failure(.failedCancellingVerification)
}
}
.value
}
// MARK: - Private
@@ -93,16 +93,14 @@ class UserSessionStore: UserSessionStoreProtocol {
.basePath(path: baseDirectoryPath)
.username(username: credentials.userID)
let loginTask: Task<Client, Error> = Task.detached {
let client = try builder.build()
try client.restoreLogin(restoreToken: credentials.restoreToken)
return client
}
switch await loginTask.result {
case .success(let client):
do {
let client: Client = try await Task.dispatch(on: .global()) {
let client = try builder.build()
try client.restoreLogin(restoreToken: credentials.restoreToken)
return client
}
return await setupProxyForClient(client)
case .failure(let error):
} catch {
MXLog.error("Failed restoring login with error: \(error)")
return .failure(.failedRestoringLogin)
}