hi im curious how people are using actors in the real world, especially with regard to propery changes and reacting to those?
does swift basically recommend i just use a @MainActor class with @Observation?
(i've done some searching but cant find a good answer) so im curious what the pattern for actors in this kind of scenario are...
any good examples or recommendations with actors?
I'm waking up this year-old thread because I have the same question. I'm struggling with how to implement an actor property that can be monitored by subscribers, without resorting to the obsolete(?) register-a-callback-function pattern.
Problem statement: I have an actor that manages a network connection of some sort. It has a status property, an enum with cases like connecting, connected, stopped, etc. Clients [plural!] should be able to monitor this property reactively.
So I expose another property statusStream: any AsyncSequence<Status,Never>. Internally this is an AsyncChannel<Status> kept in a stored property, which I send() status changes to.
(BTW, AsyncStream has an API that's completely unsuited to this usage, as it assumes that all the items will be produced synchronously from within a closure. The docs do say that the continuation can escape the closure, but you have to do this by storing it in a captured variable. This API makes no sense to me at all. Fortunately AsyncChannel is friendlier.)
Problem 1: AsyncSequence is not Sendable(!), and an actor can't expose non-Sendable var properties. This seems like a weird omission for a protocol whose entire purpose is to bridge between tasks.
Workaround: Change the property type to any AsyncSequence<Status,Never> & Sendable.
Problem 2: This works great until I add a second observer/iterator. Now neither observer gets all the status changes, in fact each change is sent to only one of them. This seems to be an undocumented limitation of AsyncChannel: an item pushed to it is just passed to a single iterator, whichever one calls/called next first. This makes it useless for the observer pattern!
I've looked through the other tools in swift-async-algorithms but AFAICT none of them support multiple iterators. The documentation is pretty limited so I may be missing something.
[rant]IMO this is a serious design flaw in AsyncSequence: the ambiguity about what multiple iterators do. I know it inherits this from Sequence, which explicitly leaves it undefined what happens if you iterate it multiple times; but in a synchronous sequence this feels more intuitive, plus we have a broader protocol Collection for things that support multiple iteration. In a concurrent system one consumer of an async sequence doesn't really know whether there's some other consumer out there iterating it too. A single-iterator constraint would work much better with an API like Go's channels, where the factory function creates a paired producer and a consumer at the same time.[/rant]
Update: I forgot to add that @Observable is apparently useless for this because it's incompatible with Sendable; at least, I haven't been able to get them to work together.
Async-Algorithms does have a share operator, although it’s hard to find since it’s not listed in the README.
Can you elaborate on why it isn’t useful in your case?
AsyncSequence is a very well designed API, particularly in its flexibility. However, I wouldn’t recommend implementing your own AsyncSequence, as it’s not straightforward to get right. In essence, there are two requirements:
next() method.nil (or throwing an error) from next(), the iterator enters a terminal state, and all future calls to next() must return nil.Furthermore, AsyncSequence also states the following:
An
AsyncSequencedoesn’t generate or contain the values; it just defines how you access them.
This gives you the flexibility to implement both single and multi-consumer sequences.
This is a reasonable complaint. I think we should better highlight this in the documentation. I’m currently working on a reimplementation of Async{Throwing}Stream, and improving documentation will be part of that effort, but it will come as a follow-up at a later time.
Are there any creative ways SE-0475 might help?
If Observations depends on an Observable then you might run into the limitation that Observable is not currently available on actor types.
Could your Observable inside your actor be a Sendable class type that uses a lock or mutex to synchronize some internal state manually? And then your actor can call directly to the Observable that then publishes the Observations for your subscribers?
In my case observers are UI observers (view models) and because of ObservableObject / Observable Macro architecture they have to be MainActor.
This is the code I typically use to propagate changes from model actors to view models:
actor Model {
var modelState: ModelState {
didSet {
if modelState != oldValue {
sendChangeToObservers()
}
}
}
func sendChangeToObservers() {
let state = self.modelState
DispatchQueue.main.async {
MainActor.assumeIsolated {
self.observers.forEach {
observer.modelStateChanged(state)
}
}
}
}
}
// this is either `ObservableObject` with published properties or `Observable macro`.
@MainActor class ObservableObjectViewModel {
var privateViewModelState: ModelState
var viewModelState: ModelState {
get { privateViewModelState }
set { ... }
}
func modelStateChanged(_ state: ModelState) {
privateViewModelState = state
}
}
That's if there's a need to have model / viewModel separation to begin with...
In simple cases I'd just keep the model itself in MainActor and feed views directly from that w/o the need of intermediate view models.
FWIW, contrary to a popular misbelieve you could initiate network request on main actor and have it completed on the main actor – that does not mean that the UI thread will be blocked or anything during the network request execution – the actual "networking" will happen in background.
@MainActor final class TotallyFine: NSObject, URLSessionDelegate {
var session: URLSession!
override init() {
super.init()
session = URLSession(configuration: .default, delegate: self, delegateQueue: .main)
}
func doit(request: URLRequest) {
let task = session.dataTask(with: request) { data, response, error in
// request is completed on the main queue because we configured our session that way
// ...
}
// request is started on the main queue
task.resume()
}
}
Having some non trivial processing (bytes to image conversion for example, or perhaps large JSON to object conversion) might justify a need of having that processing being done in background though.
THANK YOU! That's exactly what I was looking for.
It’s a very simple API :) AsyncIterator itself is great. What’s problematic is a protocol that's a factory for them, with unspecified semantics.
I think that's less of a problem for the regular Sequence because it’s rare to create multiple iterators on one at the same time; so the ambiguity is limited to what happens if you iterate it again after it’s finished, which is usually clear from what it’s modeling: an Array can be iterated over and over, a socket stream obviously can't.
With async you very much do have the likelihood of multiple iterators at once, and usually unless the implementation goes to a lot of extra work the result is that they interfere with each other and lose items, which is bad.
I don’t see how. The @Observable macro adds boilerplate getters and setters, which do not appear to be concurrency-safe; that’s why it doesn’t work with actors. If I use an @unchecked Sendable class instead, it appeases the compiler but it doesn’t fix the race conditions. The necessary locking would need to be inside the boilerplate accessors, right?
Have you tried to get a manually written conformance to the Observable protocol to work? It is possible to manually conform, and then even wrap that conformance up into a custom macro, like @ObservableActor.
The fundamental tension here is an actor establishes a async interface, and Observable is a mechanism for synchronous notifications. The good news is that async functions can make synchronous function calls!
So the rough shape of a solution is to have an entity that "observes" the async stream by for awaiting on it , then mutate a stored property. Now we are back to the synchronous world. Layer observability on top of this stored property, and you are golden.
But all of this analysis is probably symptom of the original design. To me, protecting your state with a Mutex on a class seems to be the path of lesser resistance.
That's actually the opposite of my issue, which is how to create the async stream of events when I change the model's properties.