Fs2: Stream#map destroys chunking (sometimes?)

[djspiewak]

Using map:

    val sliceStream =
      s
        .mapChunks({ chunk => println(s"before RValue> ${chunk.size}"); chunk.toSegment })
        .map(data => RValue.fromJValueRaw(JValue.fromData(data))
        .mapChunks({ chunk => println(s"after RValue> ${chunk.size}"); chunk.toSegment })

Using mapChunks:

    val sliceStream =
      s
        .mapChunks({ chunk => println(s"before RValue> ${chunk.size}"); chunk.toSegment })
        .mapChunks(chunk => chunk.map(data => RValue.fromJValueRaw(JValue.fromData(data))).toSegment)
        .mapChunks({ chunk => println(s"after RValue> ${chunk.size}"); chunk.toSegment })

There's some incidental complexity in this example because I haven't actually minimized it, but you get the idea. Here are some println excerpts:

before RValue> 93
after RValue> 1
after RValue> 1
after RValue> 1
after RValue> 1
after RValue> 1
after RValue> 1
after RValue> 1
after RValue> 1
after RValue> 1
after RValue> 1
after RValue> 1

And with mapChunks:

before RValue> 93
after RValue> 93
in yield> 94
before RValue> 94
after RValue> 94
^Cin yield> 93
before RValue> 93
after RValue> 93
in yield> 93
before RValue> 93
after RValue> 93
in yield> 93
before RValue> 93

Yeah, so… keep your hands off my chunking? ;-) I'm not sure I understand the point of a Stream#map which doesn't preserve chunk boundaries. This manifested as a 3x performance regression between fs2 0.10.0 and 1.0.0-M.

For reference, the surrounding commit state: https://github.com/wemrysi/quasar/commit/50ab260d37

[mpilquist]

I don't think chunking is getting destroyed here, but rather that Stream.chunk(c).map(f) is equivalent to Stream.segment(Segment.chunk(c).map(f)). When you ask for the chunks that make up Segment.chunk(c).map(f), the individual elements are emitted as singleton chunks. I don't think this is the root cause of the performance issue you are seeing, as in general, the fused segment will perform better than creating a bunch of intermediate chunks (though not always!). When asking for chunks, any fused segments must be forced and individual elements output. You can confirm this theory by observing segments instead of chunks.

So what changed between 0.10 and 1.0 that might explain what you are seeing? I suspect it's this change: https://github.com/functional-streams-for-scala/fs2/commit/ff0364eedf913327a3f7e22ee996413f535a1b0b

The former version literally implemented Stream#map by pushing a map operation in to each segment. This involved converting to a pull though, unconsing a segment, mapping it, and outputting it. The optimization in this commit was intended to avoid the cost of unconsing by instead mapping the pull output directly: https://github.com/functional-streams-for-scala/fs2/blob/series/1.0/core/shared/src/main/scala/fs2/Pull.scala#L184-L193

So basically, I'm thinking that Stream#map being implemented in terms of FreeC#translate is slower than the original implementation, which used uncons. I'll benchmark this to confirm but this should be an easy fix if so.

[mpilquist]

After benchmarking, I don't think that's the cause either. JMH benchmarks don't show any performance difference between mapSegments(_.map(f)) and pull.echo.mapOutput(f).streamNoScope -- maybe a very slight edge to the former but could be measurement noise.

So then I tried running the map benchmarks against 0.10 and they appear to be equal in speed to 1.0.

Benchmark:

diff --git a/benchmark/src/main/scala/fs2/benchmark/StreamBenchmark.scala b/benchmark/src/main/scala/fs2/benchmark/StreamBenchmark.scala
index c7f5178b..8dc42c98 100644
--- a/benchmark/src/main/scala/fs2/benchmark/StreamBenchmark.scala
+++ b/benchmark/src/main/scala/fs2/benchmark/StreamBenchmark.scala
@@ -89,4 +89,20 @@ class StreamBenchmark {
   @Benchmark @BenchmarkMode(Array(Mode.AverageTime)) @OutputTimeUnit(TimeUnit.NANOSECONDS)
   def emitsThenFlatMap(N: Int): Vector[Int] =
     Stream.emits(0 until N).flatMap(Stream(_)).toVector
+
+  @GenerateN(1000000)
+  @Benchmark @BenchmarkMode(Array(Mode.AverageTime)) @OutputTimeUnit(TimeUnit.NANOSECONDS)
+  def chunkyMap(N: Int): Vector[Int] =
+    Stream.emits(0 until N).segmentN(10000).flatMap(Stream.segment(_)).map(_ + 1).toVector
+
+  @GenerateN(1000000)
+  @Benchmark @BenchmarkMode(Array(Mode.AverageTime)) @OutputTimeUnit(TimeUnit.NANOSECONDS)
+  def chunkyMapSegments(N: Int): Vector[Int] =
+    Stream
+      .emits(0 until N)
+      .segmentN(10000)
+      .flatMap(Stream.segment(_))
+      .mapSegments(_.map(_ + 1))
+      .toVector
+
 }
diff --git a/core/jvm/src/test/scala/fs2/StreamPerformanceSpec.scala b/core/jvm/src/test/scala/fs2/StreamPerformanceSpec.scala
index a758682f..b8400833 100644
--- a/core/jvm/src/test/scala/fs2/StreamPerformanceSpec.scala
+++ b/core/jvm/src/test/scala/fs2/StreamPerformanceSpec.scala
@@ -13,7 +13,7 @@ class StreamPerformanceSpec extends Fs2Spec {

     case object FailWhale extends RuntimeException("the system... is down")

-    val Ns = List(2, 3, 100, 200, 400, 800, 1600, 3200, 6400, 12800, 25600, 51200, 102400)
+    val Ns = Stream.unfold(2)(n => if (n < (2 << 23)) Some((n, n * 2)) else None).toList

     "left-associated ++" - {
       Ns.foreach { N =>
@@ -186,5 +186,15 @@ class StreamPerformanceSpec extends Fs2Spec {
         }
       }
     }
+
+    "chunky map" - {
+      Ns.foreach { N =>
+        N.toString in {
+          runLog(emits(Vector.range(0, N)).map(_ + 1).map(_ + 2)) shouldBe Vector
+            .range(0, N)
+            .map(_ + 3)
+        }
+      }
+    }
   }
 }

[mpilquist]

Spot checked a few more things but don't see anything different between 0.10.5 and 1.0.0-M1 that would explain such a performance difference. Forcing a fused segment by asking for chunks behaves the same in both versions. Any other data that might help narrow down?

[djspiewak]

@mpilquist I should clarify: the performance loss was in the overall application stemming from the data structures which arose from these two lines, which are subsequently applied to the same stream:

        .segments
        .map(c => Slice.fromRValues(unfold(c)(_.force.uncons1.toOption)))

Slice is our own internal chunking mechanism, which represents JSON data in a columnar format. It's basically the core optimization within the quasar database engine. On 0.10, this code resulted in slices which contained about 90-100 elements. (that's suboptimal – we usually want around 10000 – but the reason for the 90-100 had nothing to do with fs2) On 1.0, this code results in slices which contain exactly 1 element each. This in turn resulted in effectively deoptimizing the rest of the process by, effectively, about two orders of magnitude. The fact that it was only a 3x slowdown in the final result was, we suspect, caused by the fact that fs2 and cats-effect both improved their own performance significantly between their respective 0.10 and 1.0 releases, and so the effective slowdown was less than the slice boundary should have implied.

Anyway, I don't know if we're just observing the chunking incorrectly, but I'm not actually aware of the correct way to do this. We basically need to preserve the origin stream chunking through map operations all the way through to where we construct our own chunk representation, which will use the original chunking as an upper-bound. If that original chunking is lost, then the upper-bound becomes 1 and the entire system deoptimizes.

Hopefully that lends a bit more context. It's not really that I think fs2 itself is slowing down here, it's just that fs2 changed a visible chunk-sizing semantic in a way that I still don't quite know how to work around, which in turn had an effect on our own internal optimization tuning.

[mpilquist]

Ah okay. I just looked through the diff of series/0.10 to series/1.0 and don't see anything that would change chunking behavior.

You have a few options -- you could use map any number of times and then eventually do .mapSegments(s => convertSegmentToSlice(s)) where convertSegmentToSlice is free to call s.force.toChunk or s.force.toArray or build a fresh slice, whatever you want.

Another option is to avoid the lazy/fused map operation entirely and use s.mapChunks(_.map(f).toSegment). This evaluates strictly, allocating a new chunk for each input chunk. I could see adding this to fs2 as mapStrict(f) or something, as sometimes the constant factors of Segment outweigh the cost of intermediate copies, especially when dealing with byte arrays.

[djspiewak]

Hmm, mapSegments with force is is pretty close to what we were trying to do. The only difference is that we only called .force, we didn't call .force.toChunk.

mapStrict does sound like an interesting option, though in general I would want to address the issue of chunk boundary visibility more at the endpoint (e.g. mapSegments) rather than having to worry about it along the way, since that way we don't have some spooky action constraint propagating back up into wherever we're getting the streams from.

[djspiewak]

I guess one thing that sort of defeated my expectations a bit is I expected that Chunk would be eager, and so mapChunks would thus do whatever forcing was required to achieve that.

[mpilquist]

In an early milestone of 0.10, Chunk was a subtype of Segment -- keeping track of which operations were eager and which were lazy and dealing with name conflicts when there was a need for both variants led to so much confusion. So we removed the subtyping and introduced the wrapper. Now all operations on Chunk are eager while all operations on Segment are lazy. Any eager operation on Segment is behind the .force syntax to make it clear where evaluation will occur.

I think in your case, the confusion resulted from something upstream of mapChunks performing some lazy segment manipulation (e.g. simply calling Stream#map). I should have mentioned another option for dealing with this type of thing, though it's perhaps not a satisfactory solution -- using segmentN(idealChunkSize).flatMap(Stream.chunk(_.force.toChunk)) as a "rechunk" combinator -- this is appropriate for different use cases than forcing lazy segments to eager chunks. It's really more for cases where the chunkiness of the input varies and hence, normalizing the chunkiness results in more deterministic downstream performance.

[wemrysi]

@djspiewak @mpilquist to clarify, we went from 0.9.6 to 1.0.0-M1, not from 0.10.x, not sure if that makes a difference.

[edmundnoble]

Moving from 1.0.0-M1 to 0.10.5 has seemingly completely fixed our performance problems.

[mpilquist]

Hm very interesting! Do you see any observable differences between 0.10.5 chunk structure vs 1.0.0-M1 chunk structure? I can see how going from 0.9 to 1.0 would lead to surprises in chunk structure due to the introduction of segments, but 1.0 to 0.10.5 should be identical there. I'm wondering if the chunk stuff is a red herring and the root cause is something else entirely.

[djspiewak]

Edit: Ah, Edmund and Emrys already posted some of this. 😃 Well, more voices to chime in.

Not sure exactly what we’re seeing yet, but it’s definitely related to chunking and laziness. We’re still investigating, but things we learned further last night:

• It’s not just the Slice size stuff that is causing the performance hit. It appears to be operations on fs2 itself across the code base. We went through and converted all of our filter and map and other lazy operations to explicitly use mapChunks and run eagerly and we gained back all (and in some cases more) of the performance loss (6 minutes vs 18 on a test suite run).
• Downgrading to 0.10.5 and changing nothing also gains back the same performance (we also downgraded cats-effect in the same experiment, but I don’t think that’s relevant)
• Catenable is the big culprit in profiling. The suite allocates over 250 GiB of Append instances and over 150 GiB of Singleton instances. The pause time on the GC alone is over 5 minutes cumulatively.
• Catenable is also the big offender in terms of method profiling. It is responsible for almost half of the total method calls in the entire 18 minute run all by itself! In other words, it’s being hit as hard as the entire rest of the application combined. That seems wrong.

Taking a step back, philosophically, I’m wondering why it makes sense to do things lazily within chunk boundaries? I mean, you have an eager chunk. Sure, if your application has huge chunks and very long chains of lazy operations you can save some large allocations at the cost of more small ones, but that only seems like a very specialized win. Why shouldn’t eager chunks remain eager, with the overhead of laziness confined at the boundaries?

[mpilquist]

Downgrading to 0.10.5 and changing nothing also gains back the same performance

This is the part that confuses me. There should be no difference whatsoever with respect to chunking and laziness between 0.10.5 and 1.0.0-M1. I'll look through the diffs again to see if I missed something.

Catenable is the big culprit in profiling. The suite allocates over 250 GiB of Append instances and over 150 GiB of Singleton instances. The pause time on the GC alone is over 5 minutes cumulatively.

This is definitely concerning. I wonder if there's a lot of "uncons segment, take head chunk, push remainder back" type access patterns occurring, which will result in Segment.Catenated nodes.

Taking a step back, philosophically, I’m wondering why it makes sense to do things lazily within chunk boundaries?

I don't think that question is well-typed. :) Within chunk boundaries, everything is always strict. The only way to treat a chunk lazily is by lifting it to segment. As of 0.10.0, a stream emits segments (not chunks), and many common operations operate on segments using the lazy segment operations. I think what you are asking is whether it's better to implement things like Stream#map in terms of segment operations or chunk operations. I don't know the answer to that in general, though before this issue, my belief (informed by benchmarks and experiments) was that some workloads benefit from eager chunk based implementations but other workloads benefit from lazy segment implementations, and that the set of workloads in the segment category was much bigger. Maybe that's wrong though?

[SystemFw]

Downgrading to 0.10.5 and changing nothing also gains back the same performance

This is the part that confuses me.

I wanted to comment this as well :)

Just went through the diff and couldn't see anything that should make any difference, apart from that map commit

[edmundnoble]

So it turns out that downgrading did not give us the same performance on its own.

We still had to replace some map(f) calls with mapChunks(_.map(f)) for a giant performance increase.

[djspiewak]

@mpilquist I'll reply to the lazy-default vs eager-default point in a bit, but more experimental data…

Is this intended?

val s = Stream(Chunk(1, 2, 3), Chunk(4, 5, 6)).flatMap(Stream.chunk).covary[IO]
s.chunks.map({ c => println(c.size); c }).compile.drain.unsafeRunSync  // 3, 3
s.map(_ + 1).chunks.map({ c => println(c.size); c }).compile.drain.unsafeRunSync // 1, 1, 1, 1, 1, 1

So what we're getting at here is the fact that we would expect chunks (and mapChunks) would handle the forcing of the inner segments, collapsing the laziness and giving us eager chunks that preserve the original boundaries. Instead, what appears to be happening is we end up getting singleton chunks when the segments are unforced. Just allocating the catenables (in unconsChunks, I would assume) to carry along those chunks is extremely expensive, but also this accentuates load in other places.

Is our expectation surrounding chunks wrong? And if so, why? Is this a bug? Can it, uh… be a bug please? :-D

[djspiewak]

I think what you are asking is whether it's better to implement things like Stream#map in terms of segment operations or chunk operations. I don't know the answer to that in general, though before this issue, my belief (informed by benchmarks and experiments) was that some workloads benefit from eager chunk based implementations but other workloads benefit from lazy segment implementations, and that the set of workloads in the segment category was much bigger. Maybe that's wrong though?

At least in my experience (in other systems of this style), eager within chunk boundaries is the right default and it wins almost every time. The reason for this, at least in my experience, is that you can always reassociate chunks to get laziness if you need it, since you're always going to be lazy outside chunk boundaries. This also gives a lot more control over the evaluation semantics and allows much tighter hot-path functions, since the hot path is (by definition) "everything within a chunk boundary".

I think this probably comes down to benchmarking assumptions. What sort of workloads are assumed to be "normal"? That choice fully determines whether lazy-default or eager-default ends up being optimal.

[mpilquist]

val s = Stream(Chunk(1, 2, 3), Chunk(4, 5, 6)).flatMap(Stream.chunk).covary[IO]

Internally, this is going to look something like this (excuse the rather informal description of the free program that's constructed): Bind(Output(Segment.chunk(Chunk(1, 2, 3))), _ => Output(Segment.chunk(Chunk(4, 5, 6))))

s.chunks.map({ c => println(c.size); c }).compile.drain.unsafeRunSync  // 3, 3

This says, "give me each chunk and print its size" which turns in to "for each segment, ask for each chunk and print its size". So the interpreter hits the first segment, which yields a single Chunk(1, 2, 3) then terminates. The interpreter then hits second segment, which yields a single Chunk(4, 5, 6).

s.map(_ + 1).chunks.map({ c => println(c.size); c }).compile.drain.unsafeRunSync // 1, 1, 1, 1, 1, 1

s.map(_ + 1) transforms the original stream in to this: Bind(Output(Segment.chunk(Chunk(1, 2, 3)).map(f)), _ => Output(Segment.chunk(Chunk(4, 5, 6)).map(f))). When interpreter runs, it hits the first segment and says "give me each chunk". This results in a segment force via Segment#unconsChunk which asks for the first chunk from Segment.chunk(Chunk(1, 2, 3)).map(f). The implementation of Segment#map is given Chunk(1, 2, 3) and then basically executes emit(f(1)); emit(f(2)); emit(f(3)). The first chunk (Chunk.singleton(f(1))) is emitted and the remainder is returned as Segment.catenated(Chunk.singleton(f(2)), Chunk.singleton(f(3)), remainderFromUnconsChunk). In this case, remainderFromUnconsChunk is empty.

Note this behavior explains your Catenable observations too.

Is our expectation surrounding chunks wrong? And if so, why? Is this a bug? Can it, uh… be a bug please? :-D

I'm not necessarily tied to the current implementation. The fact that this caused such a subtle performance issue bothers me and we should definitely consider changing the default behavior. With that said, I'm not sure how wide ranging the solution should be.

A simple fix is making Stream#map eager and introducing Stream#mapLazily or something, but that's special casing the fix just for map. Throw a filter in the mix or many other transforms, and you're back to problematic behavior.

Another idea is making Segment#unconsChunk do some re-chunking -- e.g., when a single step of the segment interpreter yields N chunks, just copy them all to a new chunk and emit that instead of pushing back all but the first. The more I think about that option, the more I like it.

[djspiewak]

...

Note this behavior explains your Catenable observations too.

Ok cool, this is exactly what I thought was happening, looking at code and profiles. So I think I'm finally on the same page. :-)

A simple fix is making Stream#map eager and introducing Stream#mapLazily or something, but that's special casing the fix just for map. Throw a filter in the mix or many other transforms, and you're back to problematic behavior.

Yeah I hate that idea. :-D For the same reason you do.

Another idea is making Segment#unconsChunk do some re-chunking -- e.g., when a single step of the segment interpreter yields N chunks, just copy them all to a new chunk and emit that instead of pushing back all but the first. The more I think about that option, the more I like it.

I like this, too. This is pretty similar to something @edmundnoble and I were discussing as a potential solution, which is basically changing chunks to explicitly force segments prior to unconsing. One concern I would have though is that it might still impose the overhead of the catenated single segments as a transient state, which is slow.

[mpilquist]

Okay I'll put a PR together. Any idea on how best to rechunk? That is, how best can we implement this:

def rechunk[O](chunks: List[Chunk[O]]): Chunk[O]

Whatever we do must avoid boxing for all primitive types. We also don't have a ClassTag[O] here, which would make this trivial, though maybe there's a way to sneak that in somehow.

[djspiewak]

I'm still concerned about the fact that we're creating this List[Chunk[O]] in the middle. Like, I would ideally want a pipeline something like this:

Chunk => Segment(map) => Segment(filter) => Segment(...) => Chunk

Basically without the sequence of chunks in the middle. I'm not sure if that's compatible with the Segment architecture though.

IIRC, back in 0.9, Chunk had a concatAll function which preserved boxing and was extensible. That seems to be gone now. I don't believe it was dependent on the tricks with looking at getClass on Function1, but my memory is foggy.

[mpilquist]

Eh, it's kind of possible.

I think the fix is to rewrite Segmetn#unconsChunk -- start by replacing the current definition entirely with the implementation of unconsChunks and then instead of accumulating chunks in a Catenable[Chunk[O]], we accumulate in a list. Once a step has completed, we reduce the list to a single chunk using an optimized rechunk function. We could avoid the intermediate list if we could support an efficient unboxed accumulator that had fast snoc and append. Catenable doesn't fit the bill here b/c it will box everything. Hence it seems like a list or mutable buffer of chunks would be more performant.

You are right that Chunk.concat didn't use ClassTag[O] and instead just resorted to runtime type checks. https://github.com/functional-streams-for-scala/fs2/blob/series/0.9/core/shared/src/main/scala/fs2/Chunk.scala#L279-L305

I could restore that if you think it's the best path forward... Deleting that stuff was very satisfying though.

[djspiewak]

@mpilquist What if we special-case the Segment operations (e.g. map) which correspond to Chunk operations when evaluated on Segment.chunk values? What I mean by this is, rather than always going through the fully-generic Segment#map implementation, check to see if we're mapping on an eager chunk, and build a different Segment which will delegate down to Chunk#map when evaluated. So still lazy, and still enabling fusion, but ensuring that, ultimately, Chunk itself handles the evaluation as much as possible. Any operation which isn't directly available on Chunk would of course yield a Segment in the current style which has to do its own implementation, but some of these more basic operations would be able to avoid the intermediate sequence.

I think the fix is to rewrite Segmetn#unconsChunk -- start by replacing the current definition entirely with the implementation of unconsChunks and then instead of accumulating chunks in a Catenable[Chunk[O]], we accumulate in a list. Once a step has completed, we reduce the list to a single chunk using an optimized rechunk function. We could avoid the intermediate list if we could support an efficient unboxed accumulator that had fast snoc and append. Catenable doesn't fit the bill here b/c it will box everything. Hence it seems like a list or mutable buffer of chunks would be more performant.

Yeah, I think if we can't defer operations for ultimate delegating to the original Chunk (when relevant), then this is probably the best we can do. Obviously it's annoying, but Segment just enables more than Chunk (so far as I can tell), so sometimes it's unavoidable.

I could restore that if you think it's the best path forward... Deleting that stuff was very satisfying though.

😁 I think if we have to implement List[Chunk[O]] => Chunk[O], then it's actually the only path forward (I remember when I wrote that stuff, I tried really hard to avoid the inheritance-runtimechecks-bs stuff and couldn't).

[mpilquist]

That makes me a bit queasy as then you don't know what operations on segment are eager or lazy. What I like about pushing it all to unconsChunk is you still get arbitrary fusion in expressions like Stream.chunk(c).map(f).filter(g).map(h).unconsChunk

[djspiewak]

That makes me a bit queasy as then you don't know what operations on segment are eager or lazy.

Things would still be lazy, it's just that operations which can be represented in terms of Chunk operations (when ultimately forced) would be delegated directly to Chunk, rather than passed through the interpreter machinery. So it would look something like this:

Chunk => Segment.chunk => Segment.chunkMap => Segment.chunkFilter => Segment.chunkMap => Chunk

Or something like that. You still get arbitrary fusion and ubiquitous laziness, you just don't lose the specialization and the compactness of Chunk. And you also don't build the intermediate catenables or lists.

This actually raises a more fundamental philosophical question: why have Segment at all? I really don't understand. I can see that it would improve workflows which have some of the following properties:

• Extremely cheap chunk boundaries (e.g. IO.pure)
• Large non-primitive chunks
• Long fusion-compatible pipelines (e.g. maps, filters`, etc)

Under those circumstances, certainly laziness within the chunk boundary is great, but you can achieve almost the same benefits just by rechunking to 1 and relying on the laziness of Stream itself. For any other workflows (or even softer variants of this workflow), this sort of thing can be really bad. For example, the conventional situation for me is something like:

• Extremely expensive chunk boundaries
• Large primitive chunks
• Very very short fusion pipelines

The last one is pretty critical. I tend to not have long pipelines of fusable things, and so the "optimization" actually just results in considerably deoptimizing and generating internal structures which never would have otherwise existed.

I guess just from a design standpoint, I don't understand not using Chunk directly just in general (I mean Chunk back when it had functions on it). Chunk is eager, has visibility into its underlying primitives, and is capable of specializing its operations (though not as well as it could pre-2.12). Segment is never going to match that, and all that Segment offers (as far as I can tell) is fusion, which isn't a benefit for any use-case I've found in the wild. I feel like I'm really missing something in the design justification.

[mpilquist]

So it would look something like this:

I think I see where you are headed here. I'll poke at it some and see how it goes, unless you want to?

This actually raises a more fundamental philosophical question: why have Segment at all?

Remember in 0.9, we had adhoc special logic for preserving chunkiness -- basically everything had to go through mapChunks in order to preserve chunkiness. Segment solves all that as long as you don't try to observe chunk structure (though use of s.pull.unconsChunk). In the end, 0.10 ended up being much faster than 0.9 though most of those improvements are likely due to the FreeC encoding and associated interpreter and not necessarily replacing Chunk with Segment.

all that Segment offers (as far as I can tell) is fusion, which isn't a benefit for any use-case I've found in the wild

Fusion and O(eC) append. I think segment fusion ends up helping more than you expect given how much of the built-in combinators are implemented directly in terms of segment operations. Note segment has some non-trivial operations like zip and flatMap too. I'd really love a set of comprehensive benchmarks that put these types of things to the test. The existing segment and stream benchmarks don't cover this kind of stuff.

[mpilquist]

I think I see where you are headed here. I'll poke at it some and see how it goes, unless you want to?

Eh, I take it back. It seems like this heads in the direction of the fusion technique used by stdlib views, unless I'm misunderstanding. E.g., I don't know how'd you handle fusing map(f).filter(g).map(h) such that only a single chunk is instantiated. More generally, this feels like bolting on an alternate design for fusion to the one already supported by segment.

I'll try to implement the unconsChunk thing we talked about. If you want to attempt specializing segment operations for chunks, please feel free!

[mpilquist]

@djspiewak I'm really interested in how your app performs with #1161 included. In a microbenchmark, I saw an over 150x increase.

[djspiewak]

@mpilquist I'll locally publish and take a look asap! It definitely seems like a very good step forward, despite my concerns on the PR.

As an aside, @wemrysi and I had a long discussion about what, in a vacuum, makes sense here. Basically trying to nail down what we think we want from fs2, what would be ideal for our use-case, and then more generally what direction we want to argue is best for fs2 divorced from the specifics of our application. We came up with a few thoughts.

First off, any solution here is obviously going to be good for some cases and really terrible for others. fs2 is effectively defining a semantic for an operational runtime based on a set of assumptions regarding what sorts of applications are going to be put on top of that runtime. We can't really hope to define anything "optimal", since that answer changes depending on how you define the question, all we can define is a set of reasonable defaults.

What our broad conclusion came down to is that we still don't understand why Segment is a thing. 😃 Obviously it's really inconvenient to have ad hoc chunking logic strewn throughout fs2, but the tradeoff is that you get in-memory chunks which are guaranteed to be carried through the streaming process without overhead. Even with the new unconsChunk, that simply is not and cannot be the case with Segment. Even if I implemented the SegmentView concept (which, as you pointed out, does fail to fuse map . filter . map), it would still not necessarily guarantee that nice in-memory chunks are carried through in an efficient fashion.

What I mean by "efficient" here is "avoiding unnecessary per-element allocations". Allocations including Catenable.Singleton and even List.Cons. Obviously that's something which is very near-and-dear to quasar, but I'm going to argue that this is fundamental to how fs2 will be used by everyone. Take the following snippet (assuming jawn and http4s pseudocode):

val client = HttpClient[IO]()

val values: Stream[IO, Json] =
  readAllAsync[IO].mapChunks(c => AsyncParser.absorb(c.toByteBuffer)).map(_ \ "foo")

client.post(url, values)

Even without getting into primitive chunks and stuff, you can see why this is going to cause a problem. There's no way for me to avoid the allocation of the per-element Catenable inside of Segment due to the map. Now, with the new unconsChunk, the chunk boundaries themselves will at least be reconstituted within http4s, so it won't end up writing one Json value at a time on the socket, but it's still a lot of extra allocation overhead. And all the while, I'm not benefiting at all from fusion, since my pipeline is one operation deep.

Basically, I'm positing that even use-cases which benefit from fusion are going to benefit more from avoiding all of those intermediate per-element allocations and accesses. And the sorts of use-cases which benefit significantly from fusion are quite unusual, at least in our experience. Our experience is certainly limited, so I don't want to oversell my point! I'm sure there are plenty of applications which don't mind the allocations and benefit tremendously from the fusion. Again, there's no such thing as "optimal" here.

So in the end, we concluded that we just don't understand Segment and why it's better than having Chunk (from the perspective of the general end-user). We understand why it's a lot more straightforward for fs2's implementation, but we would personally vastly prefer the way chunking was handled in 0.9. Maybe we're missing some benefits of Segment beyond fusion, localization of chunking logic, and append within chunk boundaries?

We can obviously do explicit chunking just by making the element type some sequence value, but everything in fs2's design (e.g. readAll) pushes users to rely on the chunking that is built into the library, so that just seems really weird. Basically, fs2 claims to have efficient chunking, and the Chunk type itself certainly looks very eager and efficient, but the design of the library makes it impossible to be efficient and avoid allocations within chunk boundaries, and that problem is fundamental to Segment itself, not just the implementation of unconsChunk.

[mpilquist]

I generally agree with your summary here, though I still suspect Segment pays off in more use cases than you expect. Maybe not though, but more in that in a moment. Here's what I'm thinking:

1) Look for places in the current (0.10/1.0 segment based) architecture where accidental loss of chunking can occur and address such cases. Such remediations are likely very specific to the use case so it's hard to say more from a general strategy perspective, but the general principle is that it should be hard to get terrible performance from reasonable looking code.

2) Develop a comprehensive benchmark suite covering things like chunk & segment types/sizes, various transformations, use of primitive backed chunks, etc. This suite should also cover streams used for control flow, where there tends to be much small chunks.

3) Drastic architecture changes are informed by the results from (2). If the benchmark results show that we'd be better off without Segment, I'm happy to remove it, as it's hard to work with and confusing when learning the library. Ideally, such a benchmark suite would have guided it's inclusion in the library instead of only its removal but that didn't happen. There were certainly microbenchmarks that led to Segment's development though perhaps they aren't relevant to the wider performance of Stream.

I would really love some volunteers / PRs for the benchmark suite, including the ability to run the benchmarks against different versions of FS2 (perhaps with source incompatibilities).

Another random thought: could the Segment interpreter but changed to somehow provide a "chunk builder" that is given a size hint? Perhaps this would allow something like unconsChunk to allocate a single array sized to match the size hint and write values directly to that array from emit/emits? By no means does this solve everything but it would be an incremental improvement over the current unconsChunk in that it avoids the Catenable & List allocations. Lack of specialization on mutable.Buffer makes the implementation a pain of course.

[djspiewak]

1. Look for places in the current (0.10/1.0 segment based) architecture where accidental loss of chunking can occur and address such cases. Such remediations are likely very specific to the use case so it's hard to say more from a general strategy perspective, but the general principle is that it should be hard to get terrible performance from reasonable looking code.

From what I understand, unconsChunk should have hit nearly all of them, but obviously there are always corner cases. Very much agreed on the general principle.

I would really love some volunteers / PRs for the benchmark suite, including the ability to run the benchmarks against different versions of FS2 (perhaps with source incompatibilities).

This was something that I (and @rossabaker) started like, two years ago and then put down again. I definitely want to help push this forward if I can. No promises right now but I'm looking into things. I definitely don't want to be "that guy" who swoops in, demands "change all these things", and then doesn't contribute towards achieving that goal. :-)

Another random thought: could the Segment interpreter but changed to somehow provide a "chunk builder" that is given a size hint? Perhaps this would allow something like unconsChunk to allocate a single array sized to match the size hint and write values directly to that array from emit/emits? By no means does this solve everything but it would be an incremental improvement over the current unconsChunk in that it avoids the Catenable & List allocations. Lack of specialization on mutable.Buffer makes the implementation a pain of course.

I love this idea, but doesn't it still involve building a Catenable to represent the operations themselves on a per-element basis? Or is the Catenable built within Segment then solely proportional to the size of the operations pipeline (not the chunk size)? If it's the latter, then this sounds really perfect.

[mpilquist]

No Catenable at all - I can do a proof of concept that uses a mutable buffer and no size hint. Adding size hinting and non-boxed arrays then would just be labor.

On Jun 16, 2018, at 8:05 PM, Daniel Spiewak [e-mail] wrote:

Another random thought: could the Segment interpreter but changed to somehow provide a "chunk builder" that is given a size hint? Perhaps this would allow something like unconsChunk to allocate a single array sized to match the size hint and write values directly to that array from emit/emits? By no means does this solve everything but it would be an incremental improvement over the current unconsChunk in that it avoids the Catenable & List allocations. Lack of specialization on mutable.Buffer makes the implementation a pain of course.

I love this idea, but doesn't it still involve building a Catenable to represent the operations themselves on a per-element basis? Or is the Catenable built within Segment then solely proportional to the size of the operations pipeline (not the chunk size)? If it's the latter, then this sounds really perfect.

—
You are receiving this because you were mentioned.
Reply to this email directly, view it on GitHub, or mute the thread.

[djspiewak]

No Catenable at all - I can do a proof of concept that uses a mutable buffer and no size hint. Adding size hinting and non-boxed arrays then would just be labor.

Noice. Very very very nice. We would be quite interested in helping out as much as feasible ("feasible" still tbd, but expect PRs anyway). Thanks so much for pushing on this, @mpilquist. :-)

[djspiewak]

@mpilquist Oh, as an aside, the use of -SNAPSHOT versions (as opposed to stable hash snapshots) makes it harder than it should be to try out these sorts of things. 😢 /endrant

[djspiewak]

@mpilquist Looks like the changes make a difference, but it's still quite observably worse than manually maintaining the chunking. My guess is this is due to the List allocations, but I haven't profiled yet.

Branch is https://github.com/djspiewak/quasar/tree/wip/mimir-query-evaluation if you want to check it out yourself. Create a file it/testing.conf with the following contents:

lwc_local = "/tmp/lwc_local/"

Then run sbt 'it/testOnly *Sql2QueryRegressionSpec' to see it in action.

• With manual chunk management (https://github.com/djspiewak/quasar/commit/80a6eee50aefae7c2e830e1c1ad33acddfcba909), the test ran in 717s
• With the 1.0.0-M1 chunk managing behavior (https://github.com/djspiewak/quasar/commit/d362ff1270b07f72b4db43fa9675d0b79fe75fa4), tests ran in 1276s with three timeouts
• With the new 1.0.0-SNAPSHOT chunk management (https://github.com/djspiewak/quasar/commit/4a9c6d45e856abb3b130c171a8e1e2da6c484abb), tests ran in 827s

So definitely a huge improvement, but still worse than manual chunk management by well over the margin for error on these tests (laptop, on battery power, browser tabs open to reddit, etc etc).

Edit: Just as an experiment, I tried 1.0.0-SNAPSHOT but with the manual chunk management and it made almost no difference.

I ran JFR on the snapshot-without-manual-management version (the last bullet point). The results are here. Some thoughts:

• As we hoped, Catenable is banished from bonkers allocation-land. Object[] has the top spot now, which sounds about right given the rest of the application. Unfortunately, most of those Object[] instances are coming out of Catenable.uncons, so we're not entirely out of the woods. Specifically, Segment.loopOnChunks.
• The largest chunk of fs2-related allocations is FreeC.Bind, at 30.2 GiB
• Despite fs2 no longer being the highest allocation offender, it does appear to still be the highest method time offender. fs2 and fs2.internal account for around 30-40% of the total method profile times. Segment.stage and FreeC.go are the worst offenders

[wemrysi]

Here's another datapoint that includes 0.9.6 performance on the same machine. Even with the improvements, we're still almost twice as slow for this particular workload:

• djspeiwak/quasar@d15d5cd (0.9.6): 339s
• djspiewak/quasar@80a6eee (1.0.0-M1, manual chunking): 709s
• djspiewak/quasar@d362ff1 (1.0.0-M1 chunk management): 1136s
• djspiewak/quasar@4a9c6d4 (1.0.0-SNAPSHOT chunk management): 654s

[mpilquist]

Unfortunately, most of those Object[] instances are coming out of Catenable.uncons, so we're not entirely out of the woods. Specifically, Segment.loopOnChunks.

This is interesting -- loopOnChunks is inside the private[fs2] foldRightLazy method, which is only called from Stream#flatMap.

I think foldRightLazy would be pretty easy to optimize by giving it its own interpreter instead of piggybacking on unconsChunks. I'll take a look at doing so and if it works, I'll open a speculative PR.

[djspiewak]

@mpilquist Just a quick reply to let you know I haven't forgotten about this (and your PoC PR that needs contributions), I've just been busy. :-)

[mpilquist]

Closing due to #1175 getting merged. I'd love to see updated perf numbers against 1.0.0-SNAPSHOT.