Actor Lifecycle terminated abruptly in Akka.Dotnet.Streams

[tiago-andrade-dvs]

Hello.
This question is related to #6246. But this is a different problem. (Does it makes sense to merge these discussions ?)

Basically I want to be able to receive a really big amount of HTTP requests and every second send somewhere else request with the combined content of the requests received in that second.

I thought that Using Akka.Streams would allow me to create an actor (sourceTask from the code snippet) per request received and that sending the HTTP response as fast as possible would be key.

Here's my attempt. This fails for many requests but works if you send like 6 requests and use .Grouped(4), for example.

       static async Task Main(string[] args)
       {

           var actorSystem = ActorSystem.Create("AkkaStreams"); // Also tested to use a config file. Didn't seem to help
           var materializer = actorSystem.Materializer();

            Source<Akka.Streams.Dsl.Tcp.IncomingConnection, Task<Akka.Streams.Dsl.Tcp.ServerBinding>> connections
                = actorSystem.TcpStream().Bind("127.0.0.1", 8888);

            var (serverBind, source) = connections.PreMaterialize(materializer);

           var sourceTask = source.RunForeach(connection =>
           {
               //Console.WriteLine($"New connection from {connection.RemoteAddress} at {DateTime.Now.ToString("dd-MM-yyyy_HH:mm:ss.zzz")}");
               Flow<ByteString, ByteString, Akka.NotUsed> flShape = Flow.FromGraph(
                   GraphDsl.Create(b =>
                   {
                       var httpReceiver = Flow.Create<ByteString>()
                           .Select(c =>
                           {
                               string strValue = c.ToString();
                               string[] bodyHelper = strValue.Split(Environment.NewLine + Environment.NewLine);
                               string result = (bodyHelper.Length > 1) ? bodyHelper[1] : strValue;
                               return ByteString.FromString(result);
                           });

                       var post1 = Flow.Create<ByteString>()
                       .GroupedWithin(600, TimeSpan.FromMilliseconds(1000))
                       //.Grouped(4) // Works for very few requests recieve
                       .Select(x => x.Aggregate(ByteString.Empty, (s, byteString) => s.Concat(byteString)))
                       .SelectAsync(1, async c =>
                       {
                           // Console.WriteLine("POST Request");
                           Console.WriteLine(c.ToString());
                           return c;
                       });

                       var response = Flow.Create<ByteString>()
                           .Select(c =>
                           {
                               var cont = c.ToString();

                               var sb = new StringBuilder("HTTP/1.1");
                               sb.Append(" 200 OK\n");
                               sb.Append("Content-Type: text/plain\n");
                               sb.Append("Content-Length: " + cont.Length);
                               sb.Append("\n\n");
                               sb.Append(cont);

                               // Console.WriteLine($"Response");
                               return ByteString.FromString(sb.ToString());
                           });

                       var httpReceiverFlow = b.Add(httpReceiver);
                       var post1Flow = b.Add(post1);
                       var responseFlow = b.Add(response);

                       var broadcastFlow = b.Add(new Broadcast<ByteString>(2));
                       b.From(httpReceiverFlow).To(broadcastFlow);
                       b.From(broadcastFlow.Out(0)).To(responseFlow);
                       b.From(broadcastFlow.Out(1))
                           .Via(post1Flow)
                           .To(Sink.Ignore<ByteString>());

                       return new FlowShape<ByteString, ByteString>(httpReceiverFlow.Inlet, responseFlow.Outlet);
                   }));

               connection.HandleWith(flShape, materializer);
           }, materializer);

           await serverBind;
           await actorSystem.WhenTerminated;
       }

This error happens when I attempt to send 600 HTTP requests at the same time. It can be related to the limit of actors on a system, but I tried to reconfigure akka.stream.materializer.max-input-buffer-size from 16 to 512 and I didn't get anywhere.

This Warning message happens when I send 600 requests but with 0.1 seconds between each other. In here it also seems like the .post1 Flow is "attached" the TCP connection, even after a HTTP response has been sent.

Thank you for reading.
Hope I am not "wasting" anyone's time with too "basic" questions.

[to11mtm]

Have you tried adding some try/catch logging to your flow logic to see where things are failing?

Unless I'm missing something there may be a lack of framing in the bytestrings received, which could cause some issues in your parse logic:

var httpReceiver = Flow.Create<ByteString>()
    .Select(c =>
    {
        string strValue = c.ToString();
        string[] bodyHelper = strValue.Split(Environment.NewLine + Environment.NewLine);
        string result = (bodyHelper.Length > 1) ? bodyHelper[1] : strValue;
        return ByteString.FromString(result);
    });

If you don't have proper framing on your reads, it is possible for you to receive 'incomplete' data at this stage.

Normally, when sending data to/from Streams this way, I use .Via(Framing.SimpleFramingProtocolDecoder(TransportSettings.MaxFrameSize)) on both the sending and receiving side.

In the case of HTTP, you may need to write your own framing logic, but SimpleFramingProtocolDecoder may be a good reference for something to adapt to HTTP.