该线程如何导致内存泄漏？

Question

我们的一个程序遭受了严重的内存泄漏：它的进程内存在客户站点每天增加 1 GB。 我可以在我们的测试中心设置场景，每天可能会发生大约 700 MB 的内存泄漏。

此应用程序是用 C# 编写的 Windows 服务，它通过 CAN 总线与设备通信。

内存泄漏不取决于应用程序写入 CAN 总线的数据速率。但这显然取决于收到的消息数量。

阅读消息的“非托管”方面是：

[StructLayout(LayoutKind.Sequential, Pack = 1)]
public struct CAN_MSG
{
    public uint time_stamp;
    public uint id;
    public byte len;
    public byte rtr;
    [MarshalAs(UnmanagedType.ByValArray, SizeConst = 8)]
    public byte[] a_data;
}

[DllImport("IEICAN02.dll", EntryPoint = "#3")]
public static extern int CAN_CountMsgs(ushort card_idx, byte can_no, byte que_type);
//ICAN_API INT32 _stdcall CAN_CountMsgs(UINT16 card_idx, UINT8 can_no,UINT8 que_type);

[DllImport("IEICAN02.dll", EntryPoint = "#10")]
public static extern int CAN_ReadMsg(ushort card_idx, byte can_no, ushort count, [MarshalAs(UnmanagedType.LPArray), Out()] CAN_MSG[] msg);
//ICAN_API INT32 _stdcall CAN_ReadMsg(UINT16 card_idx, UINT8 can_no, UINT16 count, CAN_MSG* p_obj);

我们基本上使用如下：

private void ReadMessages()
{
    while (keepRunning)
    {
        // get the number of messages in the queue
        int messagesCounter = ICAN_API.CAN_CountMsgs(_CardIndex, _PortIndex, ICAN_API.CAN_RX_QUE);
        if (messagesCounter > 0)
        {
            // create an array of appropriate size for those messages
            CAN_MSG[] canMessages = new CAN_MSG[messagesCounter];
            // read them
            int actualReadMessages = ICAN_API.CAN_ReadMsg(_CardIndex, _PortIndex, (ushort)messagesCounter, canMessages);
            // transform them into "our" objects
            CanMessage[] messages = TransformMessages(canMessages);
            Thread thread = new Thread(() => RaiseEventWithCanMessages(messages))
            {
                Priority = ThreadPriority.AboveNormal
            };
            thread.Start();
        }
        Thread.Sleep(20);
    }
}

 // transformation process:
new CanMessage
{
    MessageData = (byte[])messages[i].a_data.Clone(),
    MessageId = messages[i].id
};

循环每约 30 毫秒执行一次。

当我在同一个线程中调用 RaiseEventWithCanMessages(messages) 时，内存泄漏消失了（嗯，不完全，每天大约 10 MB - 即大约 1% 的原始泄漏 - 仍然存在，但其他泄漏可能无关）。

我不明白这种线程的创建如何导致内存泄漏。您能否提供一些信息是如何导致内存泄漏的？

附录 2018-08-16： 该应用程序以大约 50 MB 的内存开始，并在大约 2GB 时崩溃。这意味着，千兆字节的内存在大多数时间都是可用的。 此外，CPU 大约 20% - 4 个内核中有 3 个处于空闲状态。 应用程序使用的线程数保持在大约 30 个线程左右。 总的来说，垃圾收集有很多可用的资源。尽管如此，GC 还是失败了。

每秒大约有 30 个线程，每天有 700 MB 的内存泄漏，平均每个新创建的线程有大约 300 字节的内存泄漏；每个新线程约 5 条消息，每条消息约 60 字节。 “非托管”结构没有进入新线程，其内容被复制到新实例化的类中。

那么：尽管有大量可用资源，为什么 GC 会失败？

Answer 1

您每约 30 毫秒创建 2 个数组和一个线程，它们之间没有任何协调。数组可能是个问题，但坦率地说，我非常更担心线程——创建线程真的非常昂贵。您应该不如此频繁地创建它们。

我还担心如果读取循环超出线程会发生什么 - 即如果 RaiseEventWithCanMessages 比执行查询/睡眠的代码花费更多时间。在这种情况下，您将拥有不断增长的线程。而且您可能还会遇到各种RaiseEventWithCanMessages 互相争斗。

将RaiseEventWithCanMessages 内联“修复”这一事实表明，这里的主要问题要么是正在创建的线程的绝对数量（坏），要么是许多重叠的和增长的并发 RaiseEventWithCanMessages.

---

最简单的解决方法是：不要在这里使用额外的线程。

如果你真的想要并发操作，我会在这里有两个线程——一个执行查询，一个执行RaiseEventWithCanMessages 的任何操作，两者都在一个循环中。然后，我将在线程之间进行协调，以便查询线程 等待 完成前一个 RaiseEventWithCanMessages 事情，以便它以协调的方式将其移交 - 所以总是最多有一个未完成的RaiseEventWithCanMessages，如果没有跟上，您将停止运行查询。

基本上：

CanMessage[] messages = TransformMessages(canMessages);
HandToConsumerBlockingUntilAvailable(messages); // TODO: implement

其他线程基本上在做：

var nextMessages = BlockUntilAvailableFromProducer(); // TODO: implement

一个非常基本的实现可能只是：

void HandToConsumerBlockingUntilAvailable(CanMessage[] messages) {
    lock(_queue) {
        if(_queue.Length != 0) Monitor.Wait(_queue); // block until space
        _queue.Enqueue(messages);
        if(queue.Length == 1) Monitor.PulseAll(_queue); // wake consumer
    }
}
CanMessage[] BlockUntilAvailableFromProducer() {
    lock(_queue) {
        if(_queue.Length == 0) Monitor.Wait(_queue); // block until work
        var next = _queue.Dequeue();
        Monitor.Pulse(_queue); // wake producer
        return _next;
    }
}
private readonly Queue<CanMessage[]> _queue = new Queue<CanMessage[]>;

此实现强制队列中未处理的未处理Message[] 不得超过 1 个。

这解决了创建大量线程的问题，和查询循环超过RaiseEventWithCanMessages 代码的问题。

我可能也考虑使用ArrayPool<T>.Shared 来租用超大数组（意思是：您需要注意不要读取比实际写入更多的数据，因为您可能要求了一个 500 的数组，但得到了一个大小为 512 的数组），而不是不断分配数组。