Can I have a small sample of sending 40k messages/second?

[haiduong87]

I have a simple ASP.NET Core API project, hosted as a windows service.
I've tried to optimized anything I can (know), but sending 1000 requests (each request publish 20 messages via nats client) isn't as fast as expectation.

Here's a sample of what's I'm doing:

    [Route("api/[controller]")]
    [ApiController]
    public class ValuesController : ControllerBase
    {
        private IConnection _connection;

        public ValuesController(IConnection connection)
        {
            _connection = connection;
        }

        [HttpPost("notification")]
        public ActionResult PublishMessage(IEnumerable<Notification> payload)
        {
            if (payload == null)
                return BadRequest("Empty payload");

            Task.Run(() =>
            {
                var notificationBatches = payload.Batch(5, true);
                foreach (var notifications in notificationBatches)
                {
                    Parallel.ForEach(notifications, (notification, state) =>
                        {
                            _connection.Publish(
                                Encoding.UTF8.GetBytes(System.Text.Json.JsonSerializer.Serialize(notification)));
                        });
                }
            });
            return Ok();
        }

        public class Notification
        {
        }
    }

    public static class Ext
    {
        /// <see cref="https://stackoverflow.com/a/11775295"/>
        public static IEnumerable<IEnumerable<T>> Batch<T>(
            this IEnumerable<T> source,
            int size,
            bool buffered = false,
            CancellationToken cancellationToken = default(CancellationToken))
        {
            if (source == null) throw new ArgumentNullException(nameof(source));

            if (size <= 0) throw new ArgumentException("Size must great than 0", nameof(size));

            if (buffered == false)
                using (var enumerator = source.GetEnumerator())
                {
                    IEnumerable<T> GetPart<T>(IEnumerator<T> enumerator_, int size_)
                    {
                        do
                        {
                            yield return enumerator_.Current;
                        } while (--size_ > 0 && enumerator_.MoveNext());
                    }

                    while (enumerator.MoveNext())
                    {
                        yield return GetPart(enumerator, size);
                    }
                }
            else
            {
                var batch = new List<T>(size);
                foreach (var item in source)
                {
                    batch.Add(item);
                    if (batch.Count == size)
                    {
                        yield return batch;
                        batch = new List<T>(size);
                    }
                }

                if (batch.Count > 0)
                    yield return batch;
            }
        }
    }

Then I create a small windows app that make requests.
I tried: 1000 requests, each requests with 30 Notifications
Running that test, give some response with high elapsed time (up to 2000ms)

I think the problem is with the connection, when I try above code with out publishing message (just make a task.delay(500)), the highest elapsed time is 600ms

My target: create an asp .net core webapi, that can handle up to 1000 requests, each request has up to 40 messages payload

Please help me.

Duong Nguyen

[ColinSullivan1]

Thank you for using NATS!

I'm not aware of any issues that would slow down publishing that much. You're likely encountering a lot of overhead with the additional tasks being created by Task.Run and Parallel.ForEach.

I'd suggest publishing NATS messages inline in your request handler instead of creating a Task for each request. Internally, the NATS Publish API will buffer data and then another thread will flush messages making the publish APIs extremely efficient. Also, the publish API is thread-safe so the parallelized publish calls would end up serialized anyhow.

Here's a suggested change:

    [HttpPost("notification")]
    public ActionResult PublishMessage(IEnumerable<Notification> payload)
    {
        if (payload == null)
            return BadRequest("Empty payload");
        
        var notificationBatches = payload.Batch(5, true);
        foreach (var notifications in notificationBatches)
        {
            _connection.Publish(Encoding.UTF8.GetBytes(System.Text.Json.JsonSerializer.Serialize(notification)));
        }
        _connection.FlushBuffer(); // Optional to reduce latency
        return Ok();
    }

As you progress, you'll also want be aware the that publish API can throw exceptions and you could respond with with a InternalServerError or another appropriate error for that case.

You'll also want to take care that you don't miss notifications in the batches if there are more than 5.

I hope this helps, let us know how it goes!

[haiduong87]

Thanks for quick reply, I will try it now.

About the payload size, it may up to 40 or 50, then I split it in smaller chunks (in this sample is 5 notifications a patch)
Then I use a parallel foreach to loop in side a small chunk
In my example, there will be 8 parallel foreach statements

About the task, I see there's a sample like this (I thought this best practice)
https://docs.microsoft.com/en-us/aspnet/core/performance/performance-best-practices?view=aspnetcore-3.1#do-not-capture-the-httpcontext-in-background-threads

[haiduong87]

@ColinSullivan1
Sorry, your suggestion doesn't work.
If I remove all Task.Run and Parallel.ForEach,

• Use only 1 connection (singleton) and try these cases:
  • 1000 requests, 5 notifications payload: maximum response time is 3900ms
  • 1000 requests, 20 notifications payload: maximum response time is 15995ms
• Use many connections (a singleton connection pool, contains up to 6 persistent connections)
  • 1000 requests, 5 notifications payload: It take some requests up to 500 ms
  • 1000 requests, 20 notifications payload: maximum response time is about 3000ms

Here's the code of the pool:

    public class NatsConnectionPool
    {
        private readonly LinkedList<IConnection> _connections;
        private readonly bool _useRoundRobbin;

        private LinkedListNode<IConnection> _connetionNode;

        /// <exception cref="T:NATS.Client.NATSNoServersException">No connection to a NATS Server could be established.</exception>
        /// <exception cref="T:NATS.Client.NATSConnectionException">
        ///     <para>A timeout occurred connecting to a NATS Server.</para>
        ///     <para>-or-</para>
        ///     <para>
        ///         An exception was encountered while connecting to a NATS Server. See
        ///         <see cref="P:System.Exception.InnerException" /> for more
        ///         details.
        ///     </para>
        /// </exception>
        public NatsConnectionPool(Configuration configuration)
        {
            _useRoundRobbin = configuration.UseConnectionPool;

            var cf = new ConnectionFactory();
            var opts = ConnectionFactory.GetDefaultOptions();
            opts.AllowReconnect = true;
            opts.MaxReconnect = Options.ReconnectForever;

            _connections = new LinkedList<IConnection>();
            if (_useRoundRobbin)
            {
                for (var i = 0; i < configuration.NatsConnectionPoolSize; i++)
                    foreach (var serverConfig in configuration.NatsServers)
                    {
                        opts.Servers = serverConfig.Split(',');
                        _connections.AddLast(cf.CreateConnection(opts));
                    }
            }
            else
            {
                opts.Servers = configuration.NatsServers[0].Split(',');
                _connections.AddLast(cf.CreateConnection(opts));
            }
        }

        public IConnection Connection
        {
            get
            {
                _connetionNode ??= _connections.First;
                if (_useRoundRobbin)
                    _connetionNode = _connetionNode.Next ?? _connections.First;

                // ReSharper disable once PossibleNullReferenceException
                return _connetionNode.Value;
            }
        }
    }

And the actually code I'm testing (yesterday I posted from home, now I'm at the office)

        [HttpPost("{customer}")]
        public IActionResult PublishNotificationAsync([FromRoute] [Required] string customer,
            [FromBody] IEnumerable<DatabaseNotification> databaseNotifications)
        {
            if (databaseNotifications == null)
                return BadRequest("Empty payload");

            var connection = _connectionPool.Connection; //get a connection
            var notificationGroups = databaseNotifications.Batch(_configuration.BatchSize, true);
            foreach (var notificationGroup in notificationGroups)

            foreach (var notification in notificationGroup)
            {
                var subject =
                    $"{_configuration.NatsSubject}.{customer.ToLower()}.{notification.Table?.ToLower()}.{notification.Action}";
                notification.Customer = customer;
                var serializedData =
                    JsonSerializer.Serialize(notification, _configuration.JsonSerializerOptions);
                var payload = Encoding.UTF8.GetBytes(serializedData);

                for (var i = 0; i < 3; i++)
                    try
                    {
                        connection.Publish(subject, payload);
                        break;
                    }
                    // ReSharper disable once CatchAllClause
                    catch (Exception e)
                    {
                        if (i < 2)
                            continue;
                        LogError(customer,
                            $"Fail to publishing for 3 times: [{serializedData}], error: {e.GetBaseException().Message}");
                    }
            }

            connection.FlushBuffer();
            return Ok();
        }

And the configuration

{
  "ConnectionStrings": {
    "NATSServers": [
      "nats://0.0.0.178:4222,nats://0.0.0.179:4222,nats://0.0.0.180:4222",
      "nats://0.0.0.179:4222,nats://0.0.0.180:4222,nats://0.0.0.178:4222",
      "nats://0.0.0.180:4222,nats://0.0.0.178:4222,nats://0.0.0.179:4222"
    ] //config like this to support round robin logic
  },
  "Kestrel": {
    "Limits": {
      "MaxConcurrentConnections": null,
      "MaxConcurrentUpgradedConnections": null,
      "MaxRequestBufferSize": null,
      "MaxRequestHeaderCount": 4096,
      "MaxRequestHeadersTotalSize": 28672,
      "MaxRequestBodySize": null,
      "MaxResponseBufferSize": null,
      "AddServerHeader": false
    },
    "EndPoints": {
      "Http": {
        "Url": "http://*:7500"
      }
    }
  },
  "Settings": {
    "Subject": "subject to send",
    "Batch": 5,
    "NatsConnectionPoolSize": 2, // There will be a total of ([configValue] * ConnectionStrings:NATSServers connections
    "UseConnectionPool":
      true //to use connection pool or a single connection, in both cases, the connection pool will always have connection(s) inside
  }
}

Please hep me to make a try with this setup :)
I want this api really fast, because it will be called from sql service broker, it may hurt the sql server if the response time is too long

[watfordgnf]

Your response times are not apples to apples in these cases. In your Task.Run example the created task is not awaited and the Ok response is returned as soon as the thread is created. In @ColinSullivan1's example code the Ok response is returned as soon as all of the notifications are sent. You're comparing the time it takes to spawn a task with the time it takes to send all of the messages.

In any case, Publish is the lowest overhead call in the C# NATS client, so calling it in a loop should be the best performance for a single connection. I'd pull the code out into a console application to benchmark the timing for serialization and conversion to UTF-8 byte arrays as that overhead is non-trivial compared to the time to send a NATS message.

Two notes:

1. I would not Publish in parallel using the same connection instance, you'll slow yourself down with per-connection locking. You could use Publish in parallel using different connection instances, as they won't contend for the same locks.
2. If you want to merely return upon "accepting" the messages, but not publishing, you should have a background service that handles this work. Each request would simply enqueue the messages for the background service.

Given IMessageQueue which is a singleton:

    [HttpPost("notification")]
    public ActionResult PublishMessage(
        [FromServices] IMessageQueue messageQueue,
        [FromBody] IEnumerable<Notification> payload)
    {
        if (payload == null)
            return BadRequest("Empty payload");
        
        foreach (var notification in payload)
        {
            await messageQueue.EnqueueAsync(CreateMessage(notification));
        }

        return Ok();
    }

With a background service like:

public class MessageSenderBackgroundService : BackgroundService
{
    private readonly IMessageQueue _messageQueue;

    public MessageSenderBackgroundService(IMessageQueue messageQueue)
    {
        _messageQueue = messageQueue;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        using (var connection = /* create or inject your NATS connection */)
        {
            while (!token.IsCancellationRequested)
            {
                Message message = await _messageQueue.DequeueAsync(stoppingToken);
                connection.Publish(message.Subject, message.Payload);
                // pick some heuristic to flush if the default isn't reasonable
            }
        }
    }
}

[ColinSullivan1]

We'll also want to check that the test isn't running up against network bandwidth limits.

How big is a request in bytes, and how many bytes are the NATS message payloads? Also, what are the network links you are are testing this over between the HTTP client, HTTP server, and NATS server? (e.g. 10Mbs wifi, 100Mbs ether, etc.)

[haiduong87]

Thanks you,

I will try @watfordgnf's suggestion and feedback the result.

@ColinSullivan1

• The request contains up to 40 notifications in Json format "[{},{},{}...]"
• Each notification when sending to this Api contains 3 fields like this entity (only one date is sent)

public class DatabaseNotification
    {
        [JsonPropertyName("id")] public long? Id { get; set; }

        [JsonPropertyName("table")] public string Table { get; set; }

        [JsonPropertyName("createdDate")] public DateTime? CreatedDate { get; set; }

        [JsonPropertyName("modifiedDate")] public DateTime? ModifiedDate { get; set; }

        [JsonPropertyName("deletedDate")] public DateTime? DeletedDate { get; set; }
    }

• And the test environment
  -- HttpClient and API server are on my local machine
  -- 3 NATS servers are on my vpn network
  -- About the network links I don't think they matter.
  ---- Because what I'm doing is "re-writing an ActixWeb application - rust with Asp.NET Core"
  ---- I do all the stress tests with that ActixWeb app (on my local machine too), and the result is very good (under 200ms for all the cases)
  ---- With Asp.Net Core I just want it's about 500~700ms

[watfordgnf]

@haiduong87 so the ActixWeb setup used nats.rs as its NATS client? I'd be curious to know whether it called Publish in a blocking loop or if it put it into a background queue and returned immediately.

[haiduong87]

@watfordgnf
Thank you, I end up using System.Threading.Channel for now

Controller:

[HttpPost("{customer}")]
        public IActionResult PublishNotificationAsync([FromRoute][Required] string customer,
            [FromBody] IEnumerable<DatabaseNotification> databaseNotifications)
        {
            if (databaseNotifications == null)
                return BadRequest("Empty payload");

            foreach (var notification in databaseNotifications)
            {
                notification.Customer = customer;
                _notificationChannelWriter.WriteAsync(notification);
            }

            return Ok();
        }

Worker class:

public class NatsWorker : BackgroundService
    {
        private readonly Configuration _configuration;
        private readonly NatsConnectionPool _connectionPool;
        private readonly ILogger _logger;

        private readonly ChannelReader<DatabaseNotification> _notificationChannelReader;

        public NatsWorker(Channel<DatabaseNotification> notificationChannel, Configuration configuration,
            ILogger<NatsWorker> logger, NatsConnectionPool connectionPool)
        {
            _notificationChannelReader = notificationChannel;
            _logger = logger;
            _configuration = configuration;
            _connectionPool = connectionPool;
        }

        protected override async Task ExecuteAsync(CancellationToken stoppingToken)
        {
            await foreach (var notification in _notificationChannelReader.ReadAllAsync(stoppingToken)
                .ConfigureAwait(false))
            {
                PushNotification(notification);
            }
        }

        private void PushNotification(DatabaseNotification notification)
        {
            var subject =
                $"{_configuration.NatsSubject}.{notification.Customer.ToLower()}.{notification.Table?.ToLower()}.{notification.Action}";

            var payload = Encoding.UTF8.GetBytes(notification.SerializedString);

            for (var i = 0; i < 3; i++)
                try
                {
                    _connectionPool.Connection.Publish(subject, payload);
                    _configuration.LogObject.Success(notification.Customer);
                    break;
                }
                // ReSharper disable once CatchAllClause
                catch (Exception e)
                {
                    if (i < 2)
                        continue;
                    LogError(notification.Customer,
                        $"Fail to publishing for 3 times: [{notification.SerializedString}], error: {e.GetBaseException().Message}");
                    _configuration.LogObject.Fail(notification.Customer);
                }
        }

        private void LogError(string customer, string message)
        {
            using (_logger.BeginScope(new Dictionary<string, object>
            {
                ["customer"] = customer
            }))
            {
                _logger.LogError(message);
            }
        }
    }

The response time is very fast, it's about 300~400 ms max.
I change from using JsonSerializer to my own string method, to reduce the allocation of JsonWriter (because in this case the object is so simple so I can do that)
I remove all the Task...
I can achieve my goal (1000 requests, each request up to 40 items)

But about the nats client the publish process, can you look into my code and give me advice?
Because the background worker can't send that 40k items in 1 second. I'm using the connection pool (in the above comment)

About the Actix WebApp.
I see it publishes and returns response, no background service...

[watfordgnf]

That looks more like what we would recommend. I'm not certain where the holdup is as we can easily push 100k 1KB messages/sec using just Publish. It would be interesting to see profiling data to find out where the holdup is.

[israellot]

I believe this is true for a single publisher thread.
Publishing from multiple threads hits contention in write locks. I dare say this explains why his use of a connection pool improves the numbers. On his scenario, firing 1000 requests at aspnet there will be as many threads as processor cores competing from a single lock in the publish function of NATS client. Having multiple NATS clients reduces that contention.

[haiduong87]

@watfordgnf @ColinSullivan1
I've uploaded my latest working here: https://github.com/haiduong87/NotificationServer

Please help me to check my using of NatsClient.

Thank you!