is there an idiomatic way to process an infinite s...

# announcements

[Andrew Gazelka]

is there an idiomatic way to process an infinite sequence in parallel?

[gildor]

Sequences are sequential by definition, so there is no way to make them parallel.
For parallel sequence of events you can use Flow

[Andrew Gazelka]

No what I am saying is something along the lines of {numbers from sequence} -> {split into subsequeces} -> {do computation in parallel on each sub sequence}

[gildor]

Just split into subsequeces, run computation on own thread

[Andrew Gazelka]

well I guess what I am asking is if there is an idiomatic way to split the sequence

[Andrew Gazelka]

would be cool if there were something like a grouping that could be used for an inf sequence

[gildor]

You can use Sequence.chunked, which emits list of required size

[gildor]

not exactly subsequence. but also sequence is not really needed, because anyway you should buffer some amount of items to process

[Andrew Gazelka]

buffering seems like it would complicate the code

[Andrew Gazelka]

I am simply trying to have integers [2, inf)

[Andrew Gazelka]

and split and see if the number is prime concurrently

[gildor]

One call of chunked(SIZE) doesn’t look complicated

[Andrew Gazelka]

ah ok fair

[gildor]

but I still think that it’s pretty limited solution, essentially, there is no way to return result, it’s fire and forget

[Andrew Gazelka]

hmmm

[gildor]

I believe Flow (or any other Reactive framework) is much better choice

[Andrew Gazelka]

private fun primes() {
    val threads = 6
    val infSequence = generateSequence(2) { it + 1 }
    runBlocking {
        for (i in 0 until threads) {
            launch {
                infSequence.filter { it % threads == i }
                    .filter { isPrime(it) }
                    .forEach { println(it) }
            }
        }
    }
}

eh

[Andrew Gazelka]

yeah probably

[gildor]

isPrime always blocking, cannot be used in parallerl

[gildor]

it will block sequence

[Andrew Gazelka]

it is my own implementation

[Andrew Gazelka]

even tho probably less efficient I’m guessing

[gildor]

You already have coroutines, why not use flow that gives everything out of the box

[gildor]

and much more idiomatic

[Andrew Gazelka]

how would you idiomatically do this with a flow

[gildor]

just make isPrime suspend function that runs on particular dispatcher with target parallelism

[Andrew Gazelka]

I am so confused

[Andrew Gazelka]

why would isPrime be suspending… it doesn’t need to suspend?…

[gildor]

To run on target dispatcher

[gildor]

because it’s your bottleneck

[Andrew Gazelka]

ah I see I think

[gildor]

there are a few ways actually, one more is just blocking isPrime, but you split the list and use flatMapConcat

[gildor]

Actually, for this particular task with prime number, Java’s Fork/Join framework probably the best solution, just because it’s optimized for CPU bound tasks and have good job stealing algorithm, but to make it performant you have to batch work to threads by splitting sequence by chunks

[Dennis Schröder]

Take the time to carefully read this ... https://kotlinlang.org/docs/reference/coroutines/flow.html

Flow is what you need.

[gildor]

Honestly not sure that pure Flow will be faster for this use case with prime number ir you launch coroutine per each item, with chunked/windowed it possible to optimise it much more, but those operators are not available in 1.3 yet

[Matteo Mirk]

for deeper support try asking also on #coroutines