Huge memory usage

[sachabarber]

Overview

So we use QuickFixN within some production code. We have observed that over a period of 4-5 days the memory footprint of our app grows up to around 2GB.

We profiled the app, and can see that the cause appears to be FileStore where the offsets_ Dictionary is storing too much data

One thing to note, is that left alone, our app rarely restarts

I have downloaded the QuickFixN code and been digging about to see where the offsets_ is used and how it is cleared, and the following is what I see

FileStore

The FileStore class contains the offsets_ which is only cleared, when the ConstructFromFileCache() method is called. Which in turn is called when the FileStore.open() method is called, which is only called under 2 conditions

• FileStore construction
• Reset() called on FileStore

So I was thinking perhaps the Reset needs to be called. But by whom and when?

SessionState

So I dug some more, and can see that the FileStore.Reset() is called from SessionState.Reset(string reason). Which in turn is called by various other call sites from Session.

Session

• The calls to SessionState.Reset() looks like an integral part of the QuickFixN implementation, so these call sites are obviously correct and tested by this repo. Basically I trust the existing Session calls to to SessionState.Reset()

So the question

Is this memory footprint normal, or is specific QuickFixN application code expected to also call the Session.Reset() which will call the callsites I mention above? Where the end result could be a call to the FileStore.ConstructFromFileCache.

However looking into this, I am not sure a Reset would work, as from what I can tell the FileStore.ConstructFromFileCache() method would read the contents of the files shown in image below to construct the offsets_. Which is obviously will just fill the offsets_ straight back up.

private void ConstructFromFileCache()
{
    offsets_.Clear();
    if (System.IO.File.Exists(headerFileName_))
    {
        using (System.IO.StreamReader reader = new System.IO.StreamReader(headerFileName_))
        {
            string line;
            while ((line = reader.ReadLine()) != null)
            {
                string[] headerParts = line.Split(',');
                if (headerParts.Length == 3)
                {
                    offsets_[Convert.ToInt32(headerParts[0])] = new MsgDef(
                        Convert.ToInt64(headerParts[1]), Convert.ToInt32(headerParts[2]));
                }
            }
        }
    }

    if (System.IO.File.Exists(seqNumsFileName_))
    {
        using (System.IO.StreamReader seqNumReader = new System.IO.StreamReader(seqNumsFileName_))
        {
            string[] parts = seqNumReader.ReadToEnd().Split(':');
            if (parts.Length == 2)
            {
                cache_.SetNextSenderMsgSeqNum(Convert.ToInt32(parts[0]));
                cache_.SetNextTargetMsgSeqNum(Convert.ToInt32(parts[1]));
            }
        }
    }
}

If I look at the current memory usage of our PROD QuickFixN code, which is using these files, the memory footprint is pretty much exactly the total bytes for these files, which is what I would expect looking at how the FileStore.ConstructFromFileCache() method works (shown above)

So not really sure that the Reset approach is going to help that much

So what choices do we have to maintain a decent memory footprint for our application?

Could someone advice what we should be looking at doing?

Right now we restart the app manually every 5 days, and from there it builds up to about 2GB over 5 days, but we would be keen to find a more programmatic solution

[tomerpeled]

Any solution for that?

[gbirchmeier]

The store is needed to support the ResendRequest functionality. This is the cache of messages that will be resent if a ResendRequest situation occurs.

Calling Reset() in user-code is a not an appropriate response; you're wiping out the store, and thus eliminating the ability to resend. Internally to QF/n, I believe Reset() is only called when the session restarts.

However, I agree that 2GB is over the top, and the scenario where we'd need to replay a whole day or more is unlikely. I'm just thinking in text right now, but perhaps we could have a configurable upper-store limit, like a max number of messages or a max number of hours/minutes ago. (Actually we do have a MaxMessagesInResendRequest, but I don't think any store impls are looking at it.)

The key problem here is that the store implementations aren't looking at any other settings than FileStorePath. For instance, If Resend is disabled, then the store doesn't need to store any messages content. If the store size is limited by
MaxMessagesInResendRequest, then the store could be able to trash the oldest stuff.

I'm curious to see if QF/j handles this differently.

I'll keep this issue in my mental queue, and come back to it after I knock a few other things out.

[benjaminjnr]

How about breaking the MessageStore into buckets?

So for instance, if quickfixn is used in app that run in Azure and we use an AppendOnlyBlob, there is a definite number of writes that it can accommodate over a long trading sessions. Splitting messages into multiple blobs that can then be chained together to present a complete picture can also help address this memory issue.

Because you will not need to keep the entire "queue" of messages physically in memory at any one time. The store can "broker or proxy" the request to the appropriate bucket and return what is needed. It could introduce some latency with resend or replay but it usually won't be a "critical code path".

[benjaminjnr]

Just seeing that #685 expressed a similar idea.

[max-kuzmin]

If ResendRequest ability is not important for your case the solution can be to use LimitedMemoryStore instead of FileStore which can be implemented like this:

internal sealed class LimitedMemoryStore : IMessageStore
{
    private readonly int _maxMessagesInStore;
    private Dictionary<int, string> _messages;
    private Queue<int> _keys;
    private int _nextSenderMsgSeqNum;
    private int _nextTargetMsgSeqNum;
    private DateTime _creationTime = DateTime.UtcNow;

    public LimitedMemoryStore(int maxMessagesInStore)
    {
        _maxMessagesInStore = maxMessagesInStore;
        _messages = new Dictionary<int, string>();
        _keys = new Queue<int>();
        Reset();
    }

    public void Get(int begSeqNo, int endSeqNo, List<string> messages)
    {
        for (int key = begSeqNo; key <= endSeqNo; ++key)
        {
            if (_messages.ContainsKey(key))
                messages.Add(_messages[key]);
        }
    }

    public bool Set(int msgSeqNum, string msg)
    {
        _messages[msgSeqNum] = msg;
        _keys.Enqueue(msgSeqNum);

        if (_messages.Count > _maxMessagesInStore)
            _messages.Remove(_keys.Dequeue());

        return true;
    }

    public int GetNextSenderMsgSeqNum() => _nextSenderMsgSeqNum;

    public int GetNextTargetMsgSeqNum() => _nextTargetMsgSeqNum;

    public void SetNextSenderMsgSeqNum(int value) => _nextSenderMsgSeqNum = value;

    public void SetNextTargetMsgSeqNum(int value) => _nextTargetMsgSeqNum = value;

    public void IncrNextSenderMsgSeqNum() => ++_nextSenderMsgSeqNum;

    public void IncrNextTargetMsgSeqNum() => ++_nextTargetMsgSeqNum;

    public DateTime? CreationTime => _creationTime;

    public DateTime GetCreationTime() => _creationTime;

    public void Reset()
    {
        _nextSenderMsgSeqNum = 1;
        _nextTargetMsgSeqNum = 1;
        _messages.Clear();
        _keys.Clear();
        _creationTime = DateTime.UtcNow;
    }

    public void Refresh()
    {
    }

    public void Dispose()
    {
    }
}

[ettie933]

Solved it by using a ring buffer style (fixed capacity) hot cache to keep last x amount of messages. Else it goes to the file (cold cache) to retrieve the positions for requested messages.

@gbirchmeier if you are happy with this approach let me know and I'll make a pull request.

See below:

1. Created FileStoreBase

internal class FileStoreBase : IDisposable
{
    protected readonly FileStream _stream;
    public long Position => (long)_stream?.Position!;

    internal FileStoreBase(string path)
    {
        _stream = new FileStream(
            path,
            FileMode.OpenOrCreate,
            FileAccess.ReadWrite,
            FileShare.ReadWrite,
            1 << 20,
            FileOptions.None);
    }

    internal void Purge(string filename)
    {
        _stream.Close();
        if (File.Exists(filename))
            File.Delete(filename);
    }

    #region IDisposable Members

    public void Dispose()
    {
        Dispose(true);
        GC.SuppressFinalize(this);
    }
    protected bool _disposed;
    protected virtual void Dispose(bool disposing)
    {
        if (_disposed) return;
        if (disposing)
        {
            _stream?.Dispose();
        }
        _disposed = true;
    }

    ~FileStoreBase() => Dispose(false);
    #endregion
}

2. Created FileHeaderStore

internal class FileHeaderStore : FileStoreBase, IDisposable
{
    internal readonly struct MsgDef(long index, int size)
    {
        internal long Index { get; } = index;

        internal int Size { get; } = size;
    }

    private readonly int _capacity;
    private readonly MsgDef[] _buffer;
    private readonly ulong[] _keys; // track keys for eviction
    private readonly Dictionary<ulong, int> _index;

    private int _next; // write position
    private int _count;  // number of valid items (<= capacity)

    internal FileHeaderStore(string path, int hotCapacity) : base(path) 
    {
        if (hotCapacity == 0)
        {
            throw new ArgumentNullException("HeaderStoreCapacity");
        }

        _capacity = hotCapacity;
        _buffer = new MsgDef[_capacity];
        _keys = new ulong[_capacity];
        _index = new Dictionary<ulong, int>(_capacity);
        _next = 0;
        _count = 0;
    }

    internal void Write(ulong seqNum, MsgDef messageDef)
    {
        var slot = _next;

        if (_count == _capacity)
        {
            // buffer full so evict oldest
            var oldKey = _keys[slot];
            _index.Remove(oldKey);
        }
        else
        {
            // still filling
            _count++;
        }

        _buffer[slot] = messageDef;
        _keys[slot] = seqNum;
        _index[seqNum] = slot;

        // advance ring pointer
        _next++;
        if (_next == _capacity)
            _next = 0;

        Span<byte> buffer = stackalloc byte[20];

        BinaryPrimitives.WriteUInt64LittleEndian(buffer, seqNum);
        BinaryPrimitives.WriteInt64LittleEndian(buffer[8..], messageDef.Index);
        BinaryPrimitives.WriteInt32LittleEndian(buffer[16..], messageDef.Size);

        _stream?.Write(buffer);
        _stream?.Flush();
    }

    internal bool TryFind(SeqNumType startSeqNum, SeqNumType endSeqNum, out List<MsgDef> results)
    {
        long originalPosition = (long)_stream?.Position!;
        results = RetrieveFromHotCache(startSeqNum, endSeqNum, out SeqNumType firstSeqNumFromHotCache, out SeqNumType lastSeqNumFromHotCache);

        if (firstSeqNumFromHotCache == startSeqNum)
            return true;
        else if (results.Count() > 0)
        {
            endSeqNum = firstSeqNumFromHotCache - 1;
        }   

        //Retrieve from cold cache
        try
        {
            _stream?.Position = 0;

            Span<byte> buffer = stackalloc byte[20];

            List<MsgDef> resultsColdCache = new List<MsgDef>();

            while (_stream?.Position < _stream?.Length)
            {
                _stream?.ReadExactly(buffer);

                ulong thisSeqNum = BinaryPrimitives.ReadUInt64LittleEndian(buffer);

                if (thisSeqNum >= startSeqNum && thisSeqNum <= endSeqNum)
                {
                    long offset = BinaryPrimitives.ReadInt64LittleEndian(buffer[8..]);
                    int size = BinaryPrimitives.ReadInt32LittleEndian(buffer[16..]);
                    resultsColdCache.Add(new MsgDef(offset, size));
                }
            }

            //Reorder cold cache first (oldest) + hot cache (newest)
            resultsColdCache.AddRange(results);
            results = resultsColdCache;

            return results.Count() > 0;
        }
        finally
        {
            _stream?.Position = originalPosition;
        }
    }

    private List<MsgDef> RetrieveFromHotCache(SeqNumType startSeqNum, SeqNumType endSeqNum, out SeqNumType firstSeqNumRetrieved, out SeqNumType lastSeqNumRetrieved)
    {
        firstSeqNumRetrieved = 0;
        lastSeqNumRetrieved = 0;

        int count = _count;
        int start = (_next - count + _capacity) % _capacity;

        // Pre-size list to avoid resizing (upper bound = count)
        var result = new List<MsgDef>(count);

        for (int i = 0; i < count; i++)
        {
            int idx = (start + i) % _capacity;
            ulong key = _keys[idx];

            // Skip until we hit range
            if (key < startSeqNum)
                continue;

            if(firstSeqNumRetrieved == 0)
            {
                firstSeqNumRetrieved = key;
            }

            // Early exit if keys are ordered
            if (key > endSeqNum)
                break;

            result.Add(_buffer[idx]);
            lastSeqNumRetrieved = key;
        }

        return result;
    }
}

3. Created FileMessageStore

internal class FileMessageStore : FileStoreBase
{
    internal FileMessageStore(string path) : base(path) { }

    internal void Write(SeqNumType msgSeqNum, ReadOnlySpan<byte> msg)
    {
        Span<byte> header = stackalloc byte[12];
        BinaryPrimitives.WriteUInt64LittleEndian(header, msgSeqNum);
        BinaryPrimitives.WriteInt32LittleEndian(header[8..], msg.Length);

        _stream?.Write(header);
        _stream?.Write(msg);
        _stream?.Flush();
    }

    internal int Read(long offset, Span<byte> buffer)
    {
        long originalPosition = (long)_stream?.Position!;

        try
        {
            _stream?.Position = offset;

            Span<byte> header = stackalloc byte[12];
            _stream?.ReadExactly(header);

            int len = BinaryPrimitives.ReadInt32LittleEndian(header[8..]);

            _stream?.ReadExactly(buffer[..len]);
            return len;
        }
        finally
        {
            _stream?.Position = originalPosition;
        }
    }
}

4. Created FileSeqNumStore - mainly just because of OCD since the rest was split out

internal class FileSeqNumStore : FileStoreBase
{
    internal FileSeqNumStore(string path) : base(path) { }

    internal void Write(SeqNumType nextSenderSeqNum, SeqNumType nextTargetSeqNum)
    {
        _stream.Position = 0;
        using var writer = new StreamWriter(_stream, leaveOpen: true);
        writer.Write($"{nextSenderSeqNum.ToString("D20")}:{nextTargetSeqNum.ToString("D20")}");
    }

    internal bool Read(out ulong nextSenderSeqNum, out ulong nextTargetSeqNum)
    {
        if(_stream?.Length > 0)
        {
            Span<byte> buffer = stackalloc byte[41];
            _stream?.ReadExactly(buffer);

            // Decode only once
            Span<char> chars = stackalloc char[41];
            int charCount = Encoding.ASCII.GetChars(buffer, chars);

            // Parse fixed positions
            nextSenderSeqNum = ulong.Parse(chars.Slice(0, 20));
            nextTargetSeqNum = ulong.Parse(chars.Slice(21, 20));

            return true;
        }

        nextSenderSeqNum = 0;
        nextTargetSeqNum = 0;
        return false;
    }
}

5. Adjusted FileStore accordingly

public class FileStore : IMessageStore
{
    private readonly string _seqNumsFileName;
    private readonly string _msgFileName;
    private readonly string _headerFileName;
    private readonly string _sessionFileName;
    private readonly int _hotCacheSize;

    private FileSeqNumStore _seqNumsFile;
    private FileMessageStore _msgFile;
    private FileHeaderStore _headerFile;

    private readonly MemoryStore _cache = new();

    public static string Prefix(SessionID sessionId)
    {
        StringBuilder prefix = new StringBuilder(sessionId.BeginString)
            .Append('-').Append(sessionId.SenderCompID);
        if (SessionID.IsSet(sessionId.SenderSubID))
            prefix.Append('_').Append(sessionId.SenderSubID);
        if (SessionID.IsSet(sessionId.SenderLocationID))
            prefix.Append('_').Append(sessionId.SenderLocationID);
        prefix.Append('-').Append(sessionId.TargetCompID);
        if (SessionID.IsSet(sessionId.TargetSubID))
            prefix.Append('_').Append(sessionId.TargetSubID);
        if (SessionID.IsSet(sessionId.TargetLocationID))
            prefix.Append('_').Append(sessionId.TargetLocationID);

        if (SessionID.IsSet(sessionId.SessionQualifier))
            prefix.Append('-').Append(sessionId.SessionQualifier);

        return prefix.ToString();
    }

    /// <summary>
    ///
    /// </summary>
    /// <param name="path">
    /// All back or forward slashes in this path will be converted as needed to the running platform's preferred
    /// path separator (i.e. "/" will become "\" on windows, else "\" will become "/" on all other platforms)
    /// </param>
    /// <param name="sessionId"></param>
    public FileStore(string path, int hotCacheSize, SessionID sessionId)
    {
        string normalizedPath = StringUtil.FixSlashes(path);

        if (!System.IO.Directory.Exists(normalizedPath))
            System.IO.Directory.CreateDirectory(normalizedPath);

        string prefix = Prefix(sessionId);

        _hotCacheSize = hotCacheSize;
        _seqNumsFileName = System.IO.Path.Combine(normalizedPath, prefix + ".seqnums");
        _msgFileName = System.IO.Path.Combine(normalizedPath, prefix + ".body");
        _headerFileName = System.IO.Path.Combine(normalizedPath, prefix + ".header");
        _sessionFileName = System.IO.Path.Combine(normalizedPath, prefix + ".session");

        // The compiler isn't smart enough to see that Open() initializes these 3 vars,
        // but we can use "= null!" to make it accept that these are non-null
        _seqNumsFile = null!;
        _msgFile = null!;
        _headerFile = null!;
        Open();
    }

    private void Open()
    {
        Close();

        
        InitializeSessionCreateTime();

        _seqNumsFile = new FileSeqNumStore(_seqNumsFileName);
        ConstructFromFileCache();

        _msgFile = new FileMessageStore(_msgFileName);
        _headerFile = new FileHeaderStore(_headerFileName, _hotCacheSize);
    }

    private void Close()
    {
        // these vars will be null only during construction (ctor()->Open()->Close())
        _seqNumsFile?.Dispose();
        _msgFile?.Dispose();
        _headerFile?.Dispose();
    }

    private static void PurgeSingleFile(string filename)
    {
        if (System.IO.File.Exists(filename))
            System.IO.File.Delete(filename);
    }

    private void PurgeFileCache()
    {
        _seqNumsFile.Purge(_seqNumsFileName);
        _msgFile?.Purge(_msgFileName);
        _headerFile?.Purge(_headerFileName);
        PurgeSingleFile(_sessionFileName);
    }


    private void ConstructFromFileCache()
    {
        if(_seqNumsFile.Read(out ulong nextSenderSeqNum, out ulong nextTargetSeqNum))
        {
            _cache.NextSenderMsgSeqNum = nextSenderSeqNum;
            _cache.NextTargetMsgSeqNum = nextTargetSeqNum;
        }
    }

    private void InitializeSessionCreateTime()
    {
        if (System.IO.File.Exists(_sessionFileName) && new System.IO.FileInfo(_sessionFileName).Length > 0)
        {
            using (System.IO.StreamReader reader = new System.IO.StreamReader(_sessionFileName))
            {
                string s = reader.ReadToEnd();
                _cache.CreationTime = UtcDateTimeSerializer.FromString(s);
            }
        }
        else
        {
            using (System.IO.StreamWriter writer = new System.IO.StreamWriter(_sessionFileName, false)) {
                writer.Write(UtcDateTimeSerializer.ToString(_cache.CreationTime ?? new DateTime()));
            }
        }
    }


    #region MessageStore Members

    /// <summary>
    /// Get messages within the range of sequence numbers
    /// </summary>
    /// <param name="startSeqNum"></param>
    /// <param name="endSeqNum"></param>
    /// <param name="messages"></param>
    public void Get(SeqNumType startSeqNum, SeqNumType endSeqNum, List<string> messages)
    {
        _headerFile.TryFind(startSeqNum, endSeqNum, out List<FileHeaderStore.MsgDef> messageLookup);

        foreach(var lookup in messageLookup)
        {
            byte[] msgBytes = ArrayPool<byte>.Shared.Rent(lookup.Size);
            try
            {
                _msgFile.Read(lookup.Index, new Span<byte>(msgBytes, 0, lookup.Size));
                messages.Add(CharEncoding.SelectedEncoding.GetString(new ReadOnlySpan<byte>(msgBytes, 0, lookup.Size)));
            }
            finally
            {
                ArrayPool<byte>.Shared.Return(msgBytes);
            }
        }
    }

    /// <summary>
    /// Store a message
    /// </summary>
    /// <param name="msgSeqNum"></param>
    /// <param name="msg"></param>
    /// <returns></returns>
    public bool Set(SeqNumType msgSeqNum, string msg)
    {
        long offset = _msgFile.Position;

        using ValueDisposable _ = CharEncoding.GetBytes(msg.AsSpan(), out ReadOnlySpan<byte> msgBytes);

        _headerFile.Write(msgSeqNum, new FileHeaderStore.MsgDef(offset, msgBytes.Length));
        _msgFile.Write(msgSeqNum, msgBytes);

        return true;
    }

    public SeqNumType NextSenderMsgSeqNum {
        get => _cache.NextSenderMsgSeqNum;
        set {
            _cache.NextSenderMsgSeqNum = value;
            SetSeqNum();
        }
    }

    public SeqNumType NextTargetMsgSeqNum {
        get => _cache.NextTargetMsgSeqNum;
        set {
            _cache.NextTargetMsgSeqNum = value;
            SetSeqNum();
        }
    }

    public void IncrNextSenderMsgSeqNum()
    {
        _cache.IncrNextSenderMsgSeqNum();
        SetSeqNum();
    }

    public void IncrNextTargetMsgSeqNum()
    {
        _cache.IncrNextTargetMsgSeqNum();
        SetSeqNum();
    }

    private void SetSeqNum()
    {
        _seqNumsFile.Write(NextSenderMsgSeqNum, NextTargetMsgSeqNum);
    }

    public DateTime? CreationTime => _cache.CreationTime;

    public void Reset()
    {
        _cache.Reset();
        PurgeFileCache();
        Open();
    }

    public void Refresh()
    {
        _cache.Reset();
        Open();
    }

    #endregion

    #region IDisposable Members

    public void Dispose()
    {
        Dispose(true);
        GC.SuppressFinalize(this);
    }
    private bool _disposed;
    protected virtual void Dispose(bool disposing)
    {
        if (_disposed) return;
        if (disposing)
        {
            Close();
        }
        _disposed = true;
    }

    ~FileStore() => Dispose(false);
    #endregion
}

6. Passed the cap to FileStoreFactory

public class FileStoreFactory : IMessageStoreFactory
{
    private readonly SessionSettings _settings;
    private readonly int _hotCacheSize;

    /// <summary>
    /// Create the factory with configuration in session settings
    /// </summary>
    /// <param name="settings"></param>
    public FileStoreFactory(SessionSettings settings, int hotCacheSize)
    {
        _settings = settings;
        _hotCacheSize = hotCacheSize;
    }

    /// <summary>
    /// Creates a file-based message store
    /// </summary>
    /// <param name="sessionId">session ID for the message store</param>
    /// <returns></returns>
    public IMessageStore Create(SessionID sessionId)
    {
        return new FileStore(_settings.Get(sessionId).GetString(SessionSettings.FILE_STORE_PATH), _hotCacheSize, sessionId);
    }
}

[gbirchmeier]

@ettie933 Rather than changing FileStore, I suggest you turn this into new implementation of IMessageStore.

And yes, I would love to see it in a PR.

[ettie933]

@ettie933 Rather than changing FileStore, I suggest you turn this into new implementation of IMessageStore.

And yes, I would love to see it in a PR.

Will do. Thanks @gbirchmeier

[ettie933]

Hello @gbirchmeier

We are still making adjustments on our side.
The above failed acceptance internally due to a delay in initial/first response when sending a resend request for a more considerable amount of messages, sometimes exceeding heartbeat interval resulting in a connection drop. We noted the same with the master branch.

Currently exploring the following solution:

1. Header file to contain ranges (stored in binary format) e.g.

• Messages with seq 1 starts at pos 0 in body file
• Messages with seq 50_000 to start at pos x in body file
• and so on
• Header store has a dictionary with these for fast lookup during runtime (16 bytes per entry so memory usage is minimal)
• Recoverable from the header file on restart

2. Body file stores seq + length + message content - all pure binary

3. Replay uses above sequence ranges in point 1 to get estimate start position within the body file, then seeks the body file from the start seq since the sequences will always be sequential.

4. Passed a action method to the Get method, so that we can start sending immediately as messages are read and avoid the initial delay of first completing the entire read before it starts sending

Thus far we have also made some other changes, all together also resulted in just a little over double the throughput per second.

Will get in contact once our team is done with QA and we have internal signoff then we can take it from there