内存泄漏

[AmdRyZen]

drogon::app().loadConfigFile("config.json").run();
上次说的 在 cmake 使用 mimalloc -static 内存泄漏问题
今天我在使用 gcc pgo 优化代码的时候又遇到了 还是希望可以解决

macos 14.1 m2 pro
gcc 11.4

[AmdRyZen]

今天在做 Websocket 压测的时候 内存占用越来越大 http服务压测 一直都是 22mb 很正常

[AmdRyZen]

[an-tao]

最好分享一个demo可以在我本地复现的。

[AmdRyZen]

#include "EchoWebsocket.h"
#include "utils/redisUtils.h"
#include "boost/format.hpp"
#include "rapidjson/document.h"
#include "rapidjson/prettywriter.h"

using namespace rapidjson;

struct Subscriber
{
std::string chatRoomName_;
SubscriberID id_{};
};

void EchoWebsocket::handleNewMessage(const WebSocketConnectionPtr& wsConnPtr, std::string&& message, const WebSocketMessageType& type)
{
try
{
if (type == WebSocketMessageType::Ping)
{
wsConnPtr->send("pong", WebSocketMessageType::Pong);
LOG_DEBUG << "recv a ping";
return;
}

    if (!message.empty())
    {
        Document document;
        if (document.Parse(message.c_str()).HasParseError())
        {
            return;
        }

        if (document.IsNull())
        {
            std::cerr << "JSON is empty!" << std::endl;
            return;
        }

        std::string command;
        if (document.HasMember("key") && document["key"].IsString())
        {
            command = std::format("get {}", document["key"].GetString());
        }

        std::string action;
        if (document.HasMember("action") && document["action"].IsString())
        {
            action = document["action"].GetString();
        }
        std::string msgContent = document["msgContent"].GetString();

        // 在处理用户退出时检查连接状态
        if (!wsConnPtr->disconnected())
        {
            const auto& subscriber = wsConnPtr->getContextRef<Subscriber>();
            const auto& [chatRoomName, id] = subscriber;

            auto sharedThis = shared_from_this();
            async_run([action, msgContent, command, chatRoomName, id, sharedThis]() -> Task<>
            {
                try
                {
                    std::string_view data;
                    if (!command.empty())
                    {
                        data = co_await redisUtils::getCoroRedisValue(command);
                    }

                    if (!action.empty())
                    {
                       if (action == "message")
                        {
                            // 发送消息到聊天室
                           const std::string formattedMessage = std::format(R"({{"sender": "{}", "message": "{} ====> {}}})", id, msgContent, data);
                           sharedThis->chatRooms_.publish(chatRoomName, formattedMessage);
                        }
                        // 其他操作...
                    }
                }
                catch (const std::exception& e)
                {
                    std::cerr << "Error in async task: " << e.what() << std::endl;
                }
                co_return;
            });
        }
        command.clear();
    }
}
catch (...)
{
    std::cout << "handleNewMessage ..." << std::endl;
}

}
void EchoWebsocket::handleNewConnection(const HttpRequestPtr& req, const WebSocketConnectionPtr& wsConnPtr)
{
//write your application logic here
std::cout << "handleNewConnection" << std::endl;

Subscriber s;
s.chatRoomName_ = req->getParameter("room_name");
const std::string_view userName_ = req->getParameter("name");
// 处理用户加入聊天室
wsConnPtr->send(std::format("欢迎 {} 加入我们 {}", userName_, s.chatRoomName_));
s.id_ = chatRooms_.subscribe(s.chatRoomName_,
                             [wsConnPtr](const std::string& topic,
                                         const std::string& message) {
                                 // Supress unused variable warning
                                 (void)topic;
                                 wsConnPtr->send(message);
                             });
std::cout << "id = " << s.id_ << std::endl;
std::cout << "chatRoomName = " << s.chatRoomName_ << std::endl;
wsConnPtr->setContext(std::make_shared<Subscriber>(std::move(s)));

}
void EchoWebsocket::handleConnectionClosed(const WebSocketConnectionPtr& wsConnPtr)
{
//write your application logic here
try
{
//std::cout << "handleConnectionClosed" << std::endl;
// 获取Subscriber引用
const auto& subscriber = wsConnPtr->getContextRef();
// 使用结构化绑定提取成员变量
const auto& [chatRoomName, id] = subscriber;
// 退出所有房间
chatRooms_.unsubscribe(chatRoomName, id);
// 清理资源
wsConnPtr->clearContext();

    std::cout << "handleConnectionClosed id = " << id << std::endl;
    std::cout << "handleConnectionClosed chatRoomName = " << chatRoomName << std::endl;
}
catch (...)
{
    std::cout << "handleConnectionClosed ..." << std::endl;
}

}

#pragma once
#include <drogon/PubSubService.h>
#include <drogon/WebSocketController.h>

using namespace drogon;
class EchoWebsocket final : public WebSocketController, public std::enable_shared_from_this
{
public:
void handleNewMessage(const WebSocketConnectionPtr&,
std::string&&,
const WebSocketMessageType&) override;
void handleNewConnection(const HttpRequestPtr&,
const WebSocketConnectionPtr&) override;
void handleConnectionClosed(const WebSocketConnectionPtr&) override;
WS_PATH_LIST_BEGIN
// list path definitions here;
// WS_PATH_ADD("/path","filter1","filter2",...);
WS_PATH_ADD("/echo");
WS_PATH_LIST_END

private:
PubSubServicestd::string chatRooms_;
};

[AmdRyZen]

import ws from 'k6/ws';
import { check } from 'k6';
import { sleep } from 'k6';

// k6 run -u 100 ./websocket.js
export let options = {
vus: 10, // 同时运行的虚拟用户数量
duration: '30s', // 测试持续时间，这里设置为30秒
};

export default function () {
const url = 'ws://127.0.0.1:9090/echo?room_name=001聊天室&name=cat'; // WebSocket URL
const params = { tags: { my_tag: 'websocket' } };

const message = JSON.stringify({
    "key1": "aa",
    "action": "message",
    "msgContent": "hahahahh"
    /* "Name": "Liming",
    "Age": 26,
    "Language": [
        "C++",
        "Java"
    ],
    "E-mail": {
        "Netease": "lmshao@163.com",
        "Hotmail": "liming.shao@hotmail.com"
    }*/
});

const res = ws.connect(url, params, function (socket) {
    socket.on('open', function open() {
        console.log('connected');
        socket.send(message);
    });

    socket.on('message', function (data) {
        socket.send(message);

        // 检查是否是最后一次迭代并关闭连接
       /* if (__ITER === __VU - 1) {
            console.log('Closing the socket after last iteration');
            socket.close();
        }*/
    });

    socket.on('close', function () {
        console.log('disconnected');
    });

    socket.on('error', function (e) {
        console.log('An error occurred:', e.error());
    });

    // 在一定时间后手动关闭连接
   /* socket.setTimeout(function () {
        console.log('Closing the socket after timeout');
        socket.close();
    }, 10000); // 10秒后关闭连接*/
});

check(res, { 'status is 101': (r) => r && r.status === 101 });

// 添加 sleep 以模拟持续的连接活动
//sleep(1);

}

[AmdRyZen]

我基本上就是按照你们的demo写的 稍微改动了一点点东西 而且我还在基础上加了 std::enable_shared_from_this
然后用 k6 run ./websocket.js 压测 把时间改久一点 内存越来越大

[AmdRyZen]

需要 gcc 14.1 我这边用了 std::format

[AmdRyZen]

性能还是不错的 m2 pro 单机自己压测自己 10个用户每秒可以15万的消息发送和接受

[AmdRyZen]

压测完成之后 也是断开连接的

[AmdRyZen]

void EchoWebsocket::handleConnectionClosed(const WebSocketConnectionPtr& wsConnPtr)
{
//write your application logic here
try
{
//std::cout << "handleConnectionClosed" << std::endl;
// 获取Subscriber引用
const auto& subscriber = wsConnPtr->getContextRef();
// 使用结构化绑定提取成员变量
const auto& [chatRoomName, id] = subscriber;
// 退出所有房间
chatRooms_.unsubscribe(chatRoomName, id);
// todo 暂时不确定是否需要
if (chatRooms_.size() == 0)
{
std::cout << "chatRooms_.size() = " << chatRooms_.size() << std::endl;
chatRooms_.clear();
}
// 清理资源
wsConnPtr->clearContext();

    std::cout << "handleConnectionClosed id = " << id << std::endl;
    std::cout << "handleConnectionClosed chatRoomName = " << chatRoomName << std::endl;
}
catch (...)
{
    std::cout << "handleConnectionClosed ..." << std::endl;
}

} 我新增了 chatRooms_.clear(); 内存并没有减少

[an-tao]

1. 发布订阅的key是有限个还是随着压测增加的？
2. 建立的ws连接是有限个还是随着压测增加的？如果ws不停的重新建立，确认一下服务端的session是开着还是关着
3. 用gcc的AddressSanitizer工具试试能不能检测到。

[AmdRyZen]

我给了10 个用 写在 js 里面
ws 不会不停的建立 session 我根本就没打开
#2017 (comment)

[AmdRyZen]

你用你们的 demo 用 K6 压测一下就知道了

[hwc0919]

能否整理一下贴出来的代码，现在格式很乱没办法看。

[AmdRyZen]

https://github.com/AmdRyZen/drogon-http
这是我最新的代码 在

[AmdRyZen]

其实就一个控制器 .cc 和 .h
剩下的 是 K6 压测的js 你可以用其他的工具去压测
或者你直接用你们 demo 压测 看看有没有同样的问题
我压测http 是正常的没有内存问题 ws 长期压测也是用了你们的demo

[hwc0919]

我用你的脚本测试 example/websocket_server, 两轮后内存稳定在500M不再增长，没有观察到内存泄露迹象。

[AmdRyZen]

那你能压测一下我这个吗 https://github.com/AmdRyZen/drogon-http 就多了一个Task<> 不知道 gcc14 mimalloc 不同的操作系统会影响结果

[AmdRyZen]

我已经把 handleNewMessage 非常的精简 内存还是暴涨
macos
gcc 14.1

[hwc0919]

如果有这么明显的内存泄漏，valgrind应该能找到，你可以试一下.

我不清楚你一直在说内存泄漏的依据是什么。内存持续增长并不意味着一定发生了内存泄漏。没有内存泄漏的程序也不意味着不会出现OOM。 我不清楚 k6 的压测具体实现，以及它的连接管理策略。如果输入量超过系统处理速度，内存持续增长几乎是必然的。

我用 ”-u 100“ 参数进行压测，内存占用也会上升到10G, 但是测试结束后又恢复到了20M

[AmdRyZen]

我的就是不会恢复 用的这个没检测到 而且我确定在close的时候把资源全部清理了 你是Linux？

k6 或者其他压测 的连接管理策略 应该是结束之后也会恢复 你的 -u 100 指的是什么 drogon 自己的吗

[an-tao]

标准库里很多容器都是贪婪策略，会保持峰值的内存用量，这就跟写法有关了，你第一次压测停止后，再压测一次，如果内存没有增长到第一次压测的两倍并且变化不大，那么就是前面说的这种情形。

[AmdRyZen]

我有时间再试一下 到时候截图出来

[AmdRyZen]

这是 http的压测

[AmdRyZen]

ws 第一次

[AmdRyZen]

ws 第二次

[AmdRyZen]

ws 第三次

[AmdRyZen]

每次会增长一点点 不会翻倍 不过性能都比上一次 会差一点点

[AmdRyZen]

close 之后 内存保持不动了 不会减少

[an-tao]

内存增加没有正比于压测次数，不像典型的内存泄露，但是也很难下定论，建议你把服务端的逻辑逐步减少，看看哪里的功能导致的内存增加。

[an-tao]

close之后不减少不一定就是内存泄露，原因我说过了，很多容器对内存使用会保持峰值的状态，比如std::vector，缩小它的size并不会减少内存，它只是内部调整边界。

[AmdRyZen]

嗯 不过我就是简单的 广播 没有特别的业务
有可能mac的 gcc14 的问题

[AmdRyZen]

dg_ctl(78019,0x201608c00) malloc: *** error for object 0x102bff910: pointer being freed was not allocated
dg_ctl(78019,0x201608c00) malloc: *** set a breakpoint in malloc_error_break to debug
这个问题还是没办法解决对吧 释放了空指针

[an-tao]

没在本地复现

[AmdRyZen]

mac m2 pro
gcc 14.1 才有
gcc 11 12 13 都是正常的 上次不知道你还记得 mimalloc 和 mimalloc-static的问题 估计就是这个问题
gcc 14 可能把问题跑出来了

[an-tao]

mac下怎么装的14.1？我用brew没搜到。。。先观察下，等等14的后续版本

[AmdRyZen]

怪不得你们不能复现 这个问题其实是删除了一个已经不存在的内存 只是老版本的 不会抛出来

[AmdRyZen]

14 已经支持 std::format 了