Add MessagePack item serializer and validator

[jarmuszz]

This pull request brings serialization and validation to the msgpack.low module.

Serialization is provided for the outside use through tobinary: Pipe[F, MsgpackItem, Byte] and toNonValidatedBinary: Pipe[F, MsgpackItem, Byte] functions. Akin to how the cbor.low serialization is constructed, serialization without validation can potentialy produce malformed data.

Validation API is also exposed through validated: Pipe[F, MsgpackItem, MsgpackItem].

[jarmuszz]

The initial concept of having two serializers was dropped as they ran at a nearly identical time 😃.

These are the current benchmarks on i7-8550U:

Benchmark                                       Mode  Cnt     Score    Error  Units
MsgPackItemSerializerBenchmarks.serialize       avgt   10  3337.518 ± 39.333  us/op
MsgPackItemSerializerBenchmarks.withValidation  avgt   10  5107.575 ± 65.148  us/op

Validation seems to slow things down a bit and I think that maybe it is possible to make it a little faster.

[satabin]

I like the approach, however, this creates a new Out instance for each incoming item. If you run the benchmark with allocation and CPU flame graphs, does it appear to be a bottleneck?

[jarmuszz]

It doesn't seem like a significant bottleneck to me.

Here's a screenshot from VisualVM memory profiler:

I cannot find any reference to Out's methods in CPU samples so either I'm doing something wrong or these methods get optimized out :)

[satabin]

I was thinking about runnin the benchmark with async-profiler outputting flamegraph. Something like:

$ sbt
> project benchmarksJVM
> Jmh/run fs2.data.benchmarks.MsgPackItemParserBenchmarks -prof "async:libPath=/path/to/libasyncProfiler.so;output=flamegraph;event=cpu;alluser=true"

[jarmuszz]

It takes around 10% of CPU and 17% of allocation graphs.

Making theOut class an implicit around Chunk[Byte] lowers the results into 7% and 15% respectively. I'm not sure how could we optimize this more.

[satabin]

Time and memory seem to be dominated by ByteVector creation at every step. I need to review again these creations, and why they occur. But I could live with merging the current state and improving later on.

[satabin]

Thanks a ton for this new contribution. I had a first look at it and left a few comments. But it looks great already. 👏

[ybasket]

Looks good overall 👏 , left only some smaller comments.

[ybasket]

If I'm not mistaken, this destroys all chunking, you'll work with one gigantic Chunk – not what real life code should usually do. I think you can simply combine those two streams roughly along the lines of

fs2.io
        .readClassLoaderResource[SyncIO]("twitter_msgpack.txt", 4096)
        .through(fs2.text.utf8.decode)
        .map(str => Chunk.byteVector(ByteVector.fromHex(str).get))
        .unchunks
        .through(fs2.data.msgpack.low.items[SyncIO])
        .compile
        .toList
        .unsafeRunSync()

[jarmuszz]

Because the twitter_msgpack.txt file contained space separated hex representations of bytes, the fs2.text.utf8.decode pipe could slice some bytes in half between two chunks. E.g. "A1 B2 C3" would be converted into Chunk(0x0a, 0x1b) and Chunk(0x02, 0xc3) if our byte stream had a chunk size equal to 4.

We could remove spaces from input data but even then we would rely on the chunk size being a multiple of two which I think is a bit fragile and hard to debug.

Instead, I converted the test data into a raw binary form which is loaded into memory as a chunked stream before benchmarks are run. The implementation looks a bit clunky but I'm not aware of any better way to achieve this.

[satabin]

Hi @jarmuszz. I am not sure anymore what the state of this PR is. Is it ready to be re-reviewed or do you still want to address some comments or improvements?

[jarmuszz]

Hi @satabin,
Sorry for the confusion and a late reply.

I'll try to address these comments this weekend. I've been trying to combine work with full-time studies which turned out to be more time sinking than I imagined 😃. I hope that in the following weeks I'll be working on the fs2-data-msgpack project in a more regular fashion since I'm resigning from the job.