Hello guys, I'm afraid I've run into similar issue as was #977.
My scenario is actually very similar as there: streaming big data from sharded database, that means running multiple DB queries in parallel where each query returns some data (potentially GBs in production so we need to it have streamed).
There is a huge difference between processing this data with Observable (Monix) and FS2 Stream, visible even in unit tests with some funny-small data. Based on the finding I've implemented a repo with benchmark and the scenario described more into deep.
The actual code is quite short and naive but something has to be wrong there:
Stream
.range(1, items)
.mapAsyncUnordered[IO, Array[Byte]](parallelism) { _ =>
IO {
DataGenerator.get(size)
}
}
.map(_.length) // don't want to drain the stream so I "process" the data this way
.reduce(_ + _)
.compile
.toList
.runAndWait
The same implementation with Observable is about 30 times faster:
Observable
.range(1, items)
.mapParallelUnordered[Array[Byte]](parallelism) { _ =>
Task {
DataGenerator.get(size)
}
}
.map(_.length)
.reduce(_ + _)
.toListL
.runAndWait
Results (shortened, see the repo):
[info] Benchmark (items) (parallelism) (size) (threads) Mode Cnt Score Error Units
[info] StreamingBenchmark.testFs2IO 500 5 10000 10 avgt 2 90,676 ms/op
[info] StreamingBenchmark.testMonix 500 5 10000 10 avgt 2 3,665 ms/op
What am I doing wrong? :-)
P.S. I am currently running full test with current setup (as is in the repo) but it will take few hours. I'll post it here later but intermediate results show no surprise.
jendakol
All 38 comments
As I've promised, here are results for current test settings:
[info] Benchmark (items) (parallelism) (size) (threads) Mode Cnt Score Error Units
[info] StreamingBenchmark.testFs2FromMonix 500 10 1000 10 avgt 20 17,905 ± 0,248 ms/op
[info] StreamingBenchmark.testFs2FromMonix 500 10 10000 10 avgt 20 17,698 ± 0,165 ms/op
[info] StreamingBenchmark.testFs2FromMonix 1000 10 1000 10 avgt 20 36,730 ± 0,518 ms/op
[info] StreamingBenchmark.testFs2FromMonix 1000 10 10000 10 avgt 20 37,042 ± 0,369 ms/op
[info] StreamingBenchmark.testFs2FromMonix 5000 10 1000 10 avgt 20 181,284 ± 2,563 ms/op
[info] StreamingBenchmark.testFs2FromMonix 5000 10 10000 10 avgt 20 174,722 ± 1,155 ms/op
[info] StreamingBenchmark.testFs2IO 500 10 1000 10 avgt 20 171,731 ± 1,701 ms/op
[info] StreamingBenchmark.testFs2IO 500 10 10000 10 avgt 20 172,712 ± 1,802 ms/op
[info] StreamingBenchmark.testFs2IO 1000 10 1000 10 avgt 20 349,657 ± 4,402 ms/op
[info] StreamingBenchmark.testFs2IO 1000 10 10000 10 avgt 20 352,718 ± 2,779 ms/op
[info] StreamingBenchmark.testFs2IO 5000 10 1000 10 avgt 20 1730,664 ± 16,963 ms/op
[info] StreamingBenchmark.testFs2IO 5000 10 10000 10 avgt 20 1745,693 ± 22,126 ms/op
[info] StreamingBenchmark.testFs2Task 500 10 1000 10 avgt 20 207,058 ± 3,564 ms/op
[info] StreamingBenchmark.testFs2Task 500 10 10000 10 avgt 20 207,450 ± 4,449 ms/op
[info] StreamingBenchmark.testFs2Task 1000 10 1000 10 avgt 20 412,818 ± 4,315 ms/op
[info] StreamingBenchmark.testFs2Task 1000 10 10000 10 avgt 20 401,606 ± 6,903 ms/op
[info] StreamingBenchmark.testFs2Task 5000 10 1000 10 avgt 20 2052,445 ± 23,115 ms/op
[info] StreamingBenchmark.testFs2Task 5000 10 10000 10 avgt 20 2049,175 ± 31,826 ms/op
[info] StreamingBenchmark.testMonix 500 10 1000 10 avgt 20 2,278 ± 0,136 ms/op
[info] StreamingBenchmark.testMonix 500 10 10000 10 avgt 20 2,912 ± 0,093 ms/op
[info] StreamingBenchmark.testMonix 1000 10 1000 10 avgt 20 4,372 ± 0,086 ms/op
[info] StreamingBenchmark.testMonix 1000 10 10000 10 avgt 20 4,558 ± 0,194 ms/op
[info] StreamingBenchmark.testMonix 5000 10 1000 10 avgt 20 19,841 ± 0,527 ms/op
[info] StreamingBenchmark.testMonix 5000 10 10000 10 avgt 20 19,108 ± 0,298 ms/op
(I intentionally don't add HW configuration because it doesn't seem relevant to me - my point is not about absolute _time_ but about it's many times slower than it should be.)
jendakol
on 20 Aug 2018
@jendakol I think this is related to fairly ineffective implementation of mapAsyncUnordered which is implemented as marge of single-streams with parJoin. Thanks for reporting this.
We shall look on this before 1.x release to improve the implementation, to make this as fast as possible
pchlupacek
on 20 Aug 2018
No stress but when do you expect to release it? :-) I guess the deadline presented at https://github.com/functional-streams-for-scala/fs2/milestone/13 is not valid anymore ;-)
BTW I'd be glad to help you with fixing this but it's far out of my skills :smiley:
jendakol
on 20 Aug 2018
I hope this will make its way in 1.0. So guess like september 2018
pchlupacek
on 20 Aug 2018
@jendakol also if there is quick workaround required you can try something in these terms :
val s: Stream[F, O] = ???
val def myWork(o: O): F[Unit] = ???
Stream.eval(Semaphore(MaxRunningInParallel)).flatMap { guard =>
s.evalMap { o =>
guard.acquire >>
Concurrent[F].start(myWork(o).attempt >> guard.release)
}
}
That would likely give you the fastest implementation possible.
If you will by chance end up with this solution, PR is always welcome :-)
pchlupacek
on 20 Aug 2018
I think that will only work properly if myWork is F[Unit]
SystemFw
on 20 Aug 2018
ah ok then this will require additional queue. :
val s: Stream[F, O] = ???
val def myWork(o: O): F[Unit] = ???
Stream.eval(Semaphore(MaxRunningInParallel)).flatMap { guard =>
Stream.eval(async.unboundedQueue[A]).flatMap { queue =>
queue.dequeue.concurrently(
s.evalMap { o =>
guard.acquire >>
Concurrent[F].start(myWork(o).flatMap(queue.enqueue1).attempt >> guard.release)
}
)
}}
still the performance will be "almost" identical to one with F[Unit]
pchlupacek
on 20 Aug 2018
@SystemFw perhaps we shall introduce Tap version of the combinator for F[Unit] case that will be still a bit "faster" :-)
pchlupacek
on 20 Aug 2018
Yeah I thought so too but tbh I'd like to see more numbers justifying a proliferation of combinators :)
SystemFw
on 20 Aug 2018
@SystemFw yeh. I bet we can sort of believe that ppl did that already as a reasoning why cats have traverse and traverse_ . By enqueuing to queue, we in fact create non-negligible contention point, that may be completely unnecessary :-)
Now I have idea, if, we couldn't somehow use this new combinator to actually define parJoin :-)
(I mean the Tap variant)
pchlupacek
on 20 Aug 2018
TBH I'm kind of a lost in your discussion however I tried proposed workaround:
val s = Stream.range(1, items)
import cats.syntax.all._
Stream
.eval(Semaphore[IO](parallelism))
.flatMap { guard =>
Stream.eval(fs2.async.unboundedQueue[IO, Array[Byte]]).flatMap { queue =>
queue.dequeue.concurrently {
s.evalMap { _ =>
guard.decrement >>
Concurrent[IO].start {
IO {
DataGenerator.get(size)
}.flatMap(queue.enqueue1).attempt >> guard.increment
}
}
}
}
}
.map(_.length)
.reduce(_ + _)
.compile
.toList
.runAndWait
Unfortunately the code never finished, I guess there is deadlock somewhere :smile: Is it possible that it actually returns an infinite Stream (from the queue) which causes my code to never end? Unfortunately applying take(10) on it didn't help. I am missing something...
jendakol
on 21 Aug 2018
there's no deadlock but you basically wait on the empty queue forever.
Stream
.eval(Semaphore[IO](parallelism))
.flatMap { guard =>
Stream.eval(fs2.async.unboundedQueue[IO, Option[Array[Byte]]]).flatMap { queue =>
queue.dequeue.unNoneTerminate.concurrently {
s.evalMap { _ =>
guard.decrement >>
Concurrent[IO].start {
IO {
DataGenerator.get(size)
}.flatMap(x => queue.enqueue1(x.some)).attempt >> guard.increment
}
}.onFinalize(queue.enqueue1(None))
}
}
}
.map(_.length)
.reduce(_ + _)
.compile
.toList
.runAndWait
try this
Edit: tbh I'm not very convinced this will change things perf wise, join is implemented very similarly under the hood (plus mechanics to handle interruption, errors and so on, which aren't handled in this snippet). I kinda suspect that by the time all these concerns are handled, you end up with join again
SystemFw
on 21 Aug 2018
@SystemFw the problem with join is that it has a lot of machinery, like interruption, resource / scope cleanup and cetera. And for one element Streams, this may be significant cost.
pchlupacek
on 21 Aug 2018
mm maybe you're right, we can experiment and see how much we can get away with for mapAsyncUnordered
SystemFw
on 21 Aug 2018
just a remark to your code, I think you have to add finalizer .onFinlaize(queue.enqueue1(None))
pchlupacek
on 21 Aug 2018
yeah, just realised that :joy:
SystemFw
on 21 Aug 2018
yeah I am thinking to start with something like evalMapChunk(f: a => F[A]) that will essentially produce something like Eval[F[Chunk[A]]]], and then evalMapChunkUnordered that will produce unordered chunk.
Also I am thinkig for 1.0 on something like evalMapChunkEarly, that will operate on chunks but will start emitting the A as soon as they will be evaluated, possibly unchuking the original chunks.
Still this combinator which doesn't care about chunks at all can be significant win, once implemented properly.
pchlupacek
on 21 Aug 2018
also for this type workloads, I will submit soon PR with distribute and broadcast that perhaps will make things so much more easier to write :-)
pchlupacek
on 21 Aug 2018
@SystemFw Yeah, that's what I thought - the queue generates an infinite stream. I'm not experienced enough with FS2 to know unNoneTerminate but it's great.
Thx guys, I've finally got it working. I've rewritten it to use for-comprehension (for my better understanding):
val itemsStream: Stream[Pure, Int] = Stream.range(1, items)
val resultsStream: Stream[IO, Int] = for {
semaphore <- Stream.eval(Semaphore[IO](parallelism))
queue <- Stream.eval(fs2.async.unboundedQueue[IO, Option[Array[Byte]]])
results <- {
queue.dequeue.unNoneTerminate.concurrently[Unit] {
itemsStream
.evalMap[IO, Unit] { _ =>
for {
_ <- semaphore.decrement
_ <- {
Concurrent[IO].start {
for {
data <- IO { DataGenerator.get(size) }
_ <- queue.enqueue1(data.some)
_ <- semaphore.increment
} yield ()
}
}
} yield ()
}
.onFinalize(queue.enqueue1(None))
}
}
} yield {
results.length
}
resultsStream.reduce(_ + _).compile.toList.runAndWait
This works however results are still quite sad compared to Observable:
[info] Benchmark (items) (parallelism) (size) (threads) Mode Cnt Score Error Units
[info] StreamingBenchmark.testFs2FromMonix 1000 10 1000 10 avgt 10 33,436 ± 1,706 ms/op
[info] StreamingBenchmark.testFs2Workaround 1000 10 1000 10 avgt 10 38,476 ± 2,146 ms/op
[info] StreamingBenchmark.testMonix 1000 10 1000 10 avgt 10 5,949 ± 1,681 ms/op
I would recommend against for because very often you want to mix other combinators that aren't flatMap, which isn't well supported (I also think they add a lot of noise in this case, but that's up to you).
Important: you need to handle errors in data, or you'll get stuck, you need to look at guarantee or at least attempt to make sure that the semaphore is incremented regardless. (also in that form, errors are swallowed, you need more machinery to make sure they are propagated down-stream).
Tbh I think that if you compare to Observable, you'll often see a perf hit, for example it might be because Observable uses impure queues under the hood. I'm kinda more interested if the performance you are getting is good for your use case, or it's giving you problems.
SystemFw
on 21 Aug 2018
I think we could still do a bit better. There is still a lot of machinery involved with every item. we will certainly get benefit if this will operate on chunks instead on individual elements.
pchlupacek
on 21 Aug 2018
@SystemFw do you also know by chynce, if monix Observable is somehow controlling the concurrency? b/c that may account for a lot of overhead, including the fact that we again synchronize on resulting value.
pchlupacek
on 21 Aug 2018
Good point, but I don't know about that
SystemFw
on 21 Aug 2018
@SystemFw it seems that Observable doesn't do actually anything. I couldn't see anything to control flow, and docs says that back-pressure must be somehow controlled. So I suspect if you put 10G tasks on it, it will OOM w/o any external control.
pchlupacek
on 21 Aug 2018
@pchlupacek Based on what you suggest with chunking I tried naive implementation with chunking the input data and following processing:
val itemsStream: Stream[Pure, Chunk[Int]] = Stream.chunkLimit$extension(Stream.range(1, items))(100)
Stream
.eval(Semaphore[IO](parallelism))
.flatMap { guard =>
Stream.eval(fs2.async.unboundedQueue[IO, Option[Array[Byte]]]).flatMap { queue =>
queue.dequeue.unNoneTerminate.concurrently {
itemsStream
.evalMap[IO, Chunk[Any]] { chunk =>
Traverse[Chunk].sequence {
chunk.map { _ =>
guard.decrement >>
Concurrent[IO].start {
IO {
DataGenerator.get(size)
}.flatMap(x => queue.enqueue1(x.some)).attempt >> guard.increment
}
}
}
}
.onFinalize(queue.enqueue1(None))
}
}
}
.map(_.length)
.reduce(_ + _)
.compile
.toList
.runAndWait
It brought some results (actually very similar as implementation in Observable converted to Stream[IO, _] but I guess you can still do better... :-)
[info] Benchmark (items) (parallelism) (size) (threads) Mode Cnt Score Error Units
[info] StreamingBenchmark.testFs2FromMonix 1000 10 1000 10 avgt 20 36,670 ± 0,402 ms/op
[info] StreamingBenchmark.testFs2FromMonix 1000 10 10000 10 avgt 20 36,618 ± 0,293 ms/op
[info] StreamingBenchmark.testFs2FromMonix 5000 10 1000 10 avgt 20 179,074 ± 2,176 ms/op
[info] StreamingBenchmark.testFs2FromMonix 5000 10 10000 10 avgt 20 176,024 ± 3,134 ms/op
[info] StreamingBenchmark.testFs2IO 1000 10 1000 10 avgt 20 346,745 ± 9,525 ms/op
[info] StreamingBenchmark.testFs2IO 1000 10 10000 10 avgt 20 349,313 ± 5,129 ms/op
[info] StreamingBenchmark.testFs2IO 5000 10 1000 10 avgt 20 1739,967 ± 25,137 ms/op
[info] StreamingBenchmark.testFs2IO 5000 10 10000 10 avgt 20 1719,916 ± 57,496 ms/op
[info] StreamingBenchmark.testFs2Workaround 1000 10 1000 10 avgt 20 55,211 ± 0,613 ms/op
[info] StreamingBenchmark.testFs2Workaround 1000 10 10000 10 avgt 20 54,650 ± 0,820 ms/op
[info] StreamingBenchmark.testFs2Workaround 5000 10 1000 10 avgt 20 275,377 ± 2,543 ms/op
[info] StreamingBenchmark.testFs2Workaround 5000 10 10000 10 avgt 20 279,504 ± 6,826 ms/op
[info] StreamingBenchmark.testFs2WorkaroundChunk100 1000 10 1000 10 avgt 20 37,047 ± 0,401 ms/op
[info] StreamingBenchmark.testFs2WorkaroundChunk100 1000 10 10000 10 avgt 20 36,411 ± 0,518 ms/op
[info] StreamingBenchmark.testFs2WorkaroundChunk100 5000 10 1000 10 avgt 20 184,140 ± 3,656 ms/op
[info] StreamingBenchmark.testFs2WorkaroundChunk100 5000 10 10000 10 avgt 20 184,702 ± 3,630 ms/op
[info] StreamingBenchmark.testFs2WorkaroundChunk20 1000 10 1000 10 avgt 20 39,115 ± 0,420 ms/op
[info] StreamingBenchmark.testFs2WorkaroundChunk20 1000 10 10000 10 avgt 20 38,245 ± 0,388 ms/op
[info] StreamingBenchmark.testFs2WorkaroundChunk20 5000 10 1000 10 avgt 20 199,825 ± 1,790 ms/op
[info] StreamingBenchmark.testFs2WorkaroundChunk20 5000 10 10000 10 avgt 20 198,666 ± 2,591 ms/op
[info] StreamingBenchmark.testMonix 1000 10 1000 10 avgt 20 4,078 ± 0,045 ms/op
[info] StreamingBenchmark.testMonix 1000 10 10000 10 avgt 20 4,344 ± 0,042 ms/op
[info] StreamingBenchmark.testMonix 5000 10 1000 10 avgt 20 20,466 ± 0,671 ms/op
[info] StreamingBenchmark.testMonix 5000 10 10000 10 avgt 20 19,842 ± 0,263 ms/op
It can be pushed down even more by using bigger Chunk up to a single one (which is possible here 'cause I know my input size but not in general code) but it saves only few more % (e.g. 34 vs. 31 ms on my PC, maybe less on server where I run benchmarks above). But again, it's just my naive implementation.
jendakol
on 21 Aug 2018
interesting. is your cpu btw fully spining when you do fs2 tests?
pchlupacek
on 21 Aug 2018
I think we still can improve this a bit, but only when we will lift some constrains, i.e. concurrency limits, and collecting the results. you may experiment with where the actual bottleneck is further, i.e. by removing semaphore or by removing the result quueue. I bet queue will be bottleneck. do you somehow collect results in monix code ?
pchlupacek
on 21 Aug 2018
No, it's far from "fully spinning", I have all cores cca 40% (after killing the benchmark it fell to 10%).
The monix code is almost identical to original FS2 one (see original post).
jendakol
on 21 Aug 2018
ok @jendakol that's what I have sort of suspected. The issue is that you are synchronising on the queue, and thats where is your contention point :-) Do you need to collect results ? If yes, how you doing this in Monix? Can you paste monix sample code we are comparing against?
pchlupacek
on 21 Aug 2018
I see, that's quite logical.
My use case is described in the original post as well as the Monix implementation.
I am not sure what are you asking about.
- The resulting library will not collect any results but provide stream of data to "user". Our own first usage counts with passing the stream to http4s (thus we need FS2 at the end).
- The benchmark does not collect the data directly too but tranforms them into some other form (_length_) which is the same from your POV I guess.
- If you mean "does your computation has any results" then the answer is yes, it's a query to DB (select)
jendakol
on 21 Aug 2018
@jendakol thanks for the info. I'll think if there is something we can do about that.
pchlupacek
on 21 Aug 2018
@jendakol I may see how the #1235 goes through. That in fact will give you I think what you really need, and your program will look so much simpler:
Stream
.range(1, items)
.distributeThrough(parallelism) {
_.evalMap { _ => IO { DataGenerator.get(size) } }.map(_.length) .reduce(_ + _)
}
.reduce(_ + _)
.compile
.toList
.unsafeRunSync
That will allow you to utilize fully your cores, and allow you very precisely control concurrency.
pchlupacek
on 22 Aug 2018
@pchlupacek It seems great but TBH this is what I expected mapAsyncUnordered to be :-) Could you please explain (maybe on examples) how these two methods differ?
However, this improvement will be much more important for the benchmark than for my real use-case. In "reality" there will be IO operation (DB query) taking few seconds instead of immediate result so the overhead may be less relevant. But still, this needs to be fixed/improved :-)
jendakol
on 23 Aug 2018
the new constructor allows you to sum chunks in parallel (10 threads) and then sum the results when each of them terminate. Whereas the solution with mapAsyncUnordered still basically synchronises you on one queue, as such you may not ever utilize full cpu potential.
Is perfectly ok if you operation is just A => F[B], still you will do evalMap and as you do it in parallel, you may fine-tune precisely cpu utilization.
As a side note, this also does much less context switching, so again, the CPU may be much better utilized.
pchlupacek
on 23 Aug 2018
OK but what is point of existence of mapAsyncUnordered then? In which scenario it would be better?
jendakol
on 23 Aug 2018
I think we may actually review that before 1.0 final :-)
pchlupacek
on 23 Aug 2018
@jendakol balance/broadcast are in #1235. I will still work on implementation of mapAsyncUnordered, but perhaps you may try with RC1, which will be released with these, if you saw performance improvement.
pchlupacek
on 24 Sep 2018
lets close this, @jendakol feel free to reopen, if you will see any perf. issues again :-)
pchlupacek
on 5 Oct 2018
Related issues
mpilquist
·
12Comments
gvolpe
·
10Comments
Taig
·
7Comments
mpilquist
·
12Comments
mpilquist
·
14Comments