Working in agency development provides the benefit of frequently starting new projects, allowing us to start fresh - with a clean slate.
It means we can take advantage of new APIs provided by Apple without worrying about supporting older iOS versions with legacy code. We can also build upon what worked well in previous projects while eliminating any previous shortcomings.
For instance, in our most recent projects, we incorporated Swift's modern concurrency, which significantly streamlined the code in our application. It resulted in easy-to-follow code and reduced the likelihood of errors compared to similar projects that used Combine.
Using Modern Concurrency and one thoughtful comment in a code review from my colleague helped me realize a lot about the obvious - a behavior of AsyncSequence1, as well as the safety issues AsyncPublisher poses.
It all started with a test
Code written with async await. The next step was to write a test for its functionality. Since the test involved observing a @Published property in a SignInViewModel, my initial instinct was to utilize Combine. One of the resulting tests took the following form. Implementation details of SignInViewModel are beyond the scope of this article. Its primary function involves validating the inputs (email and password) when certain conditions are met. In this instance, the error is shown under the text field when it stops being the first responder.
The test passed, and I was happy and submitted it as a part of a merge request with other sign-in functionality and tests.
During the code review, a colleague suggested using async-await in the test instead of Combine. We briefly discussed it, and both believed the result would be equivalent. The idea led me down a rabbit hole that took me some time to navigate. The rewritten tests using Swift’s for await loop were failing randomly. It made me question the validity of my code and the implementation details of the SignInViewModel. When I looked into the details, the failure was caused by a timeout – there was a different amount of received values than expected. As I debugged the code, I discovered that observing viewError in the for-await loop resulted in various counts of the received values. I discovered that the behavior of AsyncSequence was different from what I had expected, despite reading various books, articles, and documentation.
Let’s put rubber boots on and start digging.
The first dig: Comparing AsyncPublisher vs subscribing via Combine
I will leave SignInViewModel for the rest of this article. Let us start with something simple and continuously increase the complexity to learn what is happening here; why the tests were failing and why they did so randomly. Let us leave the documentation behind, observe their behavior and devise the reasoning once we get there.
Let us first start with preparing a publisher. PasshthroughSubject should be a good candidate for this:
And let us observe it with basic sink in Combine and via for await loop, and print the received value.
There is no observable content, so let us send something through the pipe:
Now, if you look into the console, you will see lots of printed results– and if you look deep into the prints, most probably will be the same amount of prints from both sources – AsyncPublisher being the same as a basic sink.
So, why was it failing the test and not here? I got the notion of what is happening based on the initializer on AsyncStream. The initializer looks like this:
What caught my attention is the bufferingPolicy parameter, which decides how to buffer unused sequence items – whether they should stay there (and be buffered) or go. Let us test this assumption that the difference of for-await loop and sink would be noticeable when we introduce some delay with some heavy workload – such as finding the nth prime number in form of calling primeCount(100)2.
After introducing the delay, we finally have a minimum observable example of the issue. As you can see, the number of prints no longer corresponds - some numbers in the for-await loop are skipped, resulting in varying results across runs. It's worth noting that the highest number in the async for loop often ends in numbers other than 99, such as 97, which is the last number processed correctly.
Why does the for-await loop behave like this? It likely uses a buffering policy that discards some elements that aren't immediately consumed. While the exact buffering policy setting isn't crucial to this investigation, as you will see in a section about AsyncStream, it buffers a lot.
However, what's most important is how the for-await loop works. It only continues the iteration after the closure of the current iteration is over, ensuring that the for loop iterates synchronously while getting the items in a sequence asynchronously.
Counting differences
While it is interesting to see the number of prints varies, it still requires us to look at the code and try to guesswork. Let us prepare a counter to have a more reliable representation of results. I made the counter using an actor that takes a type as a parameter so that we can have a clear console:
The implementation is as follows:
As you can see, after every call to addToCount() is made, the counter increases by one and the count of the type is printed out, which means that the last print of a given type will be the relevant one for us.
Let us test it and see the results, first by instantiating the counters and using them inside our existing code (in Combine fashion, we have to create a Task to call asynchronous function):
Now we can finally see in numbers how much has been lost. The Combine counter’s last print is 100. For async, it is 83. Therefore, 17 % have been lost on my M1 Max without optimizations.
Making for await loop of AsyncPublisher work
We now have established a baseline and a test environment. We have observed that Combine works without any issues. Now let's focus on fixing the for-await loop to obtain the desired results.
Fortunately, the fix is straightforward. We need to create a Task and handle the for-await loop asynchronously without waiting. Doing so will cause us to lose the benefits of sequential code execution that Modern Concurrency provides. However, it will fix the issue that we have been dealing with.
This solution works because creating a Task is not a computationally expensive action that would cause us to lose an integer. Under normal circumstances, this approach should work well.
AsyncPublisher is not safe
In the previous section, we got to a point in which everything worked the way we desired – by asynchronously processing the items in the sequence, we were getting the same results from observing the for-await loop as with the Combine. Does this mean we can safely use AsyncPublisher without worrying about losing inputs? Unfortunately, no.
All it takes is to send a value concurrently, and we will be back where we started. For instance, in my working environment, after the following code was executed, the counter in Combine was at 1000, on async with 695. Not great.
Observing with a for-await loop is unfortunately not secure and can lead to data loss. Depending on the situation, there may be some ways to mitigate this issue. For example, if we know how many items we will receive, we can use collect() on the publisher. However, I do not see any good way to make the for-await loop work in the same way as subscribing to the publisher in Combine, where we can always handle it through async code. I have tried several other options, such as creating an AsyncIterator out of AsyncPublisher, but that did not yield any positive results.
These limitations are inherent to the for-await loop, as it only works on one thread and does not make any guarantees. However, we can leverage this limitation to our advantage in certain use cases, particularly when using it with AsyncSequence - or its implementation, AsyncStream.
AsyncStream to the rescue
AsyncStream is a built-in implementation of AsyncSequence in an asynchronous context. It allows awaiting its results in a sequence as well as allows publishing it (yield in async terminology) to sequence at any time with its continuation.
It is very easy to instantiate it, for our purposes, we can do something like this:
As you can see, I am storing the asyncStream in a variable for us to use later in the code, as well as storing the continuation. Although we unsafely unwrap the variables, the code is safe because AsyncStream returns the continuation in its init immediately (the provided closure does not escape the current scope).
Observing the AsyncStream is simple – we just need to make a simple change in our for-await loop. Since AsyncStream is an AsyncSequence, we can iterate it directly, just like any other Collection (such as Array). The only difference is that we do it asynchronously.
Continuing with our example, here's how we would yield the integers to the continuation:
With AsyncStream, the counter gets to 100 every time. The same applies when we count to 1000 with the code we already used. Not an integer gets wasted.
We no longer have to even process the items asynchronously, so we can simplify the observing code to just:
And why is it that AsyncStream works? It is because of its buffering policy - which is defined in its initialization. By default, the policy is unbounded, meaning every integer gets buffered regardless of how long it takes to process a single iteration.
We could very well destroy the ability of our AsyncStream if we wanted. For example, if we initialized AsyncStream with bufferingPolicy: .bufferingNewest(1), the counter would show only 2 instead of 1000. It supports the hypothesis that AsyncPublisher has a buffering policy, possibly a substantial one, but we still need to understand its limits for situations where we need it.
AsyncStream is a hero in our story because it allows us to set the buffering policy to fit our needs and expectations. It enables us to observe asynchronous values in an asynchronous context.
But what if we needed to observe a Combine publisher?
The question is: How do we asynchronously observe a Combine publisher? As I showed you earlier, we can achieve this by simply calling Task {} in the sink of the subscriber. However, this method doesn't allow us to use the ordered processing of the items that a for-await loop does.
If we create our own publisher, my recommendation is to use AsyncStream instead of Combine publishers, as long as we don't need to use reactive magic for other parts of the code.
If we cannot create our own AsyncStream or we want to observe a publisher we did not create, we can always create an AsyncStream, subscribe to the publisher, and yield the values to the AsyncStream.
And – to speak less abstractly – I think this may be a nice way to signify the differences between observing a stream in a for-await loop and awaiting the results and observing in Combine with sink in showing the information messages in the application.
Let’s say we have a following function from which we show the infoBar:
If we observe the publisher in Combine, we get the following result. As you can see, the messages get shown over each other - not the greatest behavior.
But we can always do it from AsyncStream and await the result of the completion via withCheckedContinuation, like this:
And the result is a readable message that is shown one after the other. We get the behavior we want without any difficulties. And what is best, we can use Modern Concurrency, so the code is quite readable.
Final Thoughts
Incomplete documentation is the cause of our limited understanding of Swift's Modern Concurrency, leaving us with much still to learn. As a result, we can only presume and guess. While we could learn more by analyzing the source code, doing so may not be a good use of our time, and it may be difficult to grasp all the details just by looking at it.
Therefore, further investigation in this area is needed. I hope more people will delve deeper instead of only scratching the surface.
In the meantime, let's not rush into rewriting every bit of code in async-await without careful consideration. We should always check our assumptions to avoid introducing unnecessary bugs.
1 In essence, AsyncPublisher enables observing any result from a publisher in an async for–await loop. You can find more in documentation https://developer.apple.com/documentation/combine/asyncpublisher
2 You can find a gist here: https://gist.github.com/lvalenta/07770221e7d653f6cfad6f2bda63af73 – It is just a random function a colleague found
