Hey Everyone! so is I am getting started with mult...

# flow

[Ansh Tyagi]

Hey Everyone! so is I am getting started with multiprocessing and wanted to know if there is something similar to pythons Pool.map() in kotlin? Or an example showing how can we make a process pool and then run an iterable callback in it.

[Big Chungus]

flowOf{... Your publisher goes here ...}.map{}

[CLOVIS]

@Big Chungus that is sequential, I don't think that's what they asked for.

@Ansh Tyagi in Kotlin we tend not to use sub processes much because they're very expensive. Instead, we use threads. Because of its design, Python cannot efficiently use threads.

If you want parallelism with threads, you can either go the Java 8 approach with Stream, or the KotlinX.Coroutines approach with Channel. Flow is sequential.

[CLOVIS]

If you really want to use sub processes, it's possible, but it's far from being as clean as these solutions.

[Big Chungus]

You can make use of intermediate flow operators like buffer ir shareIn

[Big Chungus]

Scratch that, the only way you can make fliw consumption parallel is to launch a bunch of coroutines in collect {}

[Richard Gomez]

@Big Chungus wouldn't something like this work? May be a horrible abuse of the system though, lol.

myFlow.map { 
   async { methodToRun(it) }
 }
 .buffer()
 .collect {
    it.await()
  }

[Big Chungus]

You need to convert it to broadcast channel and then launch multiple coroutines that read from it. I think there's flow.broadcastIn() or similar utility for that

[CLOVIS]

@Richard Gomez Nope. Flow is a suspending iterator, it will only execute the map lambda for a value after the previous value has finished the pipeline.

You could write this:

someFlow
  .toList()
  .map { scope.async { ... } }
  .asFlow()
  .map { it.await() }

That would truly be parallel (depending on the scope of course), however the toList() call forces all previous results to be known before the parallel operation starts.

With Java Streams (not multiplatform), you can write:

someList
  .parallelStream()
  .map { ... }

That will execute in parallel, however it is not coroutine-aware, so it might not be convenient depending on your use case.

There used to be a map function on Channel (which would be the equivalent to Stream), but I recall it was removed, I don't remember why.

[DALDEI]

I have used this code for a year in production with no problem

https://gist.github.com/serpro69/3eccd8f38718d05949da047225e08ece

A simple real example :


suspend fun flowUserReports(userId: UserId) = flowUserReportRoots(userId)
.parallelMap(100, 16) {
safe {
logTime(log,"readSingle<QuickSightReportRequest.Data>") {
UserReportInfoItem(it.dataFile, it.dataFile.parent, getReportRequest(it.dataFile) )
}
}
}.filterNotNull()
.flowOn(FlowUtil.IO)

[CLOVIS]

Note that that implementation has a lot of overhead, so only use it if what you're doing in the loop is actually slow enough that it's worth it (measure it!)