Streams TCP Transport with Hand-Written Protobuf Deserialization #4594

to11mtm · 2020-10-28T03:01:52Z

Last One, Promise. :)

This branch has a few more tweaks:

Hand written Deserializer for Remote Protobuf payloads to minimize copies on read.
LRUSegmentCache for using Span<byte> for Input (i.e. UTF8 Actor Paths, only copying to string if needed)
Switching to ThreadStatic from ThreadLocal where possible
Moving Stream Materializer for Streaming transport to it's own dispatcher config. At the moment it is using thread pool which lowers performance a bit, but hopefully can be brought up with tweaks to batch settings or thread pool throughput.

… in perf. Could still use more tweaks.

…for IO.TCP Bindings/connections to have buffer pool sizes set separate from default settings. Use Compiled Expression to avoid copy of Protobuf ByteString to IO.Bytestring on Streams Transport Write. Create internal IO.Bytestring ReadOnlyCompacted to minimize copying on Reads for Streams Transport.

… change

…ort to get rid of one more copy.

…ransport. Use .ToArray() instead of .ToList() in ActorPath

This reverts commit 4b3d081.

- LRUSegmentCache for caching based on ReadOnlySpan<Byte> segments. - Move All parts of Streaming TCP transport possible to their own Dispatcher for stability

Aaronontheweb · 2020-10-28T21:55:42Z

I'll take a look!

…ppy about it relatively speaking. Use ValueTuple return version of SplitNameAndUid to lower allocations for things like ActorPath op /

to11mtm · 2020-10-29T16:49:28Z

What This needs, copied/adapted from #4581

Needs AKA "I need help!" :

Tests around SSL Concept
Any adaptations necessary for SSL in transport (I think this will actually be minimal thanks to the pattern of it just being an extra Inet.SocketOption)
Hardcoded entries moved into configuration (I'll try to work on this if I have time.)
- Config entries are no longer hard coded, but we should decide whether to fill in certain parts or fallback to Dotnetty Settings.
Hardening around using no delay in batch grouping. RemotePingpong becomes unstable in this scenario.
Potentially testing/hardening around public/private configuration scenarios. TCP is not my forte.
Checks against dotnetty/helios compatibility. When I squint at DotNetty I'm hopeful the Akka.Streams framing is the same. If it's not we would have to make framing stages for that as well as helios. But it certainly seems simple.
MUST Check/Fix Not entirely certain the protobuf varint128 code handles negative numbers correctly. I borrowed it from the google repo but might have picked something that didn't do negative numbers
This is a very strict and opinionated parser. We may want to provide a fallback mode in case the way older clients encode messages is not able to be decoded. Google's protobuf serializer does NOT guarantee deterministic order of fields, this parser expects it.

Nice to have:

Someone who knows streams better than I may know a few tricks to make the pipeline better. The places I put Async stages at were based on repeated RemotePingPong benchmarks as well as some debug checks (i.e. checking how many bytes were being pushed in the final bufferedSelectFlow stage.)

to11mtm · 2020-10-29T16:51:24Z

@Aaronontheweb I added what this needs if anyone in the community or elsewhere wishes to help!

If some of this is too extreme, we can move the important parts (scheduler, batch tweaks,etc) and still do a couple more zero-copy tricks on one of the other branches.

FWIW I'd say this is the most 'extreme' branch, The one where we are only serializing the protobuf payloads I would say is pretty dang safe.

Aaronontheweb · 2020-10-29T18:20:58Z

@to11mtm have any before / after performance numbers you can share?

to11mtm · 2020-10-30T00:07:48Z

These are round-about numbers ATM, will try to run more real benchmarks later.

But these numbers are fair in that they do show some scale:

RemotePingPong, I7 8750H, 32GB Ram.

Everything (Materializer, IO.TCP,remoting) running under single `dedicatedthreadpool`:

Streaming, No special Protobuf Serialization (i.e. just minimizing copies from ByteString to ByteArray for serialization/deserialization:
~200,000-210,000 messages per sec for numclients >1

Dedicated Thread Pool for Materializer/IO.TCP, Remoting under Remoting Thread Pool:

Streaming, No special Protobuf Serialization (i.e. just minimizing copies from ByteString to ByteArray for serialization/deserialization:
- ~170,000-180,000 messages per sec for numclients>1
Streaming, Hand-Written Protobuf Serializer/Deserializer (this branch)
- 220,000-240,000 messages per sec for numclients >1
DotNetty, TCP
- 220,000-240,000 Messages per sec for numclients >1

Running streaming all streaming transport bits (i.e. Materializer, remoting, IO.TCP actors) on `akka.actor.default-dispatcher`:

Streaming, No special Protobuf Serialization
- ~210,000-240,000 messages per sec for numclients >1
Streaming, Hand written Protobuf Serializer/Deserializer (This branch)
- ~240,000-300,000 Messages per sec for numclients >1

to11mtm · 2020-10-30T00:34:02Z

I added one more bench number to above, since I think it does a decent job of showing that the more threadpools we have involved the more things seem to hurt.

That said, I think these are some good baselines for everyone to see where things are at.

Aaronontheweb · 2020-10-30T00:35:21Z

@to11mtm started work reviewing this PR - been working my way through it. I'll go ahead and try to run these benchmarks locally too.

Aaronontheweb · 2020-10-30T00:36:36Z

I added one more bench number to above, since I think it does a decent job of showing that the more threadpools we have involved the more things seem to hurt.

The mutual exclusion benefits don't really show up well under single connection benchmarks - but we definitely need to be judicious about limiting the number of independent threadpools we use. Returns diminish quickly on each additional one.

Aaronontheweb

Leaving my initial comments up - not done yet, but best I could get to this evening

Aaronontheweb · 2020-10-29T18:21:29Z

src/benchmark/RemotePingPong/Program.cs

+            AppDomain.CurrentDomain.UnhandledException += (sender, eventArgs) =>
+            {
+                Console.WriteLine(eventArgs.ExceptionObject as Exception);
+                Console.WriteLine("STACKOVERFLOW");


How did this come up inside the benchmark?

Uhhhh, it rather involved some Spans and FastHash while I was debugging... :)

Aaronontheweb · 2020-10-29T18:21:46Z

src/benchmark/RemotePingPong/Program.cs

+                Console.WriteLine(sender);
+                Console.WriteLine(eventArgs.ExceptionObject as Exception);
+            };
+            Process.GetCurrentProcess().PriorityClass = ProcessPriorityClass.AboveNormal;


We should probably port that change back into a separate PR - good idea.

edit: just to clarify - this change is so good and small that it should find its way into the dev branch straight away

Aaronontheweb · 2020-10-29T18:22:08Z

src/contrib/transports/Akka.Remote.Transport.Streams.Tcp/Class1.cs

@@ -0,0 +1,313 @@
+// //-----------------------------------------------------------------------


Need to rename this file

Aaronontheweb · 2020-10-29T18:23:05Z

...ontrib/transports/Akka.Remote.Transport.Streams.Tcp/Akka.Remote.Transport.Streams.Tcp.csproj

@@ -0,0 +1,12 @@
+<Project Sdk="Microsoft.NET.Sdk">
+


Need to reference common.props here.

For example:

akka.net/src/contrib/persistence/Akka.Persistence.Sqlite/Akka.Persistence.Sqlite.csproj

Lines 2 to 10 in 3705c8c

<Import Project="..\..\..\common.props" />

<PropertyGroup>

<AssemblyTitle>Akka.Persistence.Sqlite</AssemblyTitle>

<Description>Akka.NET Persistence journal and snapshot store backed by SQLite.</Description>

<TargetFrameworks>$(NetStandardLibVersion)</TargetFrameworks>

<PackageTags>$(AkkaPackageTags);persistence;eventsource;sql;sqlite</PackageTags>

<GenerateDocumentationFile>true</GenerateDocumentationFile>

</PropertyGroup>

Aaronontheweb · 2020-10-29T18:29:40Z

src/contrib/transports/Akka.Remote.Transport.Streams.Tcp/Class1.cs

+
+            try
+            {
+                return _queue?.OfferAsync(IO.ByteString.FromBytes(payload.ToByteArray())).Result is QueueOfferResult


Should we just change the API here to make Write an async method? Having the blocking .Result call here and the same types of stuff we have to do for DotNetty is less than ideal.

(Question still valid on live code so I'll just answer here:)
It proably wouldn't hurt. I was suprised at how much time .OfferAsync was taking in the profiler.

IIRC the challenge was the actors calling it aren't ReceiveActors and I don't know how to make other ones work with Async :)

Aaronontheweb · 2020-10-30T01:09:05Z

src/contrib/transports/Akka.Remote.Transport.Streams.Tcp/Class1.cs

+        public sealed override string SchemeIdentifier { get; protected set; } =
+            "tcp";
+        private ActorMaterializer _mat;
+        private DotNettyTransportSettings Settings;


nice, for making this backwards compatible...

Aaronontheweb · 2020-10-30T01:20:18Z

src/contrib/transports/Akka.Remote.Transport.Streams.Tcp/Class1.cs

+                            ex =>
+                            {
+                                handle.Notify(
+                                    new Disassociated(DisassociateInfo.Unknown));


Good way to error-handle this

Aaronontheweb · 2020-10-30T01:21:18Z

src/contrib/transports/Akka.Remote.Transport.Streams.Tcp/Class1.cs

+                    listener.Notify(new InboundAssociation(handle));
+                });
+                return NotUsed.Instance;
+            }).ToMaterialized(Sink.Ignore<NotUsed>(),Keep.Left).Run(_mat);


We need to pass in some hints to the materializer here as to which dispatcher the materialized actors should run on, IIRC

the /core/akka.remote/ code has dispatcher hints :)

Ah ok - I was wondering if I was missing something

Aaronontheweb · 2020-10-30T01:22:29Z

src/contrib/transports/Akka.Remote.Transport.Streams.Tcp/Class1.cs

+                .Queue<IO.ByteString>(512, OverflowStrategy.DropNew)
+                .Via(
+                    Framing.SimpleFramingProtocolEncoder(Settings.MaxFrameSize))
+                .GroupedWithin(256, TimeSpan.FromMilliseconds(40))


Nice, a much more concise way of doing the "batching" than what our current transport does. We'll still need a flag to disable the batching stage via config for connections with low traffic (i.e. a system like Lighthouse)

Not gonna lie, I want to FRAME that block of code. While I know there's room for improvement it just feels so elegant.

Aaronontheweb · 2020-10-30T01:22:51Z

src/contrib/transports/Akka.Remote.Transport.Streams.Tcp/Class1.cs

+        {
+            //TODO: Improve batching voodoo in pipeline.
+            return Source
+                .Queue<IO.ByteString>(512, OverflowStrategy.DropNew)


Probably better for us if we just reject the new write to the queue, so the EndpointWriter can backoff

@Aaronontheweb my bad on this, you want to look at /src/core/akka.remote/ for the most up to date bits.

Moving everything into its own /contrib/ will be a bit more work because of all the internal things a transport needs

I think this is a great start - I’m very impressed with your hard work. I think this will be an amazing improvement over DotNetty

Looking forward to reading the rest of it

to11mtm · 2020-10-30T02:22:10Z

Leaving my initial comments up - not done yet, but best I could get to this evening

No problem. In fact, I apologize in advance for AkkaPduCodec.cs . You may want to skip that for the moment; right now it is a poster-child for "Why Projects have Contributor Guidelines".

…m/to11mtm/akka.net into remote-full-manual-protobuf-deser

…g, minor change to HWTS.

to11mtm · 2020-10-31T16:03:39Z

src/benchmark/RemotePingPong/Program.cs

@@ -182,16 +202,18 @@ private static async Task<(bool, long, int)> Benchmark(int numberOfClients, long
                throw new Exception("Received report that 1 or more remote actor is unable to begin the test. Aborting run.");
            }

+            var rng = new Random();
+            var rand = new byte[2048];
+            rng.NextBytes(rand);


I had this bit in here because it was interesting to note that some of my tweaks had to change when I started throwing messages a bit larger; If this number used passing this differs a LOT from smaller messages in ping-pong, changing some of the batching parameters (i.e. count, size) usually brings things back in-line.

to11mtm · 2020-10-31T16:07:06Z

src/core/Akka.Remote/Configuration/Remote.conf

@@ -359,7 +359,41 @@ akka {
 	helios.tcp.transport-class = "Akka.Remote.Transport.Helios.HeliosTcpTransport, Akka.Remote.Transport.Helios"

    ### Default configuration for the DotNetty based transport drivers
-
+streaming.tcp{


Left a LOT of the options out here even though they can be read from code. Idea here was to make migration from DotNetty as easy as possible

to11mtm · 2020-10-31T16:08:24Z

src/core/Akka.Remote/Configuration/Remote.conf

@@ -602,9 +636,23 @@ akka {
    default-remote-dispatcher {
      type = ForkJoinDispatcher
 	  executor = fork-join-executor
+      throughput = 30


30 is default, but I kinda enjoy the Cognitive hint to Users that this is a number they can touch.
(Also, protects from defaults being changed elsewhere)

to11mtm · 2020-10-31T16:09:15Z

src/core/Akka.Remote/Endpoint.cs

+            FastMessageParser.
+                PayloadParser messageOptionSerializedMessage,
+            IActorRef messageOptionSenderOptional);
+        void Dispatch(IInternalActorRef messageOptionRecipient,


This last overload is an experiement gone wrong in async deserialization and should be removed.

to11mtm · 2020-10-31T16:10:06Z

src/core/Akka.Remote/Endpoint.cs

+            IActorRef messageOptionSenderOptional);
+    }
+
+    public class DefaultMessageDispatcherActor : ActorBase


This needs to be removed or put somewhere else to be figured out,
Part of attempt to parallelize the Remote Dispatch. Failed miserably :)

to11mtm · 2020-11-14T19:11:37Z

src/core/Akka/IO/TlsConnection.cs

+                    //_sendArgs.SetBuffer(_dataToSend);
+#pragma warning disable 4014
+                    Task.Run(async () =>
+                    {
+                        foreach (var byteString in _dataToSend)
+                        {
+                            foreach (var buffer in byteString.Buffers)
+                            {
+                                await _connection.SslStream.WriteAsync(
+                                    buffer.Array,
+                                    buffer.Offset, buffer.Count);
+                            }
+                        }
+
+                        return NotUsed.Instance;


This is probably Bad and Wrong. There's definitely some ugly going on:

Task.Run is a smell

Lots of back-and-forth here with awaits when we have small buffers (i.e. after manual serialization/framing/batching)

to11mtm · 2020-11-14T19:17:53Z

src/core/Akka/Util/ByteString.cs

+        public ByteString(IList<ByteString> buffers)
+        {
+            _count = buffers.Sum(s => s.Count);
+            _buffers = buffers.SelectMany(r=>r._buffers).ToArray();
+        }
+


So, this is useful for some cases, but looking at it I can't help but wish we had a nice way to pack this and fetch lazily (so we don't have to do this SelectMany+ToArray

to11mtm · 2020-11-14T19:21:35Z

src/core/Akka/Util/ByteString.cs

+        internal ByteBuffer ReadOnlyCompacted()
+        {
+            var c = this.Compact();
+            return c._buffers[0];
+        }


The idea here is to get a Single Contiguous Arraysegment back; For smaller messages, this will most likely not copy/move anything in .Compact() step.

However in either case, the ArraySegment(ByteBuffer) can then be used directly either by 'Proper' Protobuf deserializer or custom deserializer.

to11mtm · 2020-11-14T19:23:04Z

src/core/Akka/Util/ThreadLocalRandom.cs

-        private static ThreadLocal<Random> _rng = new ThreadLocal<Random>(() => new Random(Interlocked.Increment(ref _seed)));
+        [ThreadStatic] 
+        private static Random _rng2;
+        //private static ThreadLocal<Random> _rng = new ThreadLocal<Random>(() => new Random(Interlocked.Increment(ref _seed)));


Avoiding ThreadLocal wherever we don't need it for some special reason.

Reasoning here: https://ayende.com/blog/189761-A/production-postmortem-the-slow-slowdown-of-large-systems

to11mtm · 2020-11-14T19:23:52Z

src/core/Akka.Streams/Implementation/Fusing/GraphInterpreter.cs

-        private static readonly ThreadLocal<object[]> CurrentInterpreter = new ThreadLocal<object[]>(() => new object[1]);
+        [ThreadStatic]
+        private static object[] CurrentInterpreter;


Changing to ThreadStatic because of https://ayende.com/blog/189761-A/production-postmortem-the-slow-slowdown-of-large-systems

to11mtm added 21 commits October 2, 2020 23:38

save state

0cbffbf

save

d4e7c8e

Clean up materialization (arguably)

a2217e2

improve batching

52e931b

improve stage

831d032

Tweak Stage to where we are are about equal to Dotnetty with batching…

eb42e44

… in perf. Could still use more tweaks.

save batch updates and other fixes

5c93ec5

Update with fixes for DNS

0baec4f

clean up

72d8a5e

Make sure Protobuf bytearray hack won't blow up if protobuf internals…

07f7128

… change

Add Configuration bits

c7a3e5e

Serialize Protobuf manually using IO.ByteString on AkkaProtocolTransp…

173d15d

…ort to get rid of one more copy.

Clean up fixes to BatchWriter

9e4f20e

Allow ArraySegments to be used in InboundPayload to lower copies in T…

08739f9

…ransport. Use .ToArray() instead of .ToList() in ActorPath

Try manually serializing protobuf payloads

1efd76f

test for reverse

4b3d081

Revert "test for reverse"

985bc11

This reverts commit 4b3d081.

checkpoint all the things

e8551e5

working impl

1f5e9e4

- Hand-Written Protobuf Deserializer to minimize allocations

3a98998

- LRUSegmentCache for caching based on ReadOnlySpan<Byte> segments. - Move All parts of Streaming TCP transport possible to their own Dispatcher for stability

Aaronontheweb self-requested a review October 28, 2020 21:55

Aaronontheweb added akka-remote perf labels Oct 28, 2020

Push Changes Where everything is on it's own dispatcher and pretty ha…

98b7e11

…ppy about it relatively speaking. Use ValueTuple return version of SplitNameAndUid to lower allocations for things like ActorPath op /

to11mtm added 2 commits October 29, 2020 12:56

fix unit test

d04fbc1

Merge branch 'dev' into remote-full-manual-protobuf-deser

ba46ce5

Aaronontheweb mentioned this pull request Oct 29, 2020

Akka.Remote + ForkJoinExecutor dispatcher configuration changes #4597

Closed

Aaronontheweb reviewed Oct 30, 2020

View reviewed changes

to11mtm added 3 commits October 30, 2020 18:27

Clean up AkkaPduCodec.cs

2be3022

Merge branch 'remote-full-manual-protobuf-deser' of https://github.co…

f490009

…m/to11mtm/akka.net into remote-full-manual-protobuf-deser

Fix Negative Int32 parsing, be a little smarter about manifest cachin…

3cfd7a2

…g, minor change to HWTS.

Aaronontheweb self-requested a review November 11, 2020 20:10

to11mtm commented Nov 14, 2020

View reviewed changes

to11mtm mentioned this pull request Nov 21, 2020

High Idle CPU in DotNetty #4636

Closed

		@@ -0,0 +1,313 @@
		// //-----------------------------------------------------------------------

	<Import Project="..\..\..\common.props" />

	<PropertyGroup>
	<AssemblyTitle>Akka.Persistence.Sqlite</AssemblyTitle>
	<Description>Akka.NET Persistence journal and snapshot store backed by SQLite.</Description>
	<TargetFrameworks>$(NetStandardLibVersion)</TargetFrameworks>
	<PackageTags>$(AkkaPackageTags);persistence;eventsource;sql;sqlite</PackageTags>
	<GenerateDocumentationFile>true</GenerateDocumentationFile>
	</PropertyGroup>

Streams TCP Transport with Hand-Written Protobuf Deserialization #4594

Are you sure you want to change the base?

Streams TCP Transport with Hand-Written Protobuf Deserialization #4594

Conversation

to11mtm commented Oct 28, 2020

Aaronontheweb commented Oct 28, 2020

to11mtm commented Oct 29, 2020

What This needs, copied/adapted from #4581

to11mtm commented Oct 29, 2020

Aaronontheweb commented Oct 29, 2020

to11mtm commented Oct 30, 2020 • edited Loading

Everything (Materializer, IO.TCP,remoting) running under single dedicatedthreadpool:

Dedicated Thread Pool for Materializer/IO.TCP, Remoting under Remoting Thread Pool:

Running streaming all streaming transport bits (i.e. Materializer, remoting, IO.TCP actors) on akka.actor.default-dispatcher:

to11mtm commented Oct 30, 2020

Aaronontheweb commented Oct 30, 2020

Aaronontheweb commented Oct 30, 2020

Aaronontheweb left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

to11mtm commented Oct 30, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

to11mtm commented Oct 30, 2020 •

edited

Loading

Everything (Materializer, IO.TCP,remoting) running under single `dedicatedthreadpool`:

Running streaming all streaming transport bits (i.e. Materializer, remoting, IO.TCP actors) on `akka.actor.default-dispatcher`:

to11mtm commented Oct 30, 2020 •

edited

Loading