Jetstream Out of memory error

[arvindsouza]

Hello all,

I'm attempting to benchmark Jetstream for a computer vision project.

Code is below

The payload I am testing is an image file of ~0.4 MB

I am trying to publish a large number of messages, from multiple Tasks & with the same payload, to the jetstream server at once; However, the server throws the below error

"runtime: out of memory: cannot allocate 4194304-byte block (1761312768 in use)
fatal error: out of memory"

after some time

I have set the jetstream config as below:
max_payload: 1M
jetstream: enabled
jetstream {
store_dir: /data/jetstream
max_mem: 100G
max_file: 100G
}

I have also tested by changing MaxBytes and MaxMessages. as well as adding a delay before each publish, which seems to help

I am trying to determine the root cause of this issue and how to avoid it. From my understanding setting the limits should make Jetstream discard messages once limits are reached, maintaining stream size regardless of volume of messages and speed of publishing

Below is the code

            string streamName = "test";
            string streamName2 = "TEST";
            int totalMessageCountPerThread = 10000;
            NatsConnection nats = new NatsConnection();
            int delay = 100;
            var jetstream = new NatsJSContext(nats);

            mStream = await jetstream.CreateStreamAsync(new StreamConfig(streamName2, new[] { $"{streamName}.>" })
            {
                Retention = StreamConfigRetention.Limits,
                MaxBytes = -1,
                MaxMsgs = 1000,
                Discard = StreamConfigDiscard.Old,
                Compression = StreamConfigCompression.S2,
                Storage = StreamConfigStorage.File
            });

            FileInfo img = new FileInfo("pictures\\images.jpg");
            byte[] file = File.ReadAllBytes("pictures\\images.jpg");

            Console.WriteLine(file.Length);


            Task.Run(async () =>
            {

                for (int i = 0; i < totalMessageCountPerThread; i++)
                {
                    TransportUnit dataToSend = new TransportUnit();
                    dataToSend.DatapointKey = this.DpKey;
                    dataToSend.DatapointValue = file;
                    dataToSend.DatapointName = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff", CultureInfo.InvariantCulture);
                    jetstream.PublishAsync($"{streamName}.picture", ProtoHelper.SerializeCompressedToBytes(dataToSend));
                    Console.WriteLine($"Published {dataToSend.DatapointKey} @ {dataToSend.DatapointName} total: {i}");
                   // Console.WriteLine($"Stream size {jetstream.GetStreamAsync(streamName2).Result.Info.State.Bytes}");

                }
            });

            Task.Run(async () =>
            {

                for (int i = 0; i < totalMessageCountPerThread; i++)
                {
                    TransportUnit dataToSend = new TransportUnit();
                    dataToSend.DatapointKey = 2.ToString();
                    dataToSend.DatapointValue = file;
                    dataToSend.DatapointName = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff", CultureInfo.InvariantCulture);
                    jetstream.PublishAsync<byte[]>($"{streamName}.picture", ProtoHelper.SerializeCompressedToBytes(dataToSend));
                    Console.WriteLine($"Published at {dataToSend.DatapointKey} total: {i}");

                }
            });

            Task.Run(async () =>
            {

                for (int i = 0; i < totalMessageCountPerThread; i++)
                {
                    TransportUnit dataToSend = new TransportUnit();
                    dataToSend.DatapointKey = 3.ToString();
                    dataToSend.DatapointValue = file;
                    dataToSend.DatapointName = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss.fff", CultureInfo.InvariantCulture);
                    jetstream.PublishAsync<byte[]>($"{streamName}.picture", ProtoHelper.SerializeCompressedToBytes(dataToSend));
                    Console.WriteLine($"Published at {dataToSend.DatapointKey} total: {i}");

                }
            });
}

[mtmk]

you should await the publish calls:

var ack = await jetstream.PublishAsync($"{streamName}.picture", ...);
ack.EnsureSuccess();

Edit: JetStream publish works with acknowledgments. clients send a message and wait for a confirmation from the server that it's persisted. with .NET client when you don't await (or not keep track of the returned tasks and throttle in some fashion) you can send messages way faster then they can be persisted.

I think that is what you're observing with server queuing up messages internally while waiting for disk IO. I'm not a server expert though, @neilalexander please feel free to add to it or correct me

[arvindsouza]

That seems to have resolved it, I thought it was related to the publish speed, but was expecting the server to reject messages once the limits were exceeded, regardless of speed.

Thank you very much!

[mtmk]

@aricart @neilalexander is there a server setting to limit the internal buffer (or something) to avoid OOM errors?

[arvindsouza]

Just a quick follow up @mtmk could you point me to some good sample applications that use the v2 library? I'm still experiencing issues like slow consumers, "Internal subscription took too long" etc. on the nats-server while running benchmarks

I don't know if this has to do with config settings or disposing streams/consumers properly but a good sample application would be really helpful

[mtmk]

we some snippets in docs https://nats-io.github.io/nats.net.v2/documentation/jetstream/intro.html

If you can share your application I'm happy to have a look

edit also this https://natsbyexample.com/

edit2 also have a look here as well, in case your server doesn't have enough juice https://docs.nats.io/running-a-nats-service/introduction/installation#hardware-requirements

[arvindsouza]

Thank you! Here is the link to the repo

https://github.com/arvindsouza/Nats-Benchmarks

It isn't the neatest bit of code, wasn't expecting to share it. Please let me know if anything is unclear

I don't think it's a server issue, I have been testing on one with 256GB of RAM and 12 CPU cores.

[mtmk]

no worries. I get 404. maybe not public yet?

[arvindsouza]

Changed to public

[mtmk]

thanks @arvindsouza, these seemed to help: (arvindsouza/Nats-Benchmarks#1)

• Setting SubPendingChannelFullMode = BoundedChannelFullMode.Wait
• Using multiple connections for publish

let me know how it goes.

(note we also have plans to improve jetstream publish performance #375)