I have a database actor which vends transaction actors. These can run serially or concurrently. Through a transaction the caller can update the database async-wise.
Updating the database schema should only be done when there are no active transactions. When a schema migration is started I should therefore wait for all current transactions to complete (and refuse starting new ones).
A first approach can be seen in the snippet below.
Using Taks.sleep works but seems a pre-actor way of doing things.
With Task.yield the loop runs basically continously. (Should have seen that coming, I guess, because a transaction just sits around waiting for work or a final commit call.)
I found this thread but that is about emitting values in a producer/consumer pattern. In my case I have to wait and then run once before returning to the caller. (This will probably evolve into a Migration actor which has exclusive access so that multiple migration steps can run in series but asynchronously.)
Other approaches:
I considered storing the migration for later use. Each time a transaction is completed I would check the number of active transactions. If zero I run the stored migration. But then migrate func would return immediately. The caller might assume the new schema is in place and do followup "calls".
I also thought about using TaskGroup. Each time a transaction starts, add something to a group. When the migration starts, it would await the group.
Any patterns I could/should utilise here?
Thanks in advance for your advice/suggestions.
actor MyActor
{
enum State
{
case busy
case ready
}
var active = 0
var state = State.ready
func migrate() async throws
{
guard state == .ready else { return }
state = .busy
print("waiting \(active)")
while active > 0
{
// await Task.yield() //runs way too often
try await Task.sleep(nanoseconds: 1_000_000_000 * 1) //ugly
print("check \(active)")
}
print("migrating \(active)")
state = .ready
}
func test(_ msg: String, sleep: UInt64 = 2) async throws
{
guard state == .ready else { return }
active += 1
print("sleep \(msg)")
try await Task.sleep(nanoseconds: 1_000_000_000 * sleep)
print("awake \(msg)")
active -= 1
}
}
and to run it:
static func main() async
{
do
{
let actor = MyActor()
let taskA = Task { try await actor.test("foo", sleep: 3) }
let taskB = Task { try await actor.test("bar") }
try await Task.sleep(nanoseconds: 1_000_000_0 * 5) //otherwise migrate run first
let taskM = Task { try await actor.migrate() } //should wait for A and B
let taskC = Task { try await actor.test("food") } //should not run
let taskD = Task { try await actor.test("bars") } //should not run
print("done")
try await Task.sleep(nanoseconds: 1_000_000_000 * 5)
}
catch { print(error) }
}
There's no existing pattern for this kind of "wait", other than the pseudo-wait via Task.sleep. I think the forthcoming Task.select might be helpful for this kind of thing, in the minimal case of suspending on a single outstanding task.
The only current solutions AFAIK are: use one of the with…Continuation functions at the deepest level of waiting, essentially bridging from the old completion-handler pattern back to Swift concurrency; or, use a custom AsyncStream to deliver completions of events you're waiting on.
You do face a second problem, though. Using await inside an actor is very bad news because actors are re-entrant. At the suspension point of such an await, other actor-isolated functions and accessors can run, so the awaiting function is not completely isolated as you might expect. This is fine if your actor has no breakable invariants that can be compromised across an internal suspension point, but that's very hard to ensure — and it pretty much takes away the isolation advantages that actors were intended to provide.
To implement this functionality on your own you need to track two things basically how many active transactions are running and suspend upcoming update database schema task if your transactions task count greater than zero. Also, you need to keep track if any migration is running and suspend your incoming transactions until migration is complete. You can use continuations to suspend any task and store the continuation to be resumed/cancelled later.
Sorry I haven’t responded thus far.
First of all, thank you and QuinceyMorris for the replies.
I got something that seems to work fairly quickly after QuinceyMorris reply. I’ll study your solution in detail too,
It’s summer time and I have less available free time unfortunately. I’ve also been working on non-concurrency related issues. Currently I’m figuring out how to best deal with the schema and type safety.. The DB stores everything as enum values and strings with a strict order while the graph model uses types, CodingKeys, etc. There is more friction than I like between those worlds. So I’m refactoring while treating it as a purely local database before adding the rest back to it.
Ah well, once 5.7 is released and I have refactored it to use the cool new features, then perhaps I’ll post about my progress and experiences thus far.
It’s an interesting project and I’m still at it.
I realize this thread is a bit older, but this is still a very common "pain point" when trying to pause actor execution without blocking threads.
The Task.sleep loop (busy waiting) described was basically the only way to do it back, but it’s definitely not ideal for battery life or CPU usage.
For anyone landing here today (Swift 5.9+), the solution is to use Custom Actor Executors. This lets you move the "wait" logic out of your functions and into the scheduling layer itself.
One way to implement this is by backing your actor with an OperationQueue. You may also use GCD queues directly but you will have to manually manage the suspended state using manaul locking.
The Custom Executor
We define an executor that funnels all the actor's work into an OperationQueue. When we want to "freeze" the actor, we just tell the queue to suspend.
import Foundation
final class PausableExecutor: SerialExecutor {
private let queue: OperationQueue = {
let q = OperationQueue()
q.maxConcurrentOperationCount = 1
q.name = "com.myapp.db-actor-queue"
return q
}()
func asUnownedSerialExecutor() -> UnownedSerialExecutor {
UnownedSerialExecutor(ordinary: self)
}
// The runtime calls this whenever the actor has a job to run.
func enqueue(_ job: UnownedJob) {
let unownedExecutor = self.asUnownedSerialExecutor()
// We wrap the job in a BlockOperation and add it to the queue.
// If queue.isSuspended is true, this operation will just sit there
// until we resume, effectively "pausing" the actor.
queue.addOperation {
job.runSynchronously(on: unownedExecutor)
}
}
// MARK: - Controls
func suspend() {
queue.isSuspended = true
}
func resume() {
queue.isSuspended = false
}
}
The Actor Implementation
You just need to tell your actor to use this executor.
actor DatabaseActor {
nonisolated let executor = PausableExecutor()
nonisolated var unownedExecutor: UnownedSerialExecutor {
executor.asUnownedSerialExecutor()
}
// Implment the migration func here safely. However, make sure that you always access the executor from a nonisolated scope to ensure no deadlocks happen.
}
This is an interesting direction to take. It does give me pause, because I think it opens the door to preventing forward progress in a way that will lead to deadlocks.
I wonder if a closure-based withSuspended , instead of discrete suspend/resume, could help? But even so, I don’t know if that would be enough.
The problem happens if you call suspend()while you are currently running code on the actor. To ensure no deadlocks happen, you must call suspend() from outside the actor entirely(non isolated scope).
For example, use a separate “manager“ class that calls a non isolated func that pauses the actor.
Although I didnt get how the closure based approach will work, but it still has to follow that same rule: don't call it from inside the actor!
if !active {
if let taskThatMakesItActive {
await taskThatMakesItActive
} else {
taskThatMakesItActive = Task { ... }
await taskThatMakesItActive
}
}
This task could probably be replaced with an AsyncStream<Never>.makeStream(), so we can escape the continuation and cancel that when active becomes true. (Since the stream won't receive any values, it's fine to iterate it multiple times)
Usually what I've been doing in these kind of scenarios is spawning a new task for each "transaction" and keeping track of the handles so I can await one of them (typically the last one)[1] instead of polling constantly.
For the OP's code, I'd be something like this:
actor MyActor {
var activeTasks: OrderedDictionary<Int, Task<Void, any Error>> = [:]
var counter: Int = 0
var state = State.ready
func migrate() async throws {
guard state == .ready else { return }
state = .busy
// Find the last scheduled task.
while let activeTask = activeTasks.values.last {
// Wait for the task to complete before polling again.
try await activeTask.value
await Task.yield()
}
print("Migrating...")
print("Migrated!")
state = .ready
}
func trackedTransaction(_ body: @escaping () async throws -> Void) async throws {
let id = counter
counter += 1
let newTask = Task {
try await body()
}
activeTasks[id] = newTask
try await newTask.value
activeTasks[id] = nil
}
func test(_ msg: String, sleep: UInt64 = 2) async throws {
guard state == .ready else { return }
try await trackedTransaction {
print("sleep \(msg)")
try await Task.sleep(for: .seconds(sleep))
print("awake \(msg)")
}
}
enum State {
case busy
case ready
}
}
It's not a great solution, but I've never needed to use it for something critical, and it satisfies the constraints I usually want to impose:
No arbitrary time-based delays or sleep()s.
No infinite polling to check if a condition became true.
Happy path where there's no active work to wait for can proceed immediately.
And so far it's been working fine.
I know the tasks are not completed in FIFO order, but it's an acceptable heuristic to assume the last scheduled one will be the most likely to finish last, and getting it wrong just means that the loop will need to poll the set of active tasks an additional time. ↩︎
I think you are right that this is necessary, but I’m not certain it is sufficient. I believe this construct will be subject to all the deadlock risks of other forms of serial queues. I’m pretty sure that includes more than one actor with this kind of executor, as well as interactions with other non-actor based systems.
I like it, in that it’s a novel solution! But, I’m worried that it will ultimately make it too easy to accidentally run afoul of the forward progress principle.
