Issues with retain cycle using AsyncStream in a Task

Question

Found this issue while working with the new Swift concurrency tools.

Here's the setup:

class FailedDeinit {
    
    init() {
        print(#function, id)
        task = Task {
            await subscribe()
        }
    }
    
    deinit {
        print(#function, id)
    }
    
    func subscribe() async {
        let stream = AsyncStream<Double> { _ in }
        for await p in stream {
            print("\(p)")
        }
    }
    
    private var task: Task<(), Swift.Error>?
    let id = UUID()
}

var instance: FailedDeinit? = FailedDeinit()
instance = nil

Running this code in a Playground yields this:

init() F007863C-9187-4591-A4F4-BC6BC990A935

!!! The deinit method is never called!!!

Strangely, when I change the code to this:

class SuccessDeinit {
    
    init() {
        print(#function, id)
        task = Task {
            let stream = AsyncStream<Double> { _ in }
            for await p in stream {
                print("\(p)")
            }
        }
    }
    
    deinit {
        print(#function, id)
    }
    
    private var task: Task<(), Swift.Error>?
    let id = UUID()
}

var instance: SuccessDeinit? = SuccessDeinit()
instance = nil

By moving the code from the method subscribe() directly in the Task, the result in the console changes to this:

init() 0C455201-89AE-4D7A-90F8-D6B2D93493B1
deinit 0C455201-89AE-4D7A-90F8-D6B2D93493B1

This may be a bug or not but there is definitely something that I do not understand. I would welcome any insight about that.

~!~!~!~!

This is crazy (or maybe I am?) but with a SwiftUI macOS project. I still DON'T get the same behaviour as you. Look at that code where I kept the same definition of the FailedDeinit and SuccessDeinit classes but used them within a SwiftUI view.

struct ContentView: View {
    @State private var failed: FailedDeinit?
    @State private var success: SuccessDeinit?
    var body: some View {
        VStack {
            HStack {
                Button("Add failed") { failed = .init() }
                Button("Remove failed") { failed = nil }
            }
            HStack {
                Button("Add Success") { success = .init() }
                Button("Remove Success") { success = nil }
            }
        }
    }
}


class FailedDeinit {
    
    init() {
        print(#function, id)
        task = Task { [weak self] in
            await self?.subscribe()
        }
    }
    
    deinit {
        print(#function, id)
    }
    
    func subscribe() async {
        let stream = AsyncStream<Double> { _ in }
        for await p in stream {
            print("\(p)")
        }
    }
    
    private var task: Task<(), Swift.Error>?
    let id = UUID()
}

Answer 1

This doesn't really have anything to do with async/await or AsyncStream. It's a perfectly normal retain cycle. You (the FailedDeinit instance) are retaining the task, but the task refers to subscribe which is a method of you, i.e. self, so the task is retaining you. So simply break the retain cycle just like you would break any other retain cycle. Just change

    task = Task {
        await subscribe()
    }

To

    task = Task { [weak self] in
        await self?.subscribe()
    }

Also, be sure to test in a real project, not a playground, as playgrounds are not indicative of anything in this regard. Here's the code I used:

import UIKit

class FailedDeinit {

    init() {
        print(#function, id)
        task = Task { [weak self] in
            await self?.subscribe()
        }
    }

    deinit {
        print(#function, id)
    }

    func subscribe() async {
        let stream = AsyncStream<Double> { _ in }
        for await p in stream {
            print("\(p)")
        }
    }

    private var task: Task<(), Swift.Error>?
    let id = UUID()
}

class ViewController: UIViewController {

    override func viewDidLoad() {
        super.viewDidLoad()
        var instance: FailedDeinit? = FailedDeinit()
        instance = nil
   }
}

Answer 2

Consider the following:

task = Task {
    await subscribe()
}

It is true that introduces a strong reference to self. You can resolve that strong reference with:

task = Task { [weak self] in
    await self?.subscribe()
}

But that is only part of the problem here. This [weak self] pattern only helps us in this case if either the Task has not yet started or if it has finished.

The issue is that as soon as subscribe starts executing, despite the weak reference in the closure, it will keep a strong reference to self until subscribe finishes. So, this weak reference is prudent, but it is not the whole story.

The issue here is more subtle than appears at first glance. Consider the following:

func subscribe() async {
    let stream = AsyncStream<Double> { _ in }
    for await p in stream {
        print("\(p)")
    }
}

The subscribe method will keep executing until the stream calls finish. But you never finish the stream. (You don’t yield any values, either. Lol.) Anyway, without anything in the AsyncStream, once subscribe starts it will never complete and thus will never release self.

So let us consider your second rendition, when you create the Task, bypassing subscribe:

task = Task {
    let stream = AsyncStream<Double> { _ in }
    for await p in stream {
        print("\(p)")
    }
}

Yes, you will see the object be deallocated, but you are neglecting to notice that this Task will never finish, either! So, do not be lulled into into a false sense of security just because the containing object was released: The Task never finishes! The memory associated with that Task will never get released (even if the parent object, FailedDeinit in your example, is).

This all can be illustrated by changing your stream to actually yield values and eventually finish:

task = Task {
    let stream = AsyncStream<Double> { continuation in
        Task {
            for i in 0 ..< 10 {
                try await Task.sleep(nanoseconds: 1 * NSEC_PER_SECOND)
                continuation.yield(Double(i))
            }
            continuation.finish()
        }
    }

    for await p in stream {
        print("\(p)")
    }

    print("all done")
}

In this case, if you dismiss it while the stream is underway, you will see that the AsyncStream continues until it finishes. (And, if you happen to be doing this inside a method, the object in question will also be retained until the task is canceled.)

So, what you need to do is to cancel the Task if you want the AsyncStream to finish. And you also should implement onTermination of the continuation in such a manner that it stops the asynchronous stream.

But, the result is that if I cancel this when the view controller (or whatever) is released, then my example yielding values 0 through 9 will stop and the task will be freed.

It all comes down to what your AsyncStream is really doing. But in the process of simplifying the MCVE and removing the contents of the AsyncStream, you simultaneously do not handle cancelation and never call finish. Those two, combined, manifest the problem you describe.