Possibly a stupid question, but I'm new to AsyncSequence and friends, and I don't understand how to terminate the following loop. The function is taken from Apple's AVCam demo app:
actor CaptureService {
// ...
private func observeNotifications() {
Task {
for await error in NotificationCenter.default.notifications(named: AVCaptureSession.runtimeErrorNotification)
.compactMap({ $0.userInfo?[AVCaptureSessionErrorKey] as? AVError }) {
// code that uses actor's properties
}
}
}
}
I was trying to adapt this module to my own app and noticed that it's leaking CaptureService objects. The demo app never closes the camera view and it's why the leak is not important there.
Without the notification handling loop it's released fine, no leaks.
I believe this issue is present in the demo app available for everyone to download from here, so whatever the fix is, I think the demo code should be fixed too.
Interesting that this kind of pattern is in this demo.
Task captures self strongly in its closure, because you are setting properties of it (this is kind of invisible because of the @_implicitSelfCapture attribute of the closure in the initializer).
I don't know the implementation details, so I am assuming by how NotificationCenter worked before AsyncSequence: this sequence will never implicitly terminate.
Therefore the task will never terminate, and your actor instance will never be released. You have to get a handle of your task and cancel it at appropriate times, if you have/want to stick with unstructured Concurrency.
It's what I thought (cancel the task), but that means the actor should expose an additional method for shutting down everything, which is not ideal given that things in Swift are usually freed automatically.
If this is the only way, it means the new form of notification handling is just not great, even though tbf it looks neat.
Hm, I would not attach this problem to notification handling. This is just the reality of unstructured Concurrency. You have to take care of the Tasks lifetime on your own.
If you'd use structured Concurrency, in your case probably TaskGroups, then someone else would have to manage the lifetime of it. SwiftUI for instance helps you to manage lifetimes with the .task view modifier.
Sorry the demo is really big. Finding a more hands-on solution probably takes some time, I think the only tip I can give you here is try to stick to structured Concurrency primitives.
It is, and it's unfortunate that its modules are not reusable. Or at least some comments should be added? Because most of the apps that use camera would likely spawn it in a temporary view and close it after use, meaning the modules should provide a clean shutdown.
A SwiftUI Camera view would be pretty cool to have I think. But I think the main problem is, that AVCaptureSession is just such a massive API, it is probably quite hard to boil it down to a simple view. (I've tried several times, but those just ignore most of the edge cases handled in the demo)
This can feel like a chore (it is, honestly), but at least it has an advantage: you may be able to avoid performing unnecessary work more consistently. For example, it's common to have apps wasting resources reacting to some trigger and updating a view that is no longer visible but is still being referenced.
You'd have more control over when to stop reacting to notifications, instead of relying on "whenever the enclosing object is deallocated". Its lifetime could be unexpectedly extended by things you didn't account for.
In simple cases you can also get this to work by relying on [weak self], and the memory will be freed automatically:
private func startObservingNotifications() {
let avErrorSequence = NotificationCenter.default.notifications(
named: AVCaptureSession.runtimeErrorNotification
)
.compactMap { notification in
notification.userInfo?[AVCaptureSessionErrorKey] as? AVError
}
Task { [weak self] in // â ïž self's lifetime could be extended by something else in the code
for await error in avErrorSequence {
// No strong reference to self held so far
guard let self else { return }
// Now holding a strong reference to self!
self.handleError(error)
// â Temporal strong reference to self released here when local "self" goes out of scope
}
}
}
But in complex cases this is full of subtle behaviors and it's possible to reintroduce the issue in unexpected ways.
One of the more obvious ones (but still easy to miss!) is that strongifying self before the for await loop would have the same memory leak issue as the original code:
// ...
Task { [weak self] in
guard let self else { return } // â ïž Strong reference to self created here
for await error in avErrorSequence {
self.handleError(error)
}
// â Strong reference would only be released here, but for await loop never ends
}
Code that runs for a long time while holding the strong reference to self may also extend the lifetime of self for quite a bit:
// ...
Task { [weak self] in
for await error in avErrorSequence {
guard let self else { return }
await self.longRunningHandleError(error)
// â ïž Strong reference to self won't be released until longRunningHandleError(...) finishes
}
}
Even if the task only hold self strongly for a little bit, if there are multiple such for await loops all potentially referencing the same object, it's conceivable that at any given time there's always at least one task holding a strong reference to self, never releasing self (I created a gist a while ago demonstrating this behavior, can be run on a Playground). This is particularly true if code inside a for await loop can cause other for await loops to get new values.
The part with overlapping strongified self is especially easy to overlook.
Got me thinking, how practical for await .. in really is, if the potential of causing problems is so high?
For example, I use AsyncStream to pass real-time status changes from the camera into the view. As long as for await is in the view's .task { } modifier I'm safe since .task can cancel itself once the view goes out of sight. But what if some day I move the loop somewhere else and lose track of how or whether it is cancelled? No way for the compiler to help me at compile time, and no easy way to debug other than with some printing or the Leak instrument.
Memory leaks are in fact some of the nastiest of bugs because they are mostly invisible but may cause inconvenience to users.
Maybe it's a matter of getting used to it (I mean, for await). Anyway, was more of a rhetorical question.
This is a hot topic right now and for loops (AsyncStream) kinda force to use unstructured tasks to be able to proceeded with other work that has to be done. Especially if few AsyncStreams needs to be consumed at the same time from the same starting point.
Holding a handle to a Task to âmanuallyâ cancel it at appropriate time is probably the best way with the current available tools.
Beside some rare (in my case) use cases where e.g. SwiftUI's .task can be used.
I wonder if it would be possible to have a tool built-in Swift where the lifetime of self would determine automatic task cancellation like:
Task(withLifetimeOf: self) { self in
for await ⊠in asyncStream { ⊠}
}
This kind of concept exists in Kotlin.
But this would require that Task would not strongly held self but on the same time allow to use self in the Task.
// Edit
Iâm not in front of mac, but when I think more about this it might be possible to mix the âdeferredâ strong self within for loop to accomplish a helper like:
func withLifetime<âŠ>(of object: T: AnyObject, consume: any AsyncSequenceâŠ, with operation: ⊠(T, ElementâŠ) async â> âŠ) -> Task<âŠ> {
Task { [weak object] in
for await item in sequence {
guard let object else { return }
await operation(self, item)
}
}
}
SwiftUI's .task, or the Observation framework in general (if the use case is observing the state from the UI) are great for this.
This isn't really different from weak self captures, is it?
The problem I see is a Task must be able to extend the lifetime of self (by holding a strong reference to self), because the alternative is to write code that effectively works like this:
for await value in asyncSequence {
self?.databaseA.remove(value)
self?.databaseB.add(value)
}
But then you run into the truly horrifying bugs that happen when your code doesn't enforce transactionality (in the above example, it could be very bad if remove() is called but add() isn't).
And if you allow a Task to strongly hold self even for a little bit, then you must accept the possibility of such a system extending the lifetime of self forever (for example, by having multiple tasks in a way there's always at least one Task holding self at any given time). So weak self captures (or any system tying Task cancellation to the lifetime of self) don't look to me like a robust way to prevent the issue.
Operation would be treated as a transaction on captured object. Of course you can easily make bad decisions if e.g. operations loops over another AsyncSequence but this is always a problem regardless the pattern being used.
The problem is not so much that a for await loop explicitly loops over another for await sequence, but rather that very innocent code like this:
for await value in asyncSequence {
self.foo = value
}
May then trigger another for await loop somewhere else in the code that was awaiting new values of foo, and it's really hard to know whether this is an issue without knowing every detail about the codebase.
Best option IMO is to try to use never-ending for await loops sparingly.
FWIW, SwiftUI + Observation sidesteps this problem completely by being declarative. When SwiftUI reacts to changes to an @Observable property to update the UI it doesn't[1] write back to the model, so it can't cause other observation tasks to fire.
So in a way, it is a pattern that does avoid the problem.
The problem of .task modifier (like onAppear) is that can be called more times during view lifecycle. In case of async sequence will be cancelled when view is not visible.
Apple should made view lifecycle funcs for many use cases like observing things.
extension AsyncSequence {
@discardableResult
@MainActor
func assign<Root>(
to keyPath: ReferenceWritableKeyPath<Root, Self.Element>,
weakOn object: Root
) -> Task<Void, any Error> where Root: AnyObject, Self.Element: Sendable {
withMainActorLifetime(of: object, consuming: self, forEach: { object, element in
object[keyPath: keyPath] = element
})
}
}
@discardableResult
@MainActor
func withMainActorLifetime<Instance, OperationError, Stream>(
of object: Instance,
consuming stream: Stream,
forEach operation: @MainActor @escaping (
_ object: Instance,
_ element: Stream.Element
) throws(OperationError) -> Void
) -> Task<Void, any Error>
where Instance: AnyObject,
OperationError: Error,
Stream: AsyncSequence,
Stream.Element: Sendable
{
Task { @MainActor [weak object] in
for try await element in stream {
guard !Task.isCancelled,
let object
else { return }
try operation(object, element)
}
}
}
and for more general usage
@discardableResult
func withLifetime<Instance, OperationError, Stream>(
// isolation: isolated (any Actor)? = #isolation <- ??? can it help?
of object: Instance,
consuming stream: Stream,
forEach operation: sending @escaping (
_ object: Instance,
_ element: Stream.Element
) throws(OperationError) -> Void
) -> Task<Void, any Error>
where Instance: AnyObject & Sendable,
OperationError: Error,
Stream: AsyncSequence & Sendable,
Stream.Element: Sendable
{
Task { [weak object] in
for try await element in stream {
guard !Task.isCancelled,
let object
else { return }
try operation(object, element)
}
}
}
This might be a different topic, but I wonder if we can associate the isolation of the past operation closure with the Task that is created with existing tools. I would like to tell the compiler that the instance being sent and the operation belongs to the same isolation, and enforce the same isolation on the unstructured Task. As I did with withMainActorLifetime where instance and operation does not have to be Sendable nor sending as both belong to @MainActor .
This means, however, that if the task is cancelled and there are no more new elements in the stream, the closure will be stuck forever, unless I'm missing something? If so, you could say it's no big deal just a few bytes in memory, but it would still be a memory leak.
@nonameplum none of your links answer my question though.
My point was that your solution (withLifetime) doesn't provide automatic cancellation of the stream, which means there's still a potential memory leak. I know that I can cancel the stream explicitly, but that's not always possible like for example in the case of iOS Notifications I showed in the original post above.
You could use AnyCancellable to tie the cancellation of an unstructured Task to the lifetime of the containing object.
import Combine
actor CaptureService {
// ...
private var cancellables = Set<AnyCancellable>()
private func observeNotifications() {
let task = Task { [weak self] in
for await error in NotificationCenter.default.notifications(named: AVCaptureSession.runtimeErrorNotification)
.compactMap({ $0.userInfo?[AVCaptureSessionErrorKey] as? AVError }) {
// code that uses actor's properties
}
}
AnyCancellable { task.cancel() }.store(in: &cancellables)
}
}
You need to explicitly use a weak self capture to workaround the implicit self capture on Task, and obviously this solution only works on Apple platforms (where Combine is available). But this has worked pretty well for me in the past.