Add chunked and windowed operators #1558

circusmagnus · 2019-09-20T08:32:13Z

Attempt to solve issue with lack of chunked operator with size limit: #1290

New functions:
fun <T, R> Flow<T>.chunked(size: Int, transform: suspend (List<T>) -> R): Flow<R>

fun <T, R> Flow<T>.windowed(
    size: Int, 
    step: Int, 
    partialWindows: Boolean, 
    transform: suspend (List<T>) -> R
): Flow<R>

Might be worth considering whether it is worth to accept transform parameter instead of just using
flow.chunked(2).map { ... }

circusmagnus · 2019-09-20T08:53:51Z

kotlinx-coroutines-core/common/src/flow/operators/Chunk.kt

+
+private fun <T> makeWindow(step: Int, buffer: MutableList<T>): List<T> {
+    val window = ArrayList(buffer)
+    repeat(step) { buffer.removeAt(0) }


It would be better if our buffer was ArrayDeque or similar entity which can be both iterated and has good performance, when removing elements at the head. Is it possible to expect ArrayDeque class, just as we expect ArrayList?

I vote yes. The poor performance and heap pressure that the current implementation will apply is very unsavory. You could also leave a reference to https://youtrack.jetbrains.com/issue/KT-21327 in a comment next to it.

There is a List implementation called RingBuffer, which would suit us perfectly. It is however private inside kotlin collections. About 100 lines long.

I could also use ArrayQueue from within coroutines lib, but it is not possible to iterate over its content without removing elements (so - buffer would have to be cleared and partially filled again after each windowed emission).

What is kotlin`s team advice in such circumstances?

Stay with underperformant current version?

Drop windowed operator until we get proper public collection available?

Just write our own RingBuffer without unneeded parts (we likely do not need toArray() function)

Use existing ArrayQueue?

Ok, I have decided to write our own RingBuffer impl (it is almost identical to the one from kotlin collections) and use it in windowed operator.

kotlinx-coroutines-core/common/src/flow/operators/Chunk.kt

JakeWharton · 2019-09-20T14:32:40Z

kotlinx-coroutines-core/common/src/flow/operators/Chunk.kt

+
+private fun <T> makeWindow(step: Int, buffer: MutableList<T>): List<T> {
+    val window = ArrayList(buffer)
+    repeat(step) { buffer.removeAt(0) }


I vote yes. The poor performance and heap pressure that the current implementation will apply is very unsavory. You could also leave a reference to https://youtrack.jetbrains.com/issue/KT-21327 in a comment next to it.

JakeWharton · 2019-09-20T14:33:48Z

kotlinx-coroutines-core/common/src/flow/operators/Chunk.kt

+import kotlinx.coroutines.flow.flow
+
+
+fun <T, R> Flow<T>.chunked(size: Int, transform: suspend (List<T>) -> R): Flow<R> = nonOverlappingWindowed(size, size, true, transform)


Shouldn't this default to an identity transform?

fvasco · 2019-09-20T15:11:08Z

kotlinx-coroutines-core/common/src/flow/operators/Chunk.kt

+            } else skipped++
+
+            if (window.size == size) {
+                emit(transform(window.toList()))


The current implementation provides a copy of the current window, this force a new list on each iteration.

Sequence and Iterable pass an ephemeral list to transform's argument, I propose to apply the same criteria.

A not ephemeral list should be provided only if the transformer is missing, so flow(...).windowed(10).toList() is a valid statement.

Will be always possible to emulate the current behaviour with: flow(...).windowed(...).map(...).

zach-klippenstein · 2019-09-23T17:50:10Z

Might be worth considering whether it is worth to accept transform parameter instead of just using flow.chunked(2).map { ... }

I would vote for removing the transform parameter in the interest of limiting each operator to a single concern. If this operator is being used because some downstream API actually needs lists, or will be immediately chained with a different transform like flatMap, then the transform parameter is unnecessary.

glasser · 2020-02-29T02:55:25Z

Any chance this will get merged any time soon? Or for that matter — does this look basically good enough that copying it into my app isn't a bad idea?

elizarov

Other general comments:

public modified shall be explicitly added to function.
New functions shall be marked with @FlowPreview.
API dump shall be updated (use ./gradlew apiDump)

kotlinx-coroutines-core/common/src/internal/RingBuffer.kt

kotlinx-coroutines-core/common/src/flow/operators/Chunk.kt

Sync develop with parent project

Add @FlowPreview Add public modifier Minor improvement in docs

Minor fix in test

kotlinx-coroutines-core/common/src/internal/RingBuffer.kt

kotlinx-coroutines-core/api/kotlinx-coroutines-core.api

...and remove them from ChunkKt file api

taer · 2020-05-19T16:35:54Z

kotlinx-coroutines-core/common/src/flow/operators/Chunk.kt

+        val toSkip = max(step - size, 0)
+        var skipped = toSkip
+
+        collect { value ->


Quick observation. If the upstream flow throws an exception, you could end up with a partial buffer here.
what are your thoughts of adding this before the collect

catch { e -> if(partialWindows){ emit(transform(buffer)) } throw e }.collect { value ->

I have fear, that we can loose exception this way or otherwise obscure it.

What if emitting partial buffer gets cancelled on some suspension point, before we rethrow exception? Or even, what if exception will be thrown from emitting our partial buffer, before we manage to rethrow original exception? Perhaps we could fix this by using a finally block somewhere. But, I think, it would be simpler, to just let exception propagate right away, and loose unfortunate partial buffer.

Loosing partial buffer is somewhat analogous to coroutineScope cancelling all its children immediately, on detecting exception anywhere. It is just more consistent this way.

circusmagnus · 2020-07-16T08:09:17Z

@elizarov What are the perspectives for this PR?

elizarov · 2020-07-16T14:44:07Z

We don't plan to have just a simple windowed and chunked operators from stdlib for Kotlin Flow, because asynchrony adds a whole another dimension of time period measurement. However, all time-based operators require a common, coherent design. We want to design a minimal number of basic composed primitives as opposed to many tailor-fit operators.
We'll be undertaking this design later.

For now, you are welcome to have your own implementations of the corresponding operators in your code.

Having said that, I'm closing this PR for now.

glasser · 2020-07-16T14:53:35Z

I'm confused about how this is time-based: isn't it based on chunk lengths?

circusmagnus · 2020-07-16T14:58:04Z

Sure thing. Wise decision not to chase reactive libraries for operator count.

elizarov · 2020-07-16T15:24:21Z

I'm confused about how this is time-based: isn't it based on chunk lengths?

@glasser This implementation, just like windowed and chunked from stdlib is not time-based, but size-based only. However, there are use-cases that call for time-based operators (see #1302). We don't want separate operators for that. We want to have a consistent chunking/windowing framework with flexible time- and size-based cutoffs, to minimize the number of concepts and operators users will have to learn.

wafisher · 2020-08-21T12:33:10Z

Just confirming that if and when this code is done, it would replace this solution?

mwspellman · 2021-07-27T09:49:31Z

Hey, just wanted to confirm if this is under the Apache 2.0 License? being the same as the kotlin libraries (https://kotlinlang.org/docs/contribute.html). Would like to use it, but wanted to clarify the licensing situation.

Thank you! I have found this very useful for my use case.

circusmagnus · 2021-08-03T22:49:29Z

@mwspellman As far as I am concerned (as a PR author), you can use this code under Apache 2.0 license.

Related issues: - Kotlin/kotlinx.coroutines#1290 - Kotlin/kotlinx.coroutines#902 - Kotlin/kotlinx.coroutines#1558 - Kotlin/kotlinx.coroutines#2378

circusmagnus commented Sep 20, 2019

View reviewed changes

JakeWharton reviewed Sep 20, 2019

View reviewed changes

fvasco reviewed Sep 20, 2019

View reviewed changes

circusmagnus force-pushed the flow-chunked-operators branch from 8378ee3 to 0ed0fd4 Compare September 27, 2019 00:00

circusmagnus mentioned this pull request Oct 2, 2019

Support natural batching in flow #902

Open

qwwdfsad force-pushed the develop branch 3 times, most recently from 4a49830 to aff8202 Compare March 10, 2020 17:27

elizarov suggested changes Mar 13, 2020

View reviewed changes

kotlinx-coroutines-core/common/src/internal/RingBuffer.kt Outdated Show resolved Hide resolved

kotlinx-coroutines-core/common/src/flow/operators/Chunk.kt Outdated Show resolved Hide resolved

circusmagnus and others added 5 commits March 18, 2020 23:24

Merge pull request #2 from Kotlin/develop

c0bf01b

Sync develop with parent project

Add chunked and windowed operators

86a503d

Add changes requested by R Elizarov

dbe84ea

Add @FlowPreview Add public modifier Minor improvement in docs

Generate api dump

6155c22

Use ArrayDeque from stdlib instead of custom RingBuffer

c74bfcf

Minor fix in test

circusmagnus force-pushed the flow-chunked-operators branch from 9ff3052 to c74bfcf Compare March 18, 2020 23:35

Change package name to match convention

07f72e9

circusmagnus requested a review from elizarov March 18, 2020 23:49

elizarov suggested changes Mar 25, 2020

View reviewed changes

kotlinx-coroutines-core/common/src/internal/RingBuffer.kt Outdated Show resolved Hide resolved

kotlinx-coroutines-core/api/kotlinx-coroutines-core.api Outdated Show resolved Hide resolved

circusmagnus added 2 commits March 27, 2020 00:25

Remove RingBuffer from coroutines lib

9392bbd

Update coroutines api with chunked and windowed operators

8e7ee37

...and remove them from ChunkKt file api

circusmagnus requested a review from elizarov March 31, 2020 21:00

taer reviewed May 19, 2020

View reviewed changes

elizarov closed this Jul 16, 2020

circusmagnus mentioned this pull request Aug 7, 2020

Flow.chunked operator with size limit #1290

Closed

circusmagnus mentioned this pull request Nov 11, 2020

Flow.chunked with specified time period #1302

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add chunked and windowed operators #1558

Add chunked and windowed operators #1558

circusmagnus commented Sep 20, 2019

circusmagnus Sep 20, 2019

JakeWharton Sep 20, 2019

circusmagnus Sep 20, 2019

circusmagnus Sep 29, 2019

JakeWharton Sep 20, 2019

JakeWharton Sep 20, 2019

circusmagnus Sep 20, 2019

fvasco Sep 20, 2019

circusmagnus Sep 20, 2019

zach-klippenstein commented Sep 23, 2019

glasser commented Feb 29, 2020

elizarov left a comment

taer May 19, 2020

circusmagnus Jul 16, 2020

circusmagnus Jul 16, 2020

circusmagnus commented Jul 16, 2020

elizarov commented Jul 16, 2020

glasser commented Jul 16, 2020

circusmagnus commented Jul 16, 2020

elizarov commented Jul 16, 2020

wafisher commented Aug 21, 2020

mwspellman commented Jul 27, 2021

circusmagnus commented Aug 3, 2021

		import kotlinx.coroutines.flow.flow


		fun <T, R> Flow<T>.chunked(size: Int, transform: suspend (List<T>) -> R): Flow<R> = nonOverlappingWindowed(size, size, true, transform)

Add chunked and windowed operators #1558

Add chunked and windowed operators #1558

Conversation

circusmagnus commented Sep 20, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zach-klippenstein commented Sep 23, 2019

glasser commented Feb 29, 2020

elizarov left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

circusmagnus commented Jul 16, 2020

elizarov commented Jul 16, 2020

glasser commented Jul 16, 2020

circusmagnus commented Jul 16, 2020

elizarov commented Jul 16, 2020

wafisher commented Aug 21, 2020

mwspellman commented Jul 27, 2021

circusmagnus commented Aug 3, 2021