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

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比执行查询/ sleep 的代码花费更多的时间。在那种情况下，线程会不断增长。你可能还会拥有各种 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[]>;

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

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

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