RabbitMQ Producer C# .NET Core 5.0 Memory Leak

Question

I was wondering if someone can please help with the following situation:

I cannot solve a memory leak with a RabbitMQ Publisher written in C# and using .Net core 5.0.

This is the csproj file :

   
  
    Exe
    net5.0
    win-x64
  
  ...

I have a .net console application running inside a virtual machine which connects to a server through an API (a registered 64 bit dll and referenced as a COM reference), gets information from that server and then tries to publish this information to a RabbitMQ machines located on an AWS cloud (a load balancer with several nodes for this RMQ instance).

Accessing the API in the code is done in the following way :

        private void SetUpApi () {
            Utils.log.Info (_api.SetAPIOptions (""));
            _api.OnServerConnect += OnServerConnect;
            _api.OnServerDisconnect += OnServerDisconnect;
            _api.OnNewData += OnNewData;            
        }

        private void OnNewData(string strXML){
            try{
                if (strXML.Contains("


During peak business hours I get around 400 - 500 messages/second from the API. These messages come in the form of XML messages. For example, a message can contain several orders as shown in the example below. One  element can contain an action to insert (to create) an Order and another to remove a certain Order.

  
  
    
    ...
  
  
    
    ...
  

The RabbitMQ server is configured and I connect to it using SSL (with a certificate and a key for it).
I use to connect to the RMQ the RabbitMQ.Client v6.2.1.
The exchange, the queues and the bindings are already defined in RabbitMQ. My Producer application only connects to it and starts publishing.

It is important that I use a synchronous publishing method as the order of messages we get is very important. For example in one message we get an action to create an Order and another message which comes immediately after which is telling to remove the same Order. If I would use an async method to publish to RMQ I would be possibly get the removal action before the insert action.
  
  
  
      ...
  

The removal messages:
  
  
  
      ...
  

I use the following method for publishing to RMQ: the object pool (Microsoft provides a package named Microsoft.Extensions.ObjectPool) - method described in here - https://www.c-sharpcorner.com/article/publishing-rabbitmq-message-in-asp-net-core/ .
I'm using here the following code:
class RabbitManager : IRabbitManager
{
    private readonly DefaultObjectPool _objectPool;
    public RabbitManager(IPooledObjectPolicy objectPolicy){
        _objectPool = new DefaultObjectPool(objectPolicy, Environment.ProcessorCount * 2);
    }

    public void Publish(T message, string exchangeName, string exchangeType, string routeKey) where T : class {
        if (message == null)
            return;

        var channel = _objectPool.Get();
        try{
            var sendBytes = Encoding.UTF8.GetBytes(message.ToString());
            var properties = channel.CreateBasicProperties();
            properties.ContentType = "application/json";
            properties.DeliveryMode = 1; // Doesn't persist to disk
            properties.Timestamp = new AmqpTimestamp(DateTimeOffset.UtcNow.ToUnixTimeMilliseconds());

            channel.BasicPublish(exchangeName, routeKey, properties, sendBytes);
        }
        catch (Exception ex) {
            throw ex;
        }
        finally {
            _objectPool.Return(channel);
        }
    }
}

public class RabbitModelPooledObjectPolicy : IPooledObjectPolicy
{
    private readonly RabbitOptions _options;
    private readonly IConnection _connection;

    public RabbitModelPooledObjectPolicy(RabbitOptions _options){
        this._options = _options;
        _connection = GetConnection();
    }

    private IConnection GetConnection() {
        var factory = new ConnectionFactory() {
            HostName = _options.HostName,
            UserName = _options.UserName,
            Password = _options.Password,
            //Port = _options.Port,
            VirtualHost = _options.VHost,
        };

        if (!String.IsNullOrEmpty(_options.CertPath))
        {
            factory.RequestedConnectionTimeout = TimeSpan.FromMilliseconds(5000);
            factory.Ssl.AcceptablePolicyErrors = SslPolicyErrors.RemoteCertificateNameMismatch | SslPolicyErrors.RemoteCertificateChainErrors;
            factory.Ssl.CertificateValidationCallback += new RemoteCertificateValidationCallback(ValidateServerCertificate);
            factory.Ssl.ServerName = _options.HostName;
            factory.Ssl.CertPath = _options.CertPath;
            factory.Ssl.CertPassphrase = _options.CertPass;
            factory.Ssl.Version = SslProtocols.Tls12;
            factory.Ssl.Enabled = true;
        }

        factory.RequestedHeartbeat = TimeSpan.FromSeconds(1);
        factory.AutomaticRecoveryEnabled = true;        // enable automatic connection recovery
        factory.RequestedChannelMax = 32;

        var _connection = factory.CreateConnection();
        _connection.ConnectionShutdown += Connection_ConnectionShutdown;

        return _connection;
    }

    private void Connection_ConnectionShutdown(object sender, ShutdownEventArgs e){
        Utils.log.Info("Connection broke!");
    }

    private bool ValidateServerCertificate(object sender, X509Certificate certificate, X509Chain chain, SslPolicyErrors sslPolicyErrors){
        return true;
    }

    public IModel Create(){
        return _connection.CreateModel();
    }

    public bool Return(IModel obj) {
        if (obj.IsOpen) {
            return true;
        }
        else {
            obj?.Dispose();
            return false;
        }
    }
}

Below is a screenshot with the problem - the constant memory inscrease :

This is the stack trace of a memory snapshot taken just after the above screenshot :

Just after the screenshot above was taken I got the following error message in my program, in the console :
26-04-2021 10:41:48 - OnNewData () RabbitMQ.Client.Exceptions.AlreadyClosedException: Already closed: The AMQP operation was interrupted: AMQP close-reason, initiated by Library, code=0, text='End of stream', classId=0, methodId=0, cause=System.IO.EndOfStreamException: Reached the end of the stream. Possible authentication failure.
   at RabbitMQ.Client.Impl.InboundFrame.ReadFrom(Stream reader, Byte[] frameHeaderBuffer)
   at RabbitMQ.Client.Framing.Impl.Connection.MainLoopIteration()
   at RabbitMQ.Client.Framing.Impl.Connection.MainLoop()
   at RabbitMQ.Client.Framing.Impl.Connection.EnsureIsOpen()
   at RabbitMQ.Client.Framing.Impl.AutorecoveringConnection.CreateModel()
   at RabbitModelPooledObjectPolicy.Create() in C:\Users\user\Desktop\Projects\ConsoleCoreApp1\RabbitMQ\RabbitModelPooledObjectPolicy.cs:line 77
   at Microsoft.Extensions.ObjectPool.DefaultObjectPool`1.Create()
   at Microsoft.Extensions.ObjectPool.DefaultObjectPool`1.Get()
   at RabbitManager.Publish[T](T message, String exchangeName, String exchangeType, String routeKey) in C:\Users\user\Desktop\Projects\ConsoleCoreApp1\RabbitMQ\RabbitManager.cs:line 32
   at ConsoleApp1.Service1.SendOrderToRMQ(JObject order) in C:\Users\user\Desktop\Projects\ConsoleCoreApp1\Service1.cs:line 411
   at ConsoleApp1.Service1.ParseXMLAnswer(String strOutputXML, String caller) in C:\Users\user\Desktop\Projects\ConsoleCoreApp1\Service1.cs:line 372
   at ConsoleApp1.Service1.OnNewData(String strXML) in C:\Users\user\Desktop\Projects\ConsoleCoreApp1\Service1.cs:line 348
26-04-2021 10:41:48 - OnNewData () RabbitMQ.Client.Exceptions.AlreadyClosedException: Already closed: The AMQP operation was interrupted: AMQP close-reason, initiated by Library, code=0, text='End of stream', classId=0, methodId=0, cause=System.IO.EndOfStreamException: Reached the end of the stream. Possible authentication failure.
   at RabbitMQ.Client.Impl.InboundFrame.ReadFrom(Stream reader, Byte[] frameHeaderBuffer)
   at RabbitMQ.Client.Framing.Impl.Connection.MainLoopIteration()
   at RabbitMQ.Client.Framing.Impl.Connection.MainLoop()
   at RabbitMQ.Client.Framing.Impl.Connection.EnsureIsOpen()
   at RabbitMQ.Client.Framing.Impl.AutorecoveringConnection.CreateModel()
   at RabbitModelPooledObjectPolicy.Create() in C:\Users\user\Desktop\Projects\ConsoleCoreApp1\RabbitMQ\RabbitModelPooledObjectPolicy.cs:line 77
   at Microsoft.Extensions.ObjectPool.DefaultObjectPool`1.Create()
   at Microsoft.Extensions.ObjectPool.DefaultObjectPool`1.Get()
   at RabbitManager.Publish[T](T message, String exchangeName, String exchangeType, String routeKey) in C:\Users\user\Desktop\Projects\ConsoleCoreApp1\RabbitMQ\RabbitManager.cs:line 32
   at ConsoleApp1.Service1.SendOrderToRMQ(JObject order) in C:\Users\user\Desktop\Projects\ConsoleCoreApp1\Service1.cs:line 411
   at ConsoleApp1.Service1.ParseXMLAnswer(String strOutputXML, String caller) in C:\Users\user\Desktop\Projects\ConsoleCoreApp1\Service1.cs:line 372
   at ConsoleApp1.Service1.OnNewData(String strXML) in C:\Users\user\Desktop\Projects\ConsoleCoreApp1\Service1.cs:line 348


And this leads to the final crash of the program :

What I did to prevent the memory leak :

The classes have the IDisposable interfaces enabled

        public void Dispose()
        {
            Dispose(true);
            GC.SuppressFinalize(this);
        }

        protected virtual void Dispose(bool disposing)
        {
            if (disposing)
            {
                // free managed resources
                _onTimePerHour.Dispose();
                _api.OnNewData -= OnNewData;
            }
            // free native resources if there are any.
        }


Forced garbage collection after each message received :

private void ParseXMLAnswer(string strOutputXML, string caller) {
            ...
            doc = null;
            GC.Collect();
            GC.WaitForPendingFinalizers();
}

This helps a bit more and now the memory problem i have increases after a longer period of time.

I use the ReShaper (enter link description here) plugin for Visual Studio to give me a better understanding of the Stack Trace of the memory problem but it does not help much.


What do I think the problem is :
The RabbitMQ Producer app gets many messages per second which are then parsed, split into several JSON messages and send to RMQ using the same channel. The following cases might occur :

I'm publishing on a single RMQ channel and somehow i should use several channels (one connection but multiple channels)
I'm receiving more messages than I can parse and send through RMQ using the RabbitMQ.Client .net library
I'm holding up some references in memory for some objects (maybe messages) which do not get freed up;

Did someone had this problem before ? Because I cannot find anywhere any info on this "SingleProducerSingleConsumerQueue+Segment out of memory" issue.
Does someone know how to analyse more in depth this problem ?
Big thanks !

Edit 1
I guess more info is needed to solve this memory issue.
I have several consumers which consume data from RabbitMQ (like NodeJS and python apps). Thus, I need to design the RabbitMQ Producer in a generic way, as each consumer needs different data. And I cannot modify and restart my RabbitMQ Producer each time I have a new consumer app. So I need to publish my messages in a generic way.
For example, each consumer has its own dedicated queue, with dedicated bindings. Let's say I have consumer1 with queue cons1 and bindings :

marketName.productName.*.1 (the productName would correspond for days).

This binding is dynamic and for now it corresponds to Monday (04.April) but tomorrow it will correspond to Tueday (05.April).
So I need to store in memory the marketNames and productNames by using
private static read-only Dictionary> sequenceItemsHashed = new Dictionary>();
private static readonly Dictionary sequencesFromInstruments = new Dictionary();

I mention that sequenceItemsHashed corrrespond to marketNames and sequencesFromInstruments to productNames in my logic.
This way I send all the messages to RMQ and I do the sorting in RMQ afterwards using bindings.

Edit 2
From what I understood, in order to solve my question I need something like the following architecture (enter link description here) :

So multiple threads in my single connection to the RMQ server and one channel per thread.

Edit 3
Implemented successfully the pipelines, the ConcurrentQueue and the Consumer Thread which pushes to RMQ but still have memory issues:
private readonly TransformBlock orderFilter;
private readonly TransformBlock xmlParser;
//private readonly TransformBlock xmlToJsonTransformer;
private readonly TransformManyBlock jsonOrderFactory;
private readonly ActionBlock messageSender;

ConcurrentQueue concurrentQueue = new ConcurrentQueue();

public Service1 (string [] args) {
    ...
    // setup pipeline blocks
    orderFilter = new TransformBlock(FilterIncomingMessages);
    xmlParser = new TransformBlock(ParseXml);
    jsonOrderFactory = new TransformManyBlock(CreateOrderMessages);
    messageSender = new ActionBlock(SendMessage);

    // build your pipeline            
    orderFilter.LinkTo(xmlParser, x => !string.IsNullOrEmpty(x));
    orderFilter.LinkTo(DataflowBlock.NullTarget()); // for non-order msgs

    xmlParser.LinkTo(jsonOrderFactory);
    jsonOrderFactory.LinkTo(messageSender, new DataflowLinkOptions { PropagateCompletion = true });

    Task t2 = Task.Factory.StartNew(() =>
            {
                while (true) { 
                    if (!concurrentQueue.IsEmpty)
                    {
                        JToken number;
                        while (concurrentQueue.TryDequeue(out number))
                        {
                            _rabbitMQ.PublishMessages(
                                Encoding.ASCII.GetBytes(number.ToString()),
                                "test"
                            );
                        }
                    } else
                    {
                        Thread.Sleep(1);
                    }
                }
            });        
     ...
}

private string FilterIncomingMessages(string strXml){
    if (strXml.Contains(" CreateOrderMessages(JObject o){
    List myList = new List();
            if (o.ContainsKey("GV8APIDATA")){
                if (o["GV8APIDATA"]["ORDER"].Type is JTokenType.Object){
                    JToken order = o["GV8APIDATA"]["ORDER"];
                    myList.Add(order);
                }
                else if (o["GV8APIDATA"]["ORDER"].Type is JTokenType.Array){
                    JToken orders = o["GV8APIDATA"]["ORDER"];
                    foreach (var order in orders.Children()){
                        myList.Add(order);
                    }
                }
            }
            return myList.ToArray ();
        }

private void SendMessage(JToken order){
    concurrentQueue.Enqueue(order);
}

The new solution helps to break the logic into several small parts but I still have a constant memory increase.


Edit 4
Taking into account @Fildor's answer I did the following :
Instead of converting strings containing xml with  elements to JSON, I deserialize XML to objects using the pipelines and the code below.
I removed the part with the Thread and the ConcurrentQueue and I'm publishing directly in the last ActionBlock.
This solves my memory leak problem, but there are other problems like :

If the messages are big enough I will only be able to print around 120 messages / second. I get the rate of 1780 messages/s if I just print the simple string "test".


public Service1 (string [] args) {
            ...             
            // setup pipeline blocks
            orderFilter = new TransformBlock(FilterIncomingMessages);
            xmlParser = new TransformBlock(ParseXml);
            jsonOrderFactory = new TransformManyBlock(CreateOrderMessages);
            messageSender = new ActionBlock(SendMessage);

            // build your pipeline            
            orderFilter.LinkTo(xmlParser, x => !string.IsNullOrEmpty(x));
            orderFilter.LinkTo(DataflowBlock.NullTarget()); // for non-order msgs
            xmlParser.LinkTo(jsonOrderFactory);
            jsonOrderFactory.LinkTo(messageSender, new DataflowLinkOptions { PropagateCompletion = true });

            RunAsConsole(args);
        }

        private readonly TransformBlock orderFilter;
        private readonly TransformBlock xmlParser;
        private readonly TransformManyBlock jsonOrderFactory;
        private readonly ActionBlock messageSender;

        private void OnNewData(string strXML){
            orderFilter.Post(strXML); 
        }        

        private string FilterIncomingMessages(string strXml){
            if (strXml.Contains(" CreateOrderMessages(OrdersResponse o){
            return o.orders;
        }

        private void SendMessage(Order order) {
            _rabbitMQ.PublishMessages(
                    Encoding.ASCII.GetBytes(order.ToString()),
                    "test"
                );
        }

And the ORDER object looking like :
    [Serializable()]
    [XmlRoot (ElementName = "ORDER")]
    public class Order : IDisposable {

        public void Dispose()
        {
            EngineID = null;
            PersistentOrderID = null;
            ...
            InstrumentSpecifier.Dispose();
            InstrumentSpecifier = null;
            GC.SuppressFinalize(this);
        }

        [XmlAttribute (AttributeName = "EngineID")]
        public string EngineID { get; set; }
        [XmlAttribute (AttributeName = "PersistentOrderID")]
        public string PersistentOrderID { get; set; }
        ... 
        [XmlElement(ElementName = "INSTSPECIFIER")]
        public InstrumentSpecifier InstrumentSpecifier { get; set; }
    }

And my new RabbitMQ class :
public class RMQ : IDisposable {
    private IConnection _connection;
    public IModel Channel { get; private set; }        
    private readonly ConnectionFactory _connectionFactory;
    private readonly string _exchangeName;

    public RMQ (RabbitOptions _rabbitOptions){
        try{
            // _connectionFactory initialization
            _connectionFactory = new ConnectionFactory()
            {
                HostName = _rabbitOptions.HostName,
                UserName = _rabbitOptions.UserName,
                Password = _rabbitOptions.Password,
                VirtualHost = _rabbitOptions.VHost,
            };
            this._exchangeName = _rabbitOptions.ExchangeName;

            if (!String.IsNullOrEmpty(_rabbitOptions.CertPath)){
                _connectionFactory.RequestedConnectionTimeout = TimeSpan.FromMilliseconds(5000);
                _connectionFactory.Ssl.AcceptablePolicyErrors = SslPolicyErrors.RemoteCertificateNameMismatch | SslPolicyErrors.RemoteCertificateChainErrors;
                _connectionFactory.Ssl.CertificateValidationCallback += new RemoteCertificateValidationCallback(ValidateServerCertificate);
                _connectionFactory.Ssl.ServerName = _rabbitOptions.HostName;
                _connectionFactory.Ssl.CertPath = _rabbitOptions.CertPath;
                _connectionFactory.Ssl.CertPassphrase = _rabbitOptions.CertPass;
                _connectionFactory.Ssl.Version = SslProtocols.Tls12;
                _connectionFactory.Ssl.Enabled = true;
            }

            _connectionFactory.RequestedHeartbeat = TimeSpan.FromSeconds(1);
            _connectionFactory.AutomaticRecoveryEnabled = true;        // enable automatic connection recovery
            //_connectionFactory.RequestedChannelMax = 10;

            if (_connection == null || _connection.IsOpen == false){
                _connection = _connectionFactory.CreateConnection();
                _connection.ConnectionShutdown += Connection_ConnectionShutdown;
            }
            if (Channel == null || Channel.IsOpen == false){
                Channel = _connection.CreateModel();
            }
            Utils.log.Info("ConnectToRabbitMQ () Connecting to RabbitMQ. rabbitMQenvironment = ");
        }
        catch (Exception ex){
            Utils.log.Error("Connection to RabbitMQ failed ! HostName = " + _rabbitOptions.HostName + " VirtualHost = " + _rabbitOptions.VHost);
            Utils.printException("ConnectToRMQ ()", ex);
        }
    }        

    private void Connection_ConnectionShutdown(object sender, ShutdownEventArgs e){
        Utils.log.Info ("Connection broke!");
        try{
            if (ReconnectToRMQ()){
                Utils.log.Info("Connected!");
            }
        }
        catch (Exception ex){
            Utils.log.Info("Connect failed!" + ex.Message);
        }
    }

    private bool ReconnectToRMQ(){
        if (_connection == null || _connection.IsOpen == false){
            _connection = _connectionFactory.CreateConnection();
            _connection.ConnectionShutdown += Connection_ConnectionShutdown;                
        }

        if (Channel == null || Channel.IsOpen == false){
            Channel = _connection.CreateModel();
            return true;
        }
        return false;
    }

    private bool ValidateServerCertificate (object sender, X509Certificate certificate, X509Chain chain, SslPolicyErrors sslPolicyErrors) {
        return true;
    }

    public void DisconnectFromRMQ () {
        Channel.Close ();
        _connection.Close ();
    }     

    public void Dispose(){
        try{
            Channel?.Close();
            Channel?.Dispose();
            Channel = null;

            _connection?.Close();
            _connection?.Dispose();
            _connection = null;
        }
        catch (Exception e){
            Utils.log.Error("Cannot dispose RabbitMQ channel or connection" + e.Message);
        }
    }

    public void PublishMessages (byte [] message, string routingKey) {            
        if (this._connection == null || ! _connection.IsOpen) {
            Utils.log.Error ("PublishMessages(), Connect failed! this.conn == null || !conn.IsOpen ");
            ReconnectToRMQ();
        } else { 
            var properties = Channel.CreateBasicProperties();
            properties.Persistent = true;

            Channel.BasicPublish (_exchangeName, routingKey, properties, message);
            //serviceInstance1.Publish(message, _rabbitOptions.ExchangeName, "", routingKey);
        }
    }
}

Now the funny thing is if I publish only a small string like "test" to RabbitMQ to my predefined queue I can publish more than 1780 messages/second.

user_982149 · Answer

First, it seems you are clogging the event handling thread. So, what I'd do is decouple event handling from the actual processing:

( Untested! Just an outline!)

REMOVED FAULTY CODE

Then in serviceInstance1, I would have Publish enqueue the orders in a BlockingCollection, on which a dedicated Thread is waiting. That thread will do the actual send. So you'll marshall the orders to that thread regardless of what you chose to do in Processor and all will be decoupled and in-order.

You probably will want to set BlockOptions according to your requirements.

Mind that this is just a coarse outline, not a complete solution. You may also want to go from there and minimize string-operations etc.

EDIT

Some more thoughts that came to me since yesterday in no particular order:

May it be beneficial to ditch the first filter to filter out empty sets of JObjects later?
Maybe it's worth trying to use System.Text.Json instead of Newtonsoft?
Is there a more efficient way to get from xml to json? (I was thinking "XSLT" but really not sure)
- convert xml to json using xslt
- xml to json using xslt
I'd recommend to rig up a Benchmark.Net with MemoryAnalyzer to document / proove your changes have positive effects.
Don't forget to have a look into DataFlowBockOptions to tweak the pipeline's behavior.

For reference:

In response to EDIT 3 in the question:

Task t2 = Task.Factory.StartNew(() =>
            {
                while (true) { 
                    if (!concurrentQueue.IsEmpty)
                    {
                        JToken number;
                        while (concurrentQueue.TryDequeue(out number))
                        {
                            _rabbitMQ.PublishMessages(
                                Encoding.ASCII.GetBytes(number.ToString()),
                                "test"
                            );
                        }
                    } else
                    {
                        Thread.Sleep(1);
                    }
                }
            });

is really not a good idea.

First off, I would do that in the service class that handles sending (serviceInstance1 - don't know the type). Then, you are doing a tight loop with spin-wait while mixing TPL with Thread.Sleep. That are 2 NoNos. It also completely messes the intention of the blocking queue. Which is: the thread blocks until an item is available on that queue.

Maybe it's a better idea to drop this part for now, completely and just have the final block in the pipeline do the serviceInstance1.Publish. That may have been premature optimization.

EDIT 2

So, yesterday I did some experimenting and found that this:

using Microsoft.Extensions.Logging;
using System;
using System.Collections.Generic;
using System.IO;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
using System.Xml.Linq;
using System.Xml.Serialization;
using Newtonsoft.Json;
using System.Linq;

namespace DataFlowExperiment.PipelinesLib
{

    public class PipelineOne
    {
        private readonly IPipelineOneSteps steps;

        private readonly TransformBlock startBlock; // XML deserialize to Model
        private readonly TransformManyBlock toJsonMessagesBlock; // jsons generieren.
        private readonly ITargetBlock resultCallback;

        public PipelineOne(IPipelineOneSteps steps, ITargetBlock resultCallback = null)
        {
            this.steps = steps;

            startBlock = new TransformBlock(steps.Start);
            toJsonMessagesBlock = new TransformManyBlock(steps.ToJson);

            this.resultCallback = resultCallback ?? DataflowBlock.NullTarget();

            startBlock.LinkTo(toJsonMessagesBlock, new DataflowLinkOptions { PropagateCompletion = true });
            toJsonMessagesBlock.LinkTo(this.resultCallback, new DataflowLinkOptions { PropagateCompletion = true }, x => !string.IsNullOrEmpty(x));
            toJsonMessagesBlock.LinkTo(DataflowBlock.NullTarget(), new DataflowLinkOptions { PropagateCompletion = true });
        }

        public void Post(string input)
        {
            startBlock.Post(input);
        }

        public Task Close()
        {
            startBlock.Complete();
            return resultCallback.Completion;
        }
    }

    public interface IPipelineOneSteps
    {
        public XDocument Start(string input);
        public IEnumerable ToJson(XDocument doc);
    }

    public class PipelineOneSteps : IPipelineOneSteps
    {
        private readonly JsonSerializer jsonSerializer;

        public PipelineOneSteps()
        {
            jsonSerializer = JsonSerializer.CreateDefault();
        }

        public XDocument Start(string input)
        {
            XDocument doc = XDocument.Parse(input);
            return doc;
        }

        public IEnumerable ToJson(XDocument doc)
        {
            XNamespace ns = "api-com";
            var orders = doc.Root.Elements(ns + "ORDER");

            foreach (var order in orders)
            {
                yield return JsonConvert.SerializeXNode(order);
            }
        }
    }
}

results in this benchmark:


BenchmarkDotNet=v0.12.1, OS=Windows 10.0.19041.867 (2004/?/20H1)
Intel Core i9-10885H CPU 2.40GHz, 1 CPU, 16 logical and 8 physical cores
.NET Core SDK=5.0.202
  [Host]     : .NET Core 3.1.14 (CoreCLR 4.700.21.16201, CoreFX 4.700.21.16208), X64 RyuJIT
  DefaultJob : .NET Core 3.1.14 (CoreCLR 4.700.21.16201, CoreFX 4.700.21.16208), X64 RyuJIT

Method	N	Mean	Error	StdDev	Gen 0	Gen 1	Gen 2	Allocated
PipeLineOneBenchmark	1000	25.00 μs	0.269 μs	0.252 μs	-	-	-	-
PipeLineOneBenchmark	100000	2,491.42 μs	13.655 μs	15.177 μs	-	-	-	-

Which is similar but different from your solution.

Nevertheless, it makes me think, that the actual problem is elsewhere. (Still working on it and going to update.)

I was thinking of a little tool to see if your rabbit is simply too slow and you have substantial build-up:

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks.Dataflow;

namespace DataFlowExperiment.PipelinesLib
{
    public class RabbitWrapper : IDisposable
    {
        private readonly int batchSize = 10;

        private Thread senderThread;
        private readonly BlockingCollection messages;
        private readonly ActionBlock receiver;
        private readonly CancellationTokenSource stoppingToken;
        private readonly RabbitWrapperStats stats;

        private ITargetBlock Receiver => receiver;

        public RabbitWrapper()
        {
                                                                // Drop in your logging here
            stats = new RabbitWrapperStats(new Progress(x => Console.WriteLine(x)));
            stoppingToken = new CancellationTokenSource();
            messages = new BlockingCollection();
            receiver = new ActionBlock(Receive);
            senderThread = new Thread(HandleQueue);
            senderThread.Start();
        }

        private void Receive(string message)
        {
            messages.Add(message);
        }

        private void HandleQueue()
        {
            while (!stoppingToken.Token.IsCancellationRequested)
            {
                int batchIndex = 0;
                do {
                    string message = messages.Take(stoppingToken.Token);
                    if (!string.IsNullOrEmpty(message))
                    {
                        SendToRabbit(message);
                    }
                    batchIndex++;
                } while (!stoppingToken.Token.IsCancellationRequested &&
                         batchIndex < batchSize &&
                         messages.Count > 0);
                // Check statistics every 10 messages.
                CheckStats(messages.Count);
            }
        }

        private void SendToRabbit(string message)
        {
            // rabbit Publish goes here.
        }

        private void CheckStats(int count)
        {
            stats.CheckStats(count);
        }

        public void Close()
        {
            this.stoppingToken.Cancel();
            senderThread.Join();
        }

        public void Dispose()
        {
            Close();
        }
    }

    internal class RabbitWrapperStats
    {
        // You may want to play around with these thresholds
        // I pulled them out of thin air ...
        const int SIZE_WARN = 500000;
        const int SIZE_CRITICAL = SIZE_WARN * 2;

        private int lastTenIndex = 0;
        private int[] lastTen = new int[10] { 0, 0, 0, 0, 0, 0, 0, 0, 0, 0 };
        private int meanSizeLastTen = 0;
        private int lastMeanSize = 0;
        private int tendency = 0;

        private bool HasWarned = false;
        private bool HasPanicked = false;

        private readonly IProgress progress;

        public RabbitWrapperStats(IProgress progress)
        {
            this.progress = progress;
        }

        public void CheckStats(int queueSize)
        {
            UpdateLastTen(queueSize);

            if (!HasPanicked && queueSize > SIZE_CRITICAL)
            {
                Panic(queueSize);
                return;
            }

            if (!HasWarned && queueSize > SIZE_WARN)
            {
                Warn(queueSize);
                return;
            }

            if ((HasPanicked || HasWarned ) && meanSizeLastTen < SIZE_WARN * 0.75)
            {
                HasPanicked = false;
                HasWarned = false;
                progress?.Report($"INFO Mean size of last 10 Samples sinks below {SIZE_WARN * 0.75} : {meanSizeLastTen}");
            }
        }

        private void Warn(int size)
        {
            HasWarned = true;
            progress?.Report($"WARNING QueueSize = {size}");
        }

        private void Panic(int size)
        {
            HasPanicked = true;
            progress?.Report($"!! CRITICAL !! QueueSize = {size}");
        }

        private void UpdateLastTen(int value)
        {
            lastTen[lastTenIndex] = value;
            lastTenIndex = ++lastTenIndex % lastTen.Length;
            meanSizeLastTen = lastTen.Sum() / lastTen.Length;
            tendency = meanSizeLastTen.CompareTo(lastMeanSize);
            lastMeanSize = meanSizeLastTen;
        }
    }
}

RabbitMQ Producer C# .NET Core 5.0 Memory Leak

Edit 1

Edit 2

Edit 3

Edit 4

1 Answers

EDIT

For reference:

In response to EDIT 3 in the question:

EDIT 2