Skip to content

Commit 4ba0797

Browse files
phauslerphausler
authored
Transform the chunk and timer guides into a proposal (#231)
1 parent 7586d20 commit 4ba0797

File tree

1 file changed

+356
-0
lines changed

1 file changed

+356
-0
lines changed

‎Evolution/NNNN-chunk.md

Lines changed: 356 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,356 @@
1+
# Chunked & Timer
2+
* Proposal: [SAA-NNNN](https://github.com/apple/swift-async-algorithms/blob/main/Evolution/NNNN-chunk.md)
3+
* Author(s): [Kevin Perry](https://github.com/kperryua), [Philippe Hausler](https://github.com/phausler)
4+
* Status: **Implemented**
5+
* Implementation: [
6+
[By group](https://github.com/apple/swift-async-algorithms/blob/main/Sources/AsyncAlgorithms/AsyncChunkedByGroupSequence.swift),
7+
[On projection](https://github.com/apple/swift-async-algorithms/blob/main/Sources/AsyncAlgorithms/AsyncChunkedOnProjectionSequence.swift),
8+
[Count and signal](https://github.com/apple/swift-async-algorithms/blob/main/Sources/AsyncAlgorithms/AsyncChunksOfCountAndSignalSequence.swift)
9+
[Tests](https://github.com/apple/swift-async-algorithms/blob/main/Tests/AsyncAlgorithmsTests/TestChunk.swift)
10+
]
11+
[
12+
[Timer](https://github.com/apple/swift-async-algorithms/blob/main/Sources/AsyncAlgorithms/AsyncTimerSequence.swift) |
13+
[Tests](https://github.com/apple/swift-async-algorithms/blob/main/Tests/AsyncAlgorithmsTests/TestTimer.swift)
14+
]
15+
* Decision Notes:
16+
* Bugs:
17+
18+
## Introduction
19+
20+
Grouping of values from an asynchronous sequence is often useful for tasks that involve writing those values efficiently or useful to handle specific structured data inputs.
21+
22+
The groupings may be controlled by different ways but one most notable is to control them by regular intervals. Producing elements at regular intervals can be useful for composing with other algorithms. These can range from invoking code at specific times to using those regular intervals as a delimiter of events. There are other cases this exists in APIs however those do not currently interact with Swift concurrency. These existing APIs are ones like `Timer` or `DispatchTimer` but are bound to internal clocks that are not extensible.
23+
24+
## Proposed Solution
25+
26+
Chunking operations can be broken down into a few distinct categories: grouping according to a binary predicate used to determine whether consecutive elements belong to the same group, projecting an element's property to determine the element's chunk membership, by discrete count, by another signal asynchronous sequence which indicates when the chunk should be delimited, or by a combination of count and signal.
27+
28+
To satisfy the specific grouping by inteervals we propose to add a new type; `AsyncTimerSequence` which utilizes the new `Clock`, `Instant` and `Duration` types. This allows the interaction of the timer to custom implementations of types adopting `Clock`.
29+
30+
This asynchronous sequence will produce elements of the clock's `Instant` type after the interval has elapsed. That instant will be the `now` at the time that the sleep has resumed. For each invocation to `next()` the `AsyncTimerSequence.Iterator` will calculate the next deadline to resume and pass that and the tolerance to the clock. If at any point in time the task executing that iteration is cancelled the iteration will return `nil` from the call to `next()`.
31+
32+
## Detailed Design
33+
34+
### Grouping
35+
36+
Group chunks are determined by passing two consecutive elements to a closure which tests whether they are in the same group. When the `AsyncChunkedByGroupSequence` iterator receives the first element from the base sequence, it will immediately be added to a group. When it receives the second item, it tests whether the previous item and the current item belong to the same group. If they are not in the same group, then the iterator emits the first item's group and a new group is created containing the second item. Items declared to be in the same group accumulate until a new group is declared, or the iterator finds the end of the base sequence. When the base sequence terminates, the final group is emitted. If the base sequence throws an error, `AsyncChunkedByGroupSequence` will rethrow that error immediately and discard any current group.
37+
38+
```swift
39+
extension AsyncSequence {
40+
public func chunked<Collected: RangeReplaceableCollection>(
41+
into: Collected.Type,
42+
by belongInSameGroup: @escaping @Sendable (Element, Element) -> Bool
43+
) -> AsyncChunkedByGroupSequence<Self, Collected>
44+
where Collected.Element == Element
45+
46+
public func chunked(
47+
by belongInSameGroup: @escaping @Sendable (Element, Element) -> Bool
48+
) -> AsyncChunkedByGroupSequence<Self, [Element]>
49+
}
50+
```
51+
52+
Consider an example where an asynchronous sequence emits the following values: `10, 20, 30, 10, 40, 40, 10, 20`. Given the chunked operation to be defined as follows:
53+
54+
```swift
55+
let chunks = numbers.chunked { $0 <= $1 }
56+
for await numberChunk in chunks {
57+
print(numberChunk)
58+
}
59+
```
60+
61+
That snippet will produce the following values:
62+
63+
```swift
64+
[10, 20, 30]
65+
[10, 40, 40]
66+
[10, 20]
67+
```
68+
69+
While `Array` is the default type for chunks, thanks to the overload that takes a `RangeReplaceableCollection` type, the same sample can be chunked into instances of `ContiguousArray`, or any other `RangeReplaceableCollection` instead.
70+
71+
```swift
72+
let chunks = numbers.chunked(into: ContiguousArray.self) { $0 <= $1 }
73+
for await numberChunk in chunks {
74+
print(numberChunk)
75+
}
76+
```
77+
78+
That variant is the funnel method for the main implementation, which passes `[Element].self` in as the parameter.
79+
80+
### Projection
81+
82+
In some scenarios, chunks are determined not by comparing different elements, but by the element itself. This may be the case when the element has some sort of discriminator that can determine the chunk it belongs to. When two consecutive elements have different projections, the current chunk is emitted and a new chunk is created for the new element.
83+
84+
When the `AsyncChunkedOnProjectionSequence`'s iterator receives `nil` from the base sequence, it emits the final chunk. When the base sequence throws an error, the iterator discards the current chunk and rethrows that error.
85+
86+
Similarly to the `chunked(by:)` method this algorithm has an optional specification for the `RangeReplaceableCollection` which is used as the type of each chunk.
87+
88+
```swift
89+
extension AsyncSequence {
90+
public func chunked<Subject : Equatable, Collected: RangeReplaceableCollection>(
91+
into: Collected.Type,
92+
on projection: @escaping @Sendable (Element) -> Subject
93+
) -> AsyncChunkedOnProjectionSequence<Self, Subject, Collected>
94+
95+
public func chunked<Subject : Equatable>(
96+
on projection: @escaping @Sendable (Element) -> Subject
97+
) -> AsyncChunkedOnProjectionSequence<Self, Subject, [Element]>
98+
}
99+
```
100+
101+
The following example shows how a sequence of names can be chunked together by their first characters.
102+
103+
```swift
104+
let names = URL(fileURLWithPath: "/tmp/names.txt").lines
105+
let groupedNames = names.chunked(on: \.first!)
106+
for try await (firstLetter, names) in groupedNames {
107+
print(firstLetter)
108+
for name in names {
109+
print(" ", name)
110+
}
111+
}
112+
```
113+
114+
A special property of this kind of projection chunking is that when an asynchronous sequence's elements are known to be ordered, the output of the chunking asynchronous sequence is suitable for initializing dictionaries using the `AsyncSequence` initializer for `Dictionary`. This is because the projection can be easily designed to match the sorting characteristics and thereby guarantee that the output matches the pattern of an array of pairs of unique "keys" with the chunks as the "values".
115+
116+
In the example above, if the names are known to be ordered then you can take advantage of the uniqueness of each "first character" projection to initialize a `Dictionary` like so:
117+
118+
```swift
119+
let names = URL(fileURLWithPath: "/tmp/names.txt").lines
120+
let nameDirectory = try await Dictionary(uniqueKeysWithValues: names.chunked(on: \.first!))
121+
```
122+
123+
### Count or Signal
124+
125+
Sometimes chunks are determined not by the elements themselves, but by external factors. This final category enables limiting chunks to a specific size and/or delimiting them by another asynchronous sequence which is referred to as a "signal". This particular chunking family is useful for scenarios where the elements are more efficiently processed as chunks than individual elements, regardless of their values.
126+
127+
This family is broken down into two sub-families of methods: ones that employ a signal plus an optional count (which return an `AsyncChunksOfCountOrSignalSequence`), and the ones that only deal with counts (which return an `AsyncChunksOfCountSequence`). Both sub-families have `Collected` as their element type, or `Array` if unspecified. These sub-families have rethrowing behaviors; if the base `AsyncSequence` can throw then the chunks sequence can also throw. Likewise if the base `AsyncSequence` cannot throw then the chunks sequence also cannot throw.
128+
129+
##### Count only
130+
131+
```swift
132+
extension AsyncSequence {
133+
public func chunks<Collected: RangeReplaceableCollection>(
134+
ofCount count: Int,
135+
into: Collected.Type
136+
) -> AsyncChunksOfCountSequence<Self, Collected>
137+
where Collected.Element == Element
138+
139+
public func chunks(
140+
ofCount count: Int
141+
) -> AsyncChunksOfCountSequence<Self, [Element]>
142+
}
143+
```
144+
145+
If a chunk size limit is specified via an `ofCount` parameter, the sequence will produce chunks of type `Collected` with at most the specified number of elements. When a chunk reaches the given size, the asynchronous sequence will emit it immediately.
146+
147+
For example, an asynchronous sequence of `UInt8` bytes can be chunked into at most 1024-byte `Data` instances like so:
148+
149+
```swift
150+
let packets = bytes.chunks(ofCount: 1024, into: Data.self)
151+
for try await packet in packets {
152+
write(packet)
153+
}
154+
```
155+
156+
##### Signal only
157+
158+
```swift
159+
extension AsyncSequence {
160+
public func chunked<Signal, Collected: RangeReplaceableCollection>(
161+
by signal: Signal,
162+
into: Collected.Type
163+
) -> AsyncChunksOfCountOrSignalSequence<Self, Collected, Signal>
164+
where Collected.Element == Element
165+
166+
public func chunked<Signal>(
167+
by signal: Signal
168+
) -> AsyncChunksOfCountOrSignalSequence<Self, [Element], Signal>
169+
170+
public func chunked<C: Clock, Collected: RangeReplaceableCollection>(
171+
by timer: AsyncTimerSequence<C>,
172+
into: Collected.Type
173+
) -> AsyncChunksOfCountOrSignalSequence<Self, Collected, AsyncTimerSequence<C>>
174+
where Collected.Element == Element
175+
176+
public func chunked<C: Clock>(
177+
by timer: AsyncTimerSequence<C>
178+
) -> AsyncChunksOfCountOrSignalSequence<Self, [Element], AsyncTimerSequence<C>>
179+
}
180+
```
181+
182+
If a signal asynchronous sequence is specified, the chunking asynchronous sequence emits chunks whenever the signal emits. The signals element values are ignored. If the chunking asynchronous sequence hasn't accumulated any elements since its previous emission, then no value is emitted in response to the signal.
183+
184+
Since time is a frequent method of signaling desired delineations of chunks, there is a pre-specialized set of overloads that take `AsyncTimerSequence`. These allow shorthand initialization by using `AsyncTimerSequence`'s static member initializers.
185+
186+
As an example, an asynchronous sequence of log messages can be chunked into arrays of logs in four second segments like so:
187+
188+
```swift
189+
let fourSecondsOfLogs = logs.chunked(by: .repeating(every: .seconds(4)))
190+
for await chunk in fourSecondsOfLogs {
191+
send(chunk)
192+
}
193+
```
194+
195+
##### Count or Signal
196+
197+
```swift
198+
extension AsyncSequence {
199+
public func chunks<Signal, Collected: RangeReplaceableCollection>(
200+
ofCount count: Int,
201+
or signal: Signal,
202+
into: Collected.Type
203+
) -> AsyncChunksOfCountOrSignalSequence<Self, Collected, Signal>
204+
where Collected.Element == Element
205+
206+
public func chunks<Signal>(
207+
ofCount count: Int,
208+
or signal: Signal
209+
) -> AsyncChunksOfCountOrSignalSequence<Self, [Element], Signal>
210+
211+
public func chunked<C: Clock, Collected: RangeReplaceableCollection>(
212+
by timer: AsyncTimerSequence<C>,
213+
into: Collected.Type
214+
) -> AsyncChunksOfCountOrSignalSequence<Self, Collected, AsyncTimerSequence<C>>
215+
where Collected.Element == Element
216+
217+
public func chunked<C: Clock>(
218+
by timer: AsyncTimerSequence<C>
219+
) -> AsyncChunksOfCountOrSignalSequence<Self, [Element], AsyncTimerSequence<C>>
220+
}
221+
```
222+
223+
If both count and signal are specified, the chunking asynchronous sequence emits chunks whenever *either* the chunk reaches the specified size *or* the signal asynchronous sequence emits. When a signal causes a chunk to be emitted, the accumulated element count is reset back to zero. When an `AsyncTimerSequence` is used as a signal, the timer is started from the moment `next()` is called for the first time on `AsyncChunksOfCountOrSignalSequence`'s iterator, and it emits on a regular cadence from that moment. Note that the scheduling of the timer's emission is unaffected by any chunks emitted based on count.
224+
225+
Like the example above, this code emits up to 1024-byte `Data` instances, but a chunk will also be emitted every second.
226+
227+
```swift
228+
let packets = bytes.chunks(ofCount: 1024 or: .repeating(every: .seconds(1)), into: Data.self)
229+
for try await packet in packets {
230+
write(packet)
231+
}
232+
```
233+
234+
In any configuration of any of the chunking families, when the base asynchronous sequence terminates, one of two things will happen: 1) a partial chunk will be emitted, or 2) no chunk will be emitted (i.e. the iterator received no elements since the emission of the previous chunk). No elements from the base asynchronous sequence are ever discarded, except in the case of a thrown error.
235+
236+
## Interfaces
237+
238+
### Grouping
239+
240+
```swift
241+
public struct AsyncChunkedByGroupSequence<Base: AsyncSequence, Collected: RangeReplaceableCollection>: AsyncSequence
242+
where Collected.Element == Base.Element {
243+
public typealias Element = Collected
244+
245+
public struct Iterator: AsyncIteratorProtocol {
246+
public mutating func next() async rethrows -> Collected?
247+
}
248+
249+
public func makeAsyncIterator() -> Iterator
250+
}
251+
252+
extension AsyncChunkedByGroupSequence: Sendable
253+
where Base: Sendable, Base.Element: Sendable { }
254+
255+
extension AsyncChunkedByGroupSequence.Iterator: Sendable
256+
where Base.AsyncIterator: Sendable, Base.Element: Sendable { }
257+
```
258+
259+
### Projection
260+
261+
```swift
262+
public struct AsyncChunkedOnProjectionSequence<Base: AsyncSequence, Subject: Equatable, Collected: RangeReplaceableCollection>: AsyncSequence where Collected.Element == Base.Element {
263+
public typealias Element = (Subject, Collected)
264+
265+
public struct Iterator: AsyncIteratorProtocol {
266+
public mutating func next() async rethrows -> (Subject, Collected)?
267+
}
268+
269+
public func makeAsyncIterator() -> Iterator
270+
}
271+
272+
extension AsyncChunkedOnProjectionSequence: Sendable
273+
where Base: Sendable, Base.Element: Sendable { }
274+
extension AsyncChunkedOnProjectionSequence.Iterator: Sendable
275+
where Base.AsyncIterator: Sendable, Base.Element: Sendable, Subject: Sendable { }
276+
```
277+
278+
### Count
279+
280+
```swift
281+
public struct AsyncChunksOfCountSequence<Base: AsyncSequence, Collected: RangeReplaceableCollection>: AsyncSequence
282+
where Collected.Element == Base.Element {
283+
public typealias Element = Collected
284+
285+
public struct Iterator: AsyncIteratorProtocol {
286+
public mutating func next() async rethrows -> Collected?
287+
}
288+
289+
public func makeAsyncIterator() -> Iterator
290+
}
291+
292+
extension AsyncChunksOfCountSequence : Sendable where Base : Sendable, Base.Element : Sendable { }
293+
extension AsyncChunksOfCountSequence.Iterator : Sendable where Base.AsyncIterator : Sendable, Base.Element : Sendable { }
294+
295+
```
296+
297+
### Count or Signal
298+
299+
```swift
300+
public struct AsyncChunksOfCountOrSignalSequence<Base: AsyncSequence, Collected: RangeReplaceableCollection, Signal: AsyncSequence>: AsyncSequence, Sendable
301+
where
302+
Collected.Element == Base.Element,
303+
Base: Sendable, Signal: Sendable,
304+
Base.AsyncIterator: Sendable, Signal.AsyncIterator: Sendable,
305+
Base.Element: Sendable, Signal.Element: Sendable {
306+
public typealias Element = Collected
307+
308+
public struct Iterator: AsyncIteratorProtocol, Sendable {
309+
public mutating func next() async rethrows -> Collected?
310+
}
311+
312+
public func makeAsyncIterator() -> Iterator
313+
}
314+
```
315+
316+
### Timer
317+
318+
```swift
319+
public struct AsyncTimerSequence<C: Clock>: AsyncSequence {
320+
public typealias Element = C.Instant
321+
322+
public struct Iterator: AsyncIteratorProtocol {
323+
public mutating func next() async -> C.Instant?
324+
}
325+
326+
public init(
327+
interval: C.Instant.Duration,
328+
tolerance: C.Instant.Duration? = nil,
329+
clock: C
330+
)
331+
332+
public func makeAsyncIterator() -> Iterator
333+
}
334+
335+
extension AsyncTimerSequence where C == SuspendingClock {
336+
public static func repeating(every interval: Duration, tolerance: Duration? = nil) -> AsyncTimerSequence<SuspendingClock>
337+
}
338+
339+
extension AsyncTimerSequence: Sendable { }
340+
```
341+
342+
## Alternatives Considered
343+
344+
It was considered to make the chunked element to be an `AsyncSequence` instead of allowing collection into a `RangeReplaceableCollection` however it was determined that the throwing behavior of that would be complex to understand. If that hurdle could be overcome then that might be a future direction/consideration that would be worth exploring.
345+
346+
Variants of `chunked(by:)` (grouping) and `chunked(on:)` (projection) methods could be added that take delimiting `Signal` and `AsyncTimerSequence` inputs similar to `chunked(byCount:or:)`. However, it was decided that such functionality was likely to be underutilized and not worth the complication to the already broad surface area of `chunked` methods.
347+
348+
The naming of this family was considered to be `collect` which is used in APIs like `Combine`. This family of functions has distinct similarity to those APIs.
349+
350+
## Credits/Inspiration
351+
352+
This transformation function is a heavily inspired analog of the synchronous version [defined in the Swift Algorithms package](https://github.com/apple/swift-algorithms/blob/main/Guides/Chunked.md)
353+
354+
https://developer.apple.com/documentation/foundation/timer
355+
356+
https://developer.apple.com/documentation/foundation/timer/timerpublisher

0 commit comments

Comments
 (0)