Async await implementation

[patskovn]

New swift concurrency model consists from several parts. Two most valuable of them is synchronous-like execution flow (async await itself) and performance and safety improvements (prevention of priority inversion, much less context switches between threads via cooperative thread pool). So for proper adoption of whole swift concurrency idea we should consider several phases. Let's just describe them:

• Phase 1. Providing quick async await wrapper via continuations.
• Phase 2. Increasing performance, illuminating extra context switches and usage of cooperative thread pools through actor synchronisation mechanism under the hood for AsyncDatabaseQueue, separating async version of protocols from sync version.
• Phase 3. Usage of custom executors for AsyncDatabasePool. Phase 3 is currently not implementable because of lack of custom executors support.

This discussion will be just about Phase 1. I will provide drafts, difficulties and possibilities for adoption. A little bit later I will create discussion for Phase 2 with more options and reasoning to show idea and take more feedback and thoughts from @groue and other folks.
Let's code!

Implementation of async await via continuation if pretty straightforward for any method. Just:

func asyncAwaitVersionOfMethod<T>(_ value: @escaping (Database) throws -> T) async throws -> T {
    try await withUnsafeThrowingContinuation { continuation in // Wrap with continuation
        asyncVersionOfMethodWithClosure { result in // Call method with closure completion
            do {
	        try continuation.resume(returning: value(result.get()))
            } catch {
	        continuation.resume(throwing: error)
            }
        }
    }
}

We have a problems with incorrect implementation of method overloading diffing just with async keyword. All problems described below happened only with protocols. When we use just concrete classes, everything works perfectly. I tried several approaches to workaround the issue and here is my notes:

1. The case from swift bug that is not working

We can't create tho methods in protocols diffing with just async keyword. Compilation fails.

protocol TestProtocol {
    func test()
    func test() async
}

class TestClass: TestProtocol {
    func test() {}
    func test() async {}
}

2. Workaround that works for methods with any parameters

But! We can just create protocol extension that wraps one of another protocol methods. This compiles successfully.

protocol TestProtocol {
    func test(closure: () -> Int) -> Int
}
extension TestProtocol {
    func test(closure: () -> Int) async -> Int {
        return closure()
    }
}

class TestClass: TestProtocol {
    func test(closure: () -> Int) -> Int { return closure() } 
}

class Consumer {
    let t = TestClass()
    func caller() async -> Int {
          return await t.test(closure: { return 0 })
    }
}

3. But this not really work when we have generics

This still compiles without errors.

But! We are facing problems from the call side when we start using generics. Compiler shows us warningNo 'async' operations occur within 'await' expression when we call await t.test(closure: { return 0 })

protocol TestProtocol {
    func test<T>(closure: () -> T) -> T
}
extension TestProtocol {
    func test<T>(closure: () -> T) async -> T {
        return closure()
    }
}

class TestClass: TestProtocol {
    func test<T>(closure: () -> T) -> T { return closure() }
}

class Consumer {
    let t = TestClass()
    func caller() async -> Int {
          return await t.test(closure: { return 0 })
    }
}

4. Lets apply next filthy workaround which looks like solves our problems again

Bring good old @_disfavoredOverload. This compiles without any warnings and errors.

protocol TestProtocol {
   // NOTE: We are using @_disfavoredOverload on the sync method in protocol to handle method resolving using protocol
    @_disfavoredOverload
    func test<T>(closure: () -> T) -> T
}
extension TestProtocol {
    func test<T>(closure: () -> T) async -> T {
        return closure()
    }
}

class TestClass: TestProtocol {
   // NOTE: We are using @_disfavoredOverload on the sync method in protocol to handle method resolving using class
    @_disfavoredOverload
    func test<T>(closure: () -> T) -> T { return closure() }
}

class Consumer {
    let t = TestClass()
    func caller() async -> Int {
          return await t.test(closure: { return 0 })
    }

    func protocolCaller() async -> Int {
        let ptorocolVersion: TestProtocol = t
        return await t.test(closure: { return 0 })
    }

    func syncVersionCaller() -> Int {
          return t.test(closure: { return 0 })
    }
}

P.S. Everything tested with
Xcode Version 13.0 (13A233)
swift-driver version: 1.26.9 Apple Swift version 5.5 (swiftlang-1300.0.29.102 clang-1300.0.28.1)
Target: x86_64-apple-macosx12.0

[patskovn]

Hello, @groue
What are your thoughts about Phase 1?
Can I implement it and make a PR? Or should I at first write more about Phase 2 and actors? Or, maybe, you will try to investigate another workarounds to avoid usage of private @_disfavoredOverload?

[groue]

[...] users are not blocked. This is important in order to evaluate the urgency of shipping async GRDB apis. Yes, a "native", aka ready-made, library under the GRDB umbrella would make users' life even easier (right inside GRDB itself, or in a new companion library "AsyncGRDB"). But we have time, and this is important.

Implicit in this sentence is the fact that GRDB has always grown this way: missing features are developed in applications first, according to my needs, the needs of my coworkers, the needs of users who are not shy, ask questions, send feature requests. This makes it possible to play with an idea in a few apps and different contexts, better understand the scope of the problem, and apply the rule of 3 before the library apis are enlarged. It has never been an urgency in this lib to provide apis that can be implemented in userland. This method has allowed the lib to slowly grow over time and still address a lot of application needs.

[groue]

Regarding your comments about actors: I'm still wondering how we could provide database serialization with an unisolated actor, without using something like a serial dispatch queue or a lock, and thus ruining the benefits of actors and the Swift runtime.

[groue]

A recent though about the Phase 2/3 AsyncDatabaseQueue or AsyncDatabasePool:

Those new kinds of "connections" would be very different from the existing DatabaseQueue and DatabasePool:

• No dependency of DispatchQueue
• Fully based on genuine Swift async constructs, including actors
• Very probably async-only (no sync database accesses)

At first I was reluctant about removing support for sync database accesses (my "sync is simpler than async" paragraph in this comment).

But if we design those new connections that that the user can bring back sync access by replacing AsyncDatabaseQueue/Pool with DatabaseQueue/Pool, with a small refactoring cost, then I don't see why we shouldn't explore this area.

[patskovn]

without using something like a serial dispatch queue or a lock

No-no, sorry, I mean additional protection layer with lock.
So my initial plan is

Also, DatabaseActor expose SerializedDatabase pucblicly like:

nonisolated public func withBlocking(connector: DatabaseReader & DatabaseWriter)

or

nonisolated public var syncConnector: DatabaseReader & DatabaseWriter

My intuition tells me that additional synchronisation layer via locks is extremely cheap, considering the fact that 99% of calls will be already serialised by actor. Synchronous access is very useful for initizlization step, for example. When user wants to synchronously create database connection and perform initial setup synchronously.
But, as I told you above, I am not 100% sure about performance despite the fact that lock-based approach looks promising.

P.S. Sorry for long responses. I am busy with immigrating to another country. This took huge amount of my attention and time in this covid times.

[groue]

I do not understand yet, but I think it is because of the name SerializedDatabase. Maybe you use it as a new name and concept? It happens that there already exist a class with this name, which is internal and not intended for public use. It provides the foundations for high-level connectors like DatabaseQueue and DatabasePool, and is responsible for very low-level concurrency primitives and checks (DispatchQueue juggling and checks for transaction boundaries).

Sorry for long responses.

That's totally OK. I'm very glad to discuss this delicate subject 👍

[groue]

Which apis should get an async version?

The "Phase 1" is making progress in the dev/async branch. Those database access methods now have an async version, all based on lower level apis with completion blocks:

• read
• unsafeRead
• write
• writeWithoutTransaction
• barrierWriteWithoutTransaction
• erase
• vacuum
• vacuum(into:)

About the async version of backup, I'm waiting for #1075 by @mallman to make progress. I'm also wondering wether the backup should wait for an access to the source database first, or an access to the destination database first (who do we await for?). Advice is welcome.

I did not start thinking about how we can build an async sequence out of ValueObservation.

I'm even less sure how DatabaseRegionObservation should be turned into an async sequence. It publishes database connections, and thus requires, unlike ValueObservation, that callbacks are executed in a very specific DispatchQueue (GRDB fatal errors when a database connection is not used in a blessed DispatchQueue). To me this sounds like there are many opportunities for misuses, and that users will have great difficulties using it correctly (i.e. without crashing). I hope I'm wrong, and that we can document a reliable way to use it.

Which other apis should become async?

[StuClift]

Do you know how this sentence from the doc of AsyncThrowingStream.init should be understood? It happens that raw ValueObservation callbacks onChange and onError run on a single serial DispatchQueue and are strictly serialized. But is a single DispatchQueue a single "context" that avoids all risks of "out of order delivery"?

It is a little confusing because it kind of leads you into doubt as it seems like an obvious statement if you take it to mean it as you've stated; which I assume must be the case.

@StuClift, I don't quite understand why the user would like to provide an explicit scheduler. It looks like the .immediate scheduler, which is very useful in a synchronous context, is out of place in the async/await context. Do I miss anything?

I added it to match the parameters available to start assuming this existed so the caller might have a reason to control the queue onChange(_:) handler was called from.

Without passing this through it wouldn't it be impossible to avoid hitting the main queue (even if only to call yield(_:) for each change?

Since this is a basic bridge to the GCD implementation; this might be a detail that matters to someone not awaiting these tasks through the MainActor.

[groue]

Thanks for your answers @StuClift. OK we'll keep the scheduler argument. All those async apis will be flagged as experimental anyway, so that we can easily introduce breaking changes whenever we better understand the user needs and figure out better implementation strategies.

Practically speaking:

On the naming side, I may prefer values over stream, following the example from Combine.

I could not figure out how to trigger the observation cancellation (onTermination) when I break the for loop:

for try await value in observation.values(in: dbQueue) {
    if someCondition {
        break // I expect this to cancel the observation eventually
    }
}

No matter how I tried, I'm currently unable to cancel the observation when the loop is broken. I have tried everything I could, including making very pessimistic hypothesis, such as considering the async runtime may only, and eventually, propagate cancellation when some parent Task, in which the loop is performed, is completed. But wrapping the broken loop in a Task, and waiting for the cancellation after the Task has been awaited yields no success.

Building the async sequence out of the ready-made Combine publisher didn't bring any support for cancellation either:

extension ValueObservation {
    public typealias Stream = Combine.AsyncThrowingPublisher<DatabasePublishers.Value<Reducer.Value>>
    
    func values(in reader: DatabaseReader) -> Stream {
        publisher(in: reader, scheduling: .async(onQueue: .main)).values
    }
}

I'm quite confused. I don't know how to turn a break into a cancellation. 😕 It's time to pick up a different topic for today.

[groue]

I'm quite confused.

I asked on the Swift forums: https://forums.swift.org/t/asyncsequence-break-and-cancellation/52787

[groue]

OK, it's a bug in the Swift 5.5 compiler: break is not correctly handled. That's unfortunate for infinite sequences such as ValueObservation 🤷‍♂️

[groue]

Is this too naïve as a wrapper?

Well, @StuClift, yes it is too naïve. I've learned it the hard way...

If you're interested, you may have a look at the current implementation:

GRDB.swift/GRDB/ValueObservation/ValueObservation.swift

Lines 214 to 315 in bbd7716

	#if swift(>=5.5)
	extension ValueObservation {
	// MARK: - Asynchronous Observation
	/// The database observation, as an asynchronous sequence of
	/// database changes.
	///
	/// [**Experimental**](http://github.com/groue/GRDB.swift#what-are-experimental-features)
	///
	/// - parameter reader: A DatabaseReader.
	/// - parameter scheduler: A Scheduler. By default, fresh values are
	/// dispatched asynchronously on the main queue.
	@available(macOS 12, iOS 15, tvOS 15, watchOS 8, *)
	public func values(
	in reader: DatabaseReader,
	scheduling scheduler: ValueObservationScheduler = .async(onQueue: .main))
	-> AsyncValueObservation<Reducer.Value>
	{
	AsyncValueObservation { onError, onChange in
	self.start(in: reader, scheduling: scheduler, onError: onError, onChange: onChange)
	}
	}
	}
	/// An asynchronous sequence of database changes.
	///
	/// [**Experimental**](http://github.com/groue/GRDB.swift#what-are-experimental-features)
	///
	/// Usage:
	///
	/// let observation = ValueObservation.tracking(Player.fetchAll)
	/// let dbQueue: DatabaseQueue: ...
	///
	/// // Each database change in the player prints "Fresh players: ..."
	/// for try await players in observation.values(in: dbQueue) {
	/// print("Fresh players: \(players)")
	/// }
	///
	/// See `ValueObservation` for more information.
	///
	/// - note: This async sequence never ends.
	@available(macOS 12, iOS 15, tvOS 15, watchOS 8, *)
	public struct AsyncValueObservation<Element>: AsyncSequence {
	public typealias AsyncIterator = Iterator
	typealias StartFunction = (
	_ onError: @escaping (Error) -> Void,
	_ onChange: @escaping (Element) -> Void)
	-> DatabaseCancellable
	private var start: StartFunction
	init(start: @escaping StartFunction) {
	self.start = start
	}
	public func makeAsyncIterator() -> Iterator {
	// This cancellable will be retained by the Iterator, which itself will
	// be retained by the Swift async runtime.
	//
	// We must not retain this cancellable in any other way, in order to
	// cancel the observation when the Swift async runtime releases
	// the iterator.
	var cancellable: AnyDatabaseCancellable?
	let stream = AsyncThrowingStream(Element.self, bufferingPolicy: .unbounded) { continuation in
	cancellable = AnyDatabaseCancellable(start(
	// onError
	{ error in
	continuation.finish(throwing: error)
	},
	// onChange
	{ [weak cancellable] element in
	if case .terminated = continuation.yield(element) {
	// TODO: I could never see this code running. Is it needed?
	cancellable?.cancel()
	}
	}))
	continuation.onTermination = { @Sendable [weak cancellable] _ in
	cancellable?.cancel()
	}
	}
	let iterator = stream.makeAsyncIterator()
	if let cancellable = cancellable {
	return Iterator(
	iterator: iterator,
	cancellable: cancellable)
	} else {
	// GRDB bug: there is no point throwing any error.
	fatalError("Expected AsyncThrowingStream to have started the observation already")
	}
	}
	/// An asynchronous iterator that supplies database changes one at a time.
	public struct Iterator: AsyncIteratorProtocol {
	var iterator: AsyncThrowingStream<Element, Error>.AsyncIterator
	let cancellable: AnyDatabaseCancellable
	public mutating func next() async throws -> Element? {
	try await iterator.next()
	}
	}
	}
	#endif

Practically speaking,

We use an AsyncThrowingStream, as you suggest, because it buffers elements. We need to buffer elements because nothing guarantees that the user can loop over the database changes fast enough. And I do not want to write code that buffers elements in an async environment. This looks very hard. So AsyncThrowingStream it is.

Unfortunately, I could never have a raw AsyncThrowingStream handle Task cancellation and end-of-loop (break, early return, etc) correctly, which means cancel the ValueObservation.

So the stream is wrapped in a thin wrapper, AsyncValueObservation (which is just what AsyncPublisher is to Publisher), whose sole purpose is to introduce weak references where there need to be.

Now my goal is to test even more ways to interrupt an async loop, and make sure that the observation is properly cancelled:

• immediate break
• break after a while
• sequence prefix
• Task cancellation
• etc.

Lesson learned: never assume that if one case is handled correctly, others are as well. Writing custom async sequences that must release resources when they end is not easy. This is a great opportunity to learn.

[groue]

Pull request for Phase 1: #1079

Public announcement: GRDB: Async/await edition is looking for testers

[patskovn]

That's wonderful. Will test it in my personal project. I mostly use only reads and writes and does not rely on value observation, so my test scope won't be wide 👍🏻

[groue]

GRDB 5.17.0 is out, with the "phase 1" of async/await, as framed by @Nekitosss.

Future evolutions will be driven by feedback on this initial release.