use getLoop()->defer to send data cause memory occupy increase greatly and no reduction after send

[JaderoChan]

[machine translation]

I created the uWS::App in a thread and assigned WebSocket to a global WebSocket pointer when the server was connected, then send a lot of json data via this pointer, and send function was added to the event loop using defer function. The memory occupy increases rapidly when the data is sent, and does not decrease at all after the data is sent. Why is that? Is there a solution?

I need to use multithreading so that the server can not only respond to client requests, but also change the data sent based on the console, so threading is necessary. In multithreading, using send function directly without adding it to the event loop cause to race condition, and now using the event loop leads to a large increase in memory usage, what should I do?)

Here is the my code

    // Init Json document here
    rapidjson::Document d;
    d.Parse(jsonCommandRequest);
    
    // Init uWS::app
    uWS::App *app = nullptr;
    uWS::WebSocket<false, true, SocketData> *ws = nullptr;
    uWS::App::WebSocketBehavior<SocketData> be;
    bool isConnect = false;
    be.open = [&] (auto *_ws) {
        ws = _ws;
        isConnect = true;
    };
    be.message = [&] (auto *ws, auto message, uWS::OpCode code) {
        std::cout << "has message\n";
    };
    be.close = [&] (auto *ws, int code, auto message) {
        std::cout << "close\n";
    };
    std::thread th([&] {
        app = new uWS::App();
        app->ws<SocketData>("/*", std::move(be));
        app->listen(12345, [] (auto *) {});
        app->run();
    });

    while (true) {
        if (isConnect)
            break;
    }
    for (int i = 0; i < 1000000; i++) {
        auto &cmdLine = d[BODY][CMDLINE];
        auto &requestId = d[HEADER][REQUESTID];
        cmdLine.SetString("say hello", d.GetAllocator());
        requestId.SetString(nj::genUuid().c_str(), d.GetAllocator());
        std::string json = domToStr(d);
        app->getLoop()->defer([ws, json] {
            ws->send(json, uWS::OpCode::TEXT);
        });
    }

我在一个线程中创建了uWS::App，并且在服务器被连接时将WebSocket赋值给一个全局的WebSocket指针，而后通过这个指针去大量发送json数据，并且send使用defer被加入事件循环。在数据发送时，内存占用快速增加，并且在数据发送完成后也没有丝毫减少占用。这是为什么？有什么解决办法吗？
（我需要使用多线程使得服务器不只是响应客户端的请求，还可以根据控制台改变发送的数据，所以线程是必要的。而在多线程中，直接使用send而不将其加入事件循环则会导致竞态，而现在使用事件循环又导致内存占用大幅度增加，我到底该怎么办？）

[uNetworkingAB]

If you have issues with glibc default allocator you can look at libtcmalloc

[JaderoChan]

my code not use any data by malloc, only uWS::App and WebsocketBehavior via malloc create

[uNetworkingAB]

Your entire program relies on malloc under the hood and the behavior you post here is how glibc malloc works. I can't say more than what I already said

[JaderoChan]

thanks, i see.

[markmaker]

@JaderoChan,

I think you need some kind of flow control.

uWS::Loop::defer() just adds the whole send() call, including its arguments (buffers), to a queue. Unfortunately, the queue size is unlimited. So chances are that your main thread is much faster than the uWS sending, and it will queue up all or most the send() calls before the first ones are even sent. This means all the memory is quickly allocated, and only very slowly deallocated again.

You need to somehow limit the number of queued calls, and then block the main thread as long as there are too many in the queue. This will stop new memory allocations until prior send() calls are completed and associated memory deallocated. Later, when the main thread resumes, it can reuse these freed memory blocks to store new deferred send() calls. The memory consumption remains more or less constant, and you can choose how much by setting the limit low or high.

How to implement the limit efficiently, I can't just say from the top of my head, I'm also quite new to C++ concurrency. From old learned theory, it should be a counting semaphore.

If you find a solution, please post it here, I might actually need it too 😎

_Mark

[JaderoChan]

but after send() calls in the queue are completed, Shouldn't previously allocated memory be freed?

[markmaker]

Yes, memory is freed on the heap, but that usually just means the blocks are added to a list of free blocks that can be reused. Adjacent blocks may be fused together. In a practical case of small and medium size blocks (which I guess we have here), memory cannot be truly "freed" in the eyes the operating system. A process' memory consumption is basically stuck at its maximum consumption.

How do you measure memory consumption?

Only for very large blocks (usually >= 64kB) there is an optimization using virtual memory mapping. Linux specific background on that: see M_MMAP_THRESHOLD for mallopt(3).

[markmaker]

Similar problem, but getBufferedAmount looks at the socket buffer, not at the defer queue. Plus he needs a multi-threading solution (blocking).

[ddevienne]

Maybe use https://github.com/cameron314/readerwriterqueue. There's a blocking version.
This one is SPSC. Use https://github.com/cameron314/concurrentqueue instead, for more thread.

[markmaker]

@JaderoChan,

I guess you's need to do something like that (starting from your initial example). Note, this is untested, and like I said I'm still new to C++ concurrency.

    // Init Json document here
    rapidjson::Document d;
    d.Parse(jsonCommandRequest);
    
    // Init uWS::app
    uWS::App *app = nullptr;
    uWS::WebSocket<false, true, SocketData> *ws = nullptr;
    uWS::App::WebSocketBehavior<SocketData> be;
    bool isConnect = false;
    be.open = [&] (auto *_ws) {
        ws = _ws;
        isConnect = true;
    };
    be.message = [&] (auto *ws, auto message, uWS::OpCode code) {
        std::cout << "has message\n";
    };
    be.close = [&] (auto *ws, int code, auto message) {
        std::cout << "close\n";
    };
    std::thread th([&] {
        app = new uWS::App();
        app->ws<SocketData>("/*", std::move(be));
        app->listen(12345, [] (auto *) {});
        app->run();
    });

    while (true) {
        if (isConnect)
            break;
    }

    // a handful queued messages should suffice to keep the sender busy
    const int queueLimit = 5;
    std::counting_semaphore<queueLimit> queueSpace(queueLimit);
    
    for (int i = 0; i < 1000000; i++) {
        auto &cmdLine = d[BODY][CMDLINE];
        auto &requestId = d[HEADER][REQUESTID];
        cmdLine.SetString("say hello", d.GetAllocator());
        requestId.SetString(nj::genUuid().c_str(), d.GetAllocator());
        std::string json = domToStr(d);

        // get us a space in the queue (or block)
        queueSpace.acquire(); 

        app->getLoop()->defer([ws, json, &sem] {
            ws->send(json, uWS::OpCode::TEXT);

            // release our space in the queue
            queueSpace.release();
        });
    }

See also:
https://en.cppreference.com/w/cpp/thread/counting_semaphore

Again, this is untested.

[markmaker]

Ah and I think you need to combine this with what @uNetworkingAB said, namely use getBufferedAmount(). Otherwise the bloat / backpressure still builds in the socket send buffer. So you must use drain etc. and it all becomes complicated.

Pity we don't have a completion handler in ws->send() where you could do the queueSpace.release()

[JaderoChan]

[machine translation]
Thanks, I haven't been able to fix the memory loss caused by getLoop()->defer() yet, maybe as @uNetworkingAB said, I need to fumble around with glibc, but I don't want to. It's just that I optimized my code and figured out a way to make the memory footprint less scary (although it was ugly).
谢谢你们，我暂时还无法解决getLoop()->defer()造成内存损耗的问题，或许正如 @uNetworkingAB 所言，我需要去摸索一下glibc，但是我不想。只不过我优化了我的代码并想出了一个办法可以让内存的占用率不那么高的可怕（虽然这办法有些丑陋）

[uNetworkingAB]

A process' memory consumption is basically stuck at its maximum consumption.

Yes glibc allocator works like that but libtcmalloc and similar uses mmap for everything so you get back to zero once everything is freed

[markmaker]

Very interesting.