Messagepack-csharp: How to serialize/deserialize multiple MessagePackObjects from a network Stream?

[VitaminCpp]

Hi,

I know that there exists a streaming API Deserialize(stream, readStrict).
But this method throws an exception, if the last MessagePackObject isn't fully available yet. So IMHO this method only works, if ALL bytes are available yet.
So what's the correct way to deserialize multiple continuous MessagePackObjects from a stream where not all bytes are available yet? Im looking for a method like "bool TryDeserialize(bytes, offset, count)" or something.

[TsengSR]

As far as I understand you have to track it yourself. But using Microsofts new System.IO.Pipelines is a good use case for it. You can write continuous buffer (which will be managed, rented and released/put back into the buffer pool) by the Pipelines itself. So all you need is to check if its complete and the parse it.

A good starting point is David Fowlers great blog post System.IO.Pipelines: High performance IO in .NET.

Also check this comment on that same issue.

I think maybe it is about your transport system's issue.
MemoryStream is not a solution.
If it seems to be divided into Chunk, you need to locate the boundaries and connect them.

[neuecc]

Next version tries to support Span and Pipelines, it will solve this issue.
https://github.com/AArnott/MessagePack-CSharp/issues/9

[VitaminCpp]

Alright! Thanks very much for this information! I'll have a look on these pipelines.

[AArnott]

The change @neuecc referred to is now merged into master of this repo. But it doesn't exactly fix your use case. I'm evaluating an additional change that would allow you to do this, but for now I would recommend you create your own means to understand the boundaries of a MessagePack top-level object. For example try using the substream APIs as described here.

[TsengSR]

@AArnott: Interesting concept. Would that allows one to let's say, have multiple threads deserialize into the same stream where the data is not continuous?

I mean, thread 1 deserializes an object, writes 512 bytes to it, thread 2 deserializes an object, writes 512 bytes to it, thread 1 writes the next 512 bytes etc.?

[AArnott]

No, streams are still sequential and permit only one reader or writer at a time.
But that same library I referenced offers a multiplexing stream that does let multiple i/o threads intermix. But it targets interactive parties on both sides of the stream--not just saving to disk, for example.

[wjax]

What about prefixing the length of the message?
For streams like TCP, it is a must to be able to send the length and on deserialize keep track of it.

Protobuf has this option but messagepack is faster, less allocaty and support typeless... ;)

[AArnott]

Prefixing the length of the message is great.

[wjax]

If you guide me I would like to add it.
I have already look at the code.

Could it be done with PrefixingBufferWriter?

El mié., 25 sept. 2019 19:25, Andrew Arnott [e-mail]
escribió:

Prefixing the length of the message is great.

—
You are receiving this because you commented.
Reply to this email directly, view it on GitHub
https://github.com/neuecc/MessagePack-CSharp/issues/388?email_source=notifications&email_token=AACLVHV2JX7PQRZVFGB3HM3QLONH5A5CNFSM4GWQQKB2YY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOD7SWBHY#issuecomment-535126175,
or mute the thread
https://github.com/notifications/unsubscribe-auth/AACLVHWBZFGMN2PKNFT7OHDQLONH5ANCNFSM4GWQQKBQ
.

[AArnott]

Yes, the PrefixingBufferWriter would be good for this.

[Vasilich]

@AArnott
@neuecc
does that mean, that if i consume messagepack messages, coming over tcp from the device i cannot influence, i have no chance to split those messages correctly in case the other side (device) still sending those, and my socket receive buffer was not large enough to fit all messages completely?

[AArnott]

The best you can do if you have a network stream that you're deserializing from, that contains multiple msgpack messages with no delimiter or header, is to simply keep trying to deserialize each time you get more data. In 2.0 you'll get a MessagePackSerializationException with an inner EndOfStreamException if there's not enough data for a single message so you can just wait for more data. And if you do get a successful message back, you'll need to know how many bytes were consumed so you can save the rest for the next message. You can see where I've done something like this in #663. My fix only applies to seekable streams (network streams are not seekable), but you could do something similar for your non-seekable stream.

I can imagine some optimizations on this area. For example you could avoid the cost of deserializing till you know you have a full message by calling MessagePackReader.Skip(), which would throw if it had an incomplete message. We might also add a method that allows you to check for a full message instead of throwing an exception.

[AArnott]

I'm going to add a change to #663 to make sure we throw a consistent exception whenever we encounter a message that is too short.
Avoiding an exception (that you must catch) altogether is a larger work item than I can take on right now.

[AArnott]

I have an idea to add an API that assists in asynchronously reading from a stream, and produces a ReadOnlySequence<byte> for each top-level msgpack token as it becomes completely available. It might not actually deserialize these sequences since we have the MessagePackSerializer class that you can use to do that.

So if the API we offered did nothing but take a Stream or PipeReader and help you identify the ReadOnlySequence<byte> instances from it that you can successfully deserialize, would that suffice?

[Vasilich]

So if i could receive events that there are byte sequences that i can completely deserialize, then - sure, it would be the solution for me!

[AArnott]

Feedback welcome on the MessagePackStreamReader class that I'm authoring.

Focus on the API and possibly tests at this point. The implementation relies on exceptions being thrown from MessagePackReader as the test for a complete message but that will eventually be fixed so no exceptions are thrown.

[AArnott]

Here is a draft of an alternative solution.

@neuecc: thoughts?

[neuecc]

@AArnott
As a situation when writing to Stream
For example, a JSON log with NewLine as a delimiter

{..}
{..}
{..}

It is common to write like above data.

It is also replace to MessagePack, in that case, it would be a simple sequence of MessagePack blocks.

I thought we needed a wrapper like StreamReader that integrated read-ahead and optimal buffer management.
That is, it is your suggested MessagePackStreamReadingState.
Isn't it?

[wjax]

Hi There.

I have been missing but reading all the emails.

Shouldn't be great to have something like
Synchronous: Serializer.DeserializeWithLengthPrefix that blocks until
a complete object is available.
Async: Serializer.DeserializeWithLengthPrefixAsync that return a
Task ?

And the same to the Serializing part.
Serializer.SerializeWithLengthPrefix.

That is the way Protobuf-net works and is extremely convenient for TCP
sockets.

Thanks for the effort anyway (wheter you can add it or not).

El lun., 2 dic. 2019 a las 3:29, Andrew Arnott ([e-mail])
escribió:

Here is a draft of an alternative solution
https://github.com/neuecc/MessagePack-CSharp/compare/master...AArnott:fix388b
.

@neuecc https://github.com/neuecc: thoughts?

—
You are receiving this because you commented.
Reply to this email directly, view it on GitHub
https://github.com/neuecc/MessagePack-CSharp/issues/388?email_source=notifications&email_token=AACLVHX3VONQAU6R2KCMYM3QWRXHZA5CNFSM4GWQQKB2YY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOEFR72DQ#issuecomment-560200974,
or unsubscribe
https://github.com/notifications/unsubscribe-auth/AACLVHST4PJAXEXZNRWKU2TQWRXHZANCNFSM4GWQQKBQ
.

[AArnott]

@neuecc Yes, I think the most natural use is the MessagePackStreamReader class which is shown here. But the alternative is to achieve the same goal with an API that remains entirely in the MessagePackSerializer class as shown here.

@wjax Adding length-prefixing functions isn't something that IMO should be directly supported by the messagepack library since it's such a common need across all protocols. We have this functionality in Nerdbank.Streams that you can couple to MessagePack to get a complete solution (e.g. PrefixingBufferWriter and/or substreams).
However as was suggested elsewhere recently (I can't remember where as we've had related conversations in several places) we might add a Deserialize(Stream, int bytesToRead) overload where if you knew the length (because you read the prefix yourself) you could specify it to prevent the Deserialize(Stream method from reading too far. But even this isn't necessary if you use the Stream.ReadSlice extension method.

[wjax]

I do not share your view Andrew. Adding a length prefix is not something
out of the scope of a serializer IMHO.
I really think that it is a wonderful feature of a serializer.

A user can create its own way to extend MessagePack functionality to add it
but I believe it would be better implemented in the MessagePack core source.

Regards

El lun., 2 dic. 2019 a las 14:40, Andrew Arnott ([e-mail])
escribió:

@neuecc https://github.com/neuecc Yes, I think the most natural use is
the MessagePackStreamReader class which is shown here
https://github.com/neuecc/MessagePack-CSharp/compare/master...AArnott:fix388.
But the alternative is to achieve the same goal with an API that remains
entirely in the MessagePackSerializer class as shown here
https://github.com/neuecc/MessagePack-CSharp/compare/master...AArnott:fix388b
.

@wjax https://github.com/wjax Adding length-prefixing functions isn't
something that IMO should be directly supported by the messagepack library
since it's such a common need across all protocols. We have this
functionality in Nerdbank.Streams that you can couple to MessagePack to get
a complete solution (e.g. PrefixingBufferWriter
https://github.com/AArnott/Nerdbank.Streams/blob/master/doc/PrefixingBufferWriter.md
and/or substreams
https://github.com/AArnott/Nerdbank.Streams/blob/master/doc/Substream.md
).
However as was suggested elsewhere recently (I can't remember where as
we've had related conversations in several places) we might add a Deserialize(Stream,
int bytesToRead) overload where if you knew the length (because you read
the prefix yourself) you could specify it to prevent the
Deserialize(Stream method from reading too far. But even this isn't
necessary if you use the Stream.ReadSlice
https://github.com/AArnott/Nerdbank.Streams/blob/master/doc/ReadSlice.md
extension method.

—
You are receiving this because you were mentioned.
Reply to this email directly, view it on GitHub
https://github.com/neuecc/MessagePack-CSharp/issues/388?email_source=notifications&email_token=AACLVHVQCIWZWJWZMCRFL7LQWUF5ZA5CNFSM4GWQQKB2YY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOEFTQUZI#issuecomment-560400997,
or unsubscribe
https://github.com/notifications/unsubscribe-auth/AACLVHVYYU4HQVTUUDKCHM3QWUF5ZANCNFSM4GWQQKBQ
.

[AArnott]

The MessagePackSerializer.Deserialize(Stream, ref state) method has hit a snag: there's no way other than throwing an exception to indicate that the end of the stream was encountered even if the stream ended cleanly at the end of a msgpack structure. That is, if you're just looping until the stream is over, deserializing everything in it, you're bound to throw an exception at the end. I don't like an API design that requires that you throw exceptions in expected cases.
This, and the fact that this is resulting in a lot of duplicate code between the two Deserialize overloads and two DeserializeAsync overloads, and the need to use StrongBox<T> on the DeserializeAsync method which is a type in an uncommon namespace that most people aren't familiar with, is all leading me back to prefer the MessagePackStreamReader class approach.

@neuecc: more thoughts?

[neuecc]

Is the usage of MessagePackStreamReader like follwing?

using(var reader = new MessagePackStreamReader(stream))
{
    ReadOnlySequence<byte>? data;
    while((data = await reader.ReadAsync(token)) != null)
    {
        _ = MessagePackSerializer.Deserialize<T>(data.Value);
    }
}

It seems good.

[AArnott]

Yes, @neuecc. That is it. Thanks for the confirmation. I'll proceed with work on it.

Do you think we should keep the MessagePackSerializer.Deserialize{Async}(Stream) methods (with an xml doc comment suggesting folks use MessagePackStreamReader for multiple sequential reads), or should we remove those methods and push people to the new class?