Significant Performance Degradation When Using `EnablePublishParallelSend`

[KrawMire]

Hello! First of all, thank you for this package :)

I am using CAP with Kafka as transport and MongoDB for storage. I have noticed the significant performance degradation when EnablePublishParallelSend is enabled. Unfortunately, I can't provide detailed metrics about it, but I suspect, that if there are really lots of messages, then this approach is not optimal (IDispatcher.Default file):

if (_enableParallelSend)
{
    var tasks = new List<Task>();
    var batchSize = _pChannelSize / 50;
    for (var i = 0; i < batchSize && _publishedChannel.Reader.TryRead(out var message); I++)
    {
        var item = message;
        tasks.Add(Task.Run(async () =>
        {
            try
            {
                var result = await _sender.SendAsync(item).ConfigureAwait(false);
                if (!result.Succeeded) _logger.MessagePublishException(item.Origin.GetId(), result.ToString(), result.Exception);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, $"An exception occurred when sending a message to the transport. Id:{message.DbId}");
            }
        }));
    }

    await Task.WhenAll(tasks);
}

For example, if we have at least 200 RPS and each request produces a message to the broker, then we will create a new Task for each of this operation. I suppose it can really degrade performance.

[KrawMire]

I have captured some profiling snapshots (using dotTrace). Hope this can help to determine the problem.

Tests were performed with the following code:

public sealed class CapMessagePublisher
{
    private int _count;
    private readonly CancellationTokenSource _cts;
    private readonly ICapPublisher _publisher;
    private readonly ILogger<CapMessagePublisher> _logger;

    public CapMessagePublisher(ICapPublisher publisher, ILogger<CapMessagePublisher> logger)
    {
        _publisher = publisher;
        _logger = logger;

        _cts = new CancellationTokenSource();
        var th = new Thread(() => LogMessagesPerSecond(_cts.Token));
        th.Start();
    }

    public async Task TestPublishAsync()
    {
        _count = 0;
        var tasks = new List<Task>();

        for (var i = 0; i < 50; i++)
        {
            var i1 = i;
            tasks.Add(Task.Run(async () =>
            {
                for (var j = 0; j < 5000; j++)
                {
                    var message = new
                    {
                        msg = "hello, test message",
                    };

                    await _publisher.PublishAsync("test.messages", message);

                    Interlocked.Increment(ref _count);
                }
            }));
        }

        await Task.WhenAll(tasks);
        await _cts.CancelAsync();
    }

    private void LogMessagesPerSecond(CancellationToken cancellationToken)
    {
        while (!cancellationToken.IsCancellationRequested)
        {
            Thread.Sleep(1000);
            var messageCount = Interlocked.Exchange(ref _count, 0);
            _logger.LogInformation("Messages sent per second: {MessageCount}", messageCount);
        }
    }
}

---

First snapshot is made with EnablePublishParallelSend = false:

---

The second one is made with EnablePublishParallelSend = true:

[KrawMire]

I have found a potential workaround - using delayed messages with small delay (about 50ms). In this case, control is returned to the caller almost immediately and it allows to handle more RPS. But there is a problem: delayed messages are sent one-by-one and it slows down the sending messages speed.

I have created a PR (#1753) to enable parallel sending not only regular messages, but also the delayed ones. They are added to the publishing channel first, and then processed by the Sending() method.

[yang-xiaodong]

Thank you for raising this concern. I've conducted performance testing using wrk to simulate real-world message publishing scenarios, and I'd like to share some findings:

Performance Testing Results

After testing with EnablePublishParallelSend enabled vs. disabled, I observed:

1. No performance degradation when parallel send is enabled
2. Modest performance improvement - but not as significant as one might expect
3. Database I/O is the primary bottleneck - the majority of execution time is spent persisting messages to the database
   Current Implementation Analysis
   The current batching approach (50 messages per batch) does introduce some overhead:
   • Thread pool pressure from Task allocation
   • Increased GC activity from closure allocations
   However, these impacts are relatively minor compared to the database I/O overhead. In typical production scenarios:

Time distribution per message:
- Serialization: ~0.2ms (0.6%)
- Broker send: ~3ms (9%)
- Database UPDATE: ~30ms (90.4%)

I suggest keeping the current implementation as-is. For users experiencing performance issues, the focus should be on addressing the database bottleneck rather than optimizing the parallel send mechanism. The current approach provides a reasonable balance between code simplicity and performance.

I can merge PR (#1753), which helps us to be consistent in the processing behavior of delayed messages and non-delayed messages.