内存泄漏 about drogon HOT 38 OPEN

没在本地复现

from drogon.

from drogon.

from drogon.

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

from drogon.

#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_;
};

from drogon.

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": "[email protected]",
        "Hotmail": "[email protected]"
    }*/
});

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);

}

from drogon.

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

from drogon.

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

from drogon.

from drogon.

from drogon.

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(); 内存并没有减少

from drogon.

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

from drogon.

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

from drogon.

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

from drogon.

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

from drogon.

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

from drogon.

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

from drogon.

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

from drogon.

from drogon.

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

from drogon.

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

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

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

from drogon.

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

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

from drogon.

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

from drogon.

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

from drogon.

这是 http的压测

from drogon.

ws 第一次

from drogon.

ws 第二次

from drogon.

ws 第三次

from drogon.

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

from drogon.

from drogon.

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

from drogon.

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

from drogon.

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

from drogon.

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
这个问题还是没办法解决对吧 释放了空指针

from drogon.

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

from drogon.

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

from drogon.