Memory leak when mixing asio connection and object_t，may caused by dbus outgoing signal

[gyuqian]

Hi, All

I found that mixing object_t to create objects in asio connection may cause a serious memory leak.

1. How to reproduce my problem？

Add the following function to asio-example.cpp. I reuse the Calculator class.

void creat_calculator(boost::asio::io_context& io, std::shared_ptr<sdbusplus::asio::connection>& conn)
{
    static long suffix = 0;
    static boost::asio::steady_timer timer(io);
    if (std::filesystem::exists("./creat_flag"))
    {
        boost::asio::post(io, [&](){creat_calculator(io, conn);});
        if (suffix++ % 1000 == 0)
        {
            std::cout << "suffix = " << suffix << std::endl;
        }
        // Create a calculator object at /net/poettering/calculator
        Calculator c1{*conn, std::string{"/net/poettering/calculator"}.append(std::to_string(suffix)).c_str()};
    }
    else
    {
        timer.expires_after(std::chrono::seconds(5));
        timer.async_wait([&](const boost::system::error_code& ){
            // malloc_stats();
            creat_calculator(io, conn);
        });
    }
}

If I use connection to create a dbus object and start running it using io.run(). If I start a new io task and use object_t to create objects at a very high frequency, I can soon observe an abnormal increase in Rss in the process's /proc/$PID/smaps. Next, I stop using object_t to create objects, and Rss does not return to normal levels quickly, but remains at a relatively high abnormal level. Then, I send a method_call to this process, such as Ping(), and I will receive a response after a long time.

At the same time, if I use busctl monitor to detect the InterfaceAdded signal issued when creating an object_t object, I will observe that before sending a method_call to the process, I cannot monitor the expected InterfaceAdded signal, but after sending Ping(), I monitor the InterfaceAdded signals that should have been issued when creating the object_t object.

2. Possible reasons I found.

In the read_wait of connection, we originally wanted to monitor POLLIN, POLLOUT and timeout. If the bus is idle for long enough, get_events() can only return POLLIN, so we only monitor the EPOLLIN event of bus input_fd, and there will be no timeout. This means that we only execute process() when there is an incoming message.

Unfortunately, the InterfaceAdded sent by object_t is only an outgoing message, and it does not expect a method_return like async_method_call. At the same time, I found in the source code of sd-bus that sd_bus_send will cache outgoing messages to wqueue when there are too many messages, and sd_bus_process is required to continue to send these outgoing messages. This means that if there is no incoming message, these messages in the wqueue queue will not be sent, and at most 384*1024 messages will be accumulated in the wqueue queue and cannot be released; at the same time, if any incoming message is received, process() will start to execute, and then the messages in the wqueue queue will be properly processed.

I think we may need a timer to periodically check whether there is data to be processed in rqueue and wqueue. Considering that adding a timer may affect the performance of the program, I am not sure whether this timer should be added. I would like to seek your opinions.

[gyuqian]

Add some of my test records：

1. Using the memory information obtained by malloc_info, the process really uses a lot of memory.

suffix = 13443000
suffix = 13444000
suffix = 13445000
suffix = 13446000
suffix = 13447000
suffix = 13448000
Arena 0:
system bytes     =  612405248
in use bytes     =  612309280
Total (incl. mmap):
system bytes     =  615878656
in use bytes     =  615782688
max mmap regions =          2
max mmap bytes   =    3473408
Arena 0:
system bytes     =  612405248
in use bytes     =  612309280
Total (incl. mmap):
system bytes     =  615878656
in use bytes     =  615782688
max mmap regions =          2
max mmap bytes   =    3473408

2. Send a method_call to the process, it takes a long time to receive a reply, at the same time the memory usage of the process is reduced, and the wqueue has been sent out.

Although the Rss of the process is still relatively large, this is another bug.

[leiyu-bytedance]

This is the issue described in https://gerrit.openbmc.org/c/openbmc/sdbusplus/+/68860 and async match is proposed to fix the issue: we should create sdbusplus::asio::match instead of sdbusplus::bus::match_t for asio servers to avoid such issue.

[gyuqian]

This is the issue described in https://gerrit.openbmc.org/c/openbmc/sdbusplus/+/68860 and async match is proposed to fix the issue: we should create sdbusplus::asio::match instead of sdbusplus::bus::match_t for asio servers to avoid such issue.

Yes, sd_bus_call is used in match_t, which causes the event that EPOLLIN needs to monitor to be read into rqueue in advance by bus_poll, and asio wait read cannot read the message, and bus_process is not executed.

But the problem I encountered today is actually more complicated than the conflict between bus_poll and EPOLL. In fact, sd_bus_send executed when object_t emits a signal will not read the message into rqueue. sd_bus_send only caches the message in wqueue when there is a message in the message queue.

sd_bus_send:

        if (IN_SET(bus->state, BUS_RUNNING, BUS_HELLO) && bus->wqueue_size <= 0) {
                size_t idx = 0;
                r = bus_write_message(bus, m, &idx);

        } else {
                /* Just append it to the queue. */

                if (bus->wqueue_size >= BUS_WQUEUE_MAX)
                        return -ENOBUFS;

                if (!GREEDY_REALLOC(bus->wqueue, bus->wqueue_size + 1))
                        return -ENOMEM;

                bus->wqueue[bus->wqueue_size++] = bus_message_ref_queued(m, bus);
        }

If you want to really send these messages in wqueue out again, you need to execute bus_process or send method_call outward. Now asio::connection does not have a good event trigger to execute bus_process. If dbus has been idle for a long time before sending the signal, read_wait can only set EPOLLIN to be monitored. Coincidentally, dbus signals are broadcast and will not trigger EPOLLIN in this process.

static int process_running(sd_bus *bus, sd_bus_message **ret) {
……
        r = dispatch_wqueue(bus);
        if (r != 0)
                goto null_message;

        r = dispatch_track(bus);

[gyuqian]

I checked the source code of sd_bus_attach_event and found that sd_bus_attach_event not only has io event source, but also sets time event source. So why is it that only io event source is monitored in asio connection?

_public_ int sd_bus_attach_event(sd_bus *bus, sd_event *event, int priority) {
……

        bus->event_priority = priority;

        r = sd_event_add_time(bus->event, &bus->time_event_source, CLOCK_MONOTONIC, 0, 0, time_callback, bus);
        if (r < 0)
                goto fail;
……

        r = bus_attach_io_events(bus);
        if (r < 0)
                goto fail;

        r = bus_attach_inotify_event(bus);
        if (r < 0)
                goto fail;

……
}

[leiyu-bytedance]

See the discussion here:
https://lore.kernel.org/openbmc/CACWQX81w1ZzXx64r0UcxpZrNyWB7HudMzZpSttqEeNav98JC9g@mail.gmail.com/

As mentioned by @edtanous , it's not appropriate to mix the asio and sync sd_bus_call, so you could manually avoid the usage of mixing asio and sync calls.

And it's common usage of mixing match_t with asio, so asio::match_t is introduced to handle such case.

[gyuqian]

Oh, you may have misunderstood the problem I encountered. I did not mix sd_bus_call and sd_bus_call_async. I just created and destroyed object_t objects at a very high frequency, and object_t only sent an InterfaceAdded singal. The call stack is as follows

object::object
--->emit_object_added
   --->sd_bus_emit_object_added
      --->sd_bus_send

In this process, there is neither synchronous sd_bus_call nor asynchronous sd_bus_call_async. Frequently sending InterfaceAdded singals in the process of using asio::connection will encounter problems. Furthermore, not only InterfaceAdded singals, but also any dbus singals will encounter problems, but object_t will send InterfaceAdded singals by default.

I have completed additional tests and encountered the same problem when using sdbusplus::asio::dbus_interface instead of object_t.

void creat_object(boost::asio::io_context& io, std::shared_ptr<sdbusplus::asio::connection>& conn)
{
    static long suffix = 0;
    static boost::asio::steady_timer timer(io);
    if (std::filesystem::exists("./creat_flag"))
    {
        boost::asio::post(io, [&](){creat_object(io, conn);});
        if (suffix % 1000 == 0)
        {
            std::cout << "suffix = " << suffix << std::endl;
        }

        std::shared_ptr<sdbusplus::asio::dbus_interface> iface =
        std::make_shared<sdbusplus::asio::dbus_interface>(conn, std::string{"/xyz/openbmc_project/test"}.append(std::to_string(suffix).c_str()),  "xyz.openbmc_project.test");
        iface->register_property("int", 33, sdbusplus::asio::PropertyPermission::readWrite);
        iface->initialize();
    }
    else
    {
        timer.expires_after(std::chrono::seconds(5));
        timer.async_wait([&](const boost::system::error_code& ){
            malloc_stats();
            std::cout << "creat falg to creat obj" << std::endl;
            creat_object(io, conn);
        });
    }
}

sdbusplus::asio::dbus_interface is not very object-oriented, so this usage is not common. However, it is very common to mix object_t with asio::connection. For example, the session object of netipmid inherits object_t. If the session object is created/destroyed frequently, this problem will occur. It is easy to create/destroy sessions frequently, just make frequent connection requests to netipmid.

[edtanous]

The above examples confuse me quite a bit. You're creating 13,443,000 of objects in a (functionally) recursive unending loop, then expecting messages to be processed at the same time? I suspect that there's a misunderstanding about what boost::asio::post does. It simply enqueues the handler on the back of the queue. If there are no outstanding tasks (as there might be in a dbus connection) the method is called immediately.

Yes, DBus requires memory for storage of each object. 13 million objects on the same parent path is going to cause introspect calls from the mapper to have problems, so I'm not really sure what the real use case is for that code as written.

You also check for creat_flag, but never write it?

If you have examples that reproduce an issue, can you please put them in gerrit so we know they compile. Your above examples have function calls like malloc_stats() that I suspect don't exist. There's a lot of mention of match expressions, but none in the example.

[gyuqian]

Hi,
This is a damoo that can cause problems for me. Gerrit is a little difficult for me to access, so I put the code here

#include "sdbusplus/message.hpp"
#include "sdbusplus/message/native_types.hpp"
#include <boost/asio/detached.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/spawn.hpp>
#include <sdbusplus/asio/connection.hpp>
#include <sdbusplus/asio/object_server.hpp>
#include <sdbusplus/bus.hpp>
#include <sdbusplus/server.hpp>
#include <sdbusplus/bus/match.hpp>
#include <boost/asio/steady_timer.hpp>

#include <iostream>

void creat_object(boost::asio::io_context& io, std::shared_ptr<sdbusplus::asio::connection>& conn)
{
    static long suffix = 0;

    if (suffix >= 5 * 1024)
    {
        int events = conn->get_events();
        std::cout << "[Server] " << "suffix = " << suffix << std::endl;

        std::cout << "[Server] " << "POLLIN is " << ((events & POLLIN) ? "set" : "not set")
                << ", POLLOUT is " << ((events & POLLOUT) ? "set" : "not set")
                << std::endl;

        std::cout << "[Server] remain io: " << io.poll() << std::endl;

        std::cout << "[Server] " << "emits InterfacesAdded done." << std::endl;
        return;
    }

    // High frequency emits InterfaceAdded signal
    boost::asio::post(io, [&](){creat_object(io, conn);});

    std::string path = "/xyz/openbmc_project/test_" + std::to_string(suffix++);

    std::shared_ptr<sdbusplus::asio::dbus_interface> iface =
    std::make_shared<sdbusplus::asio::dbus_interface>(conn, path.c_str(), "xyz.openbmc_project.test");

    // emits InterfaceAdded signal
    // std::cout << "[Server] " << "emits " << path << " InterfacesAdded." << std::endl;
    iface->initialize();
    // std::cout << "[Server] " <<  path << " is destoried." << std::endl;
}

int server()
{
    // setup connection to dbus
    boost::asio::io_context io;
    auto conn = std::make_shared<sdbusplus::asio::connection>(io);

    // test object server
    conn->request_name("xyz.openbmc_project.asio-test");
    auto server = sdbusplus::asio::object_server(conn);

    creat_object(io, conn);

    io.run();

    return 0;
}

int client()
{
    // setup connection to dbus
    boost::asio::io_context io;
    auto conn = std::make_shared<sdbusplus::asio::connection>(io);

    using namespace sdbusplus::bus::match::rules;
    sdbusplus::bus::match_t match = sdbusplus::bus::match_t(
        *conn, interfacesAdded(),
        [](sdbusplus::message_t& m)
        {
            sdbusplus::message::object_path objPath;
            m.read(objPath);

            std::cout << "[Clinet] " << "InterfaceAdded: " << objPath.str << std::endl;
        });

    (void)boost::asio::spawn(
        io, [conn, &io](boost::asio::yield_context yield) {
            boost::asio::steady_timer timer(io);
            boost::system::error_code ec;

            timer.expires_after(std::chrono::seconds(60));
            timer.async_wait(yield[ec]);
            if (ec)
            {
                std::cout << "[Clinet] " << "timer error" << std::endl;
            }

            std::cout << "[Clinet] " << "async call server" << std::endl;
            conn->yield_method_call<void>(yield, ec, "xyz.openbmc_project.asio-test", "/",
                                          "org.freedesktop.DBus.Peer", "Ping");

            std::cout << "[Clinet] " << "async call server done" << std::endl;
        },
        boost::asio::detached);

    io.run();

    return 0;
}

int main(int /* argc */, const char** /* argv[] */)
{
    int pid = fork();
    if (pid == 0)
    {
        return client();
    }
    else if (pid > 0)
    {
        return server();
    }
    return pid;
}

When executing this damo, at the beginning, the Client can match the InterfaceAdded signal as expected, but after a while, the Client can no longer receive the InterfaceAdded signal.

At this time, the Server is already printing debug information, all messages have been sent, io.poll returns the number of IO tasks ready to be executed as 0, and the Server has no IO tasks ready to be executed.

In Clinet, I set a timer, and after one minute, I sent a Ping to the Server. At this time, the Client received the remaining InetrfacesAdded signals

---

Note that the interfaces I created are all local variables. These objects will be destructed after the io ends, so there will be no other object paths except the root path. Introspect is theoretically unaffected. At the same time, the interface has been destructed, and the Client received the InetrfacesAdded signal sent by the interface before it was destructed.

The example I provided before was not rigorous enough and created too many objects. Now I limit the number of objects created, just to illustrate the problem of delayed throwing of key dbus signals.

There are not many scenarios where D-Bus signals are sent frequently. Although mixed use of asio::connection and object_t is a common scenario, frequent creation of object_t objects is not a common scenario. Therefore, this is not a common usage scenario.
However, once this scenario occurs, a large amount of memory will be kept in the dbus wqueue, and if the queue is full, the outgoing of dbus will be in an unavailable state, and method_call and singal can no longer be sent until an incoming message is received.

malloc_stat is an interface provided by libc, which is used to count the memory information of the process. You can get the API description here: https://man7.org/linux/man-pages/man3/malloc_stats.3.html

Thinks.

[zhangjian3032]

I think I understand what you mean. Simply put:

1. When there are too many messages and a write fails (during my actual debugging, the sendmsg return value was 0 , anyway), the message is added to the wqueue.

2. As long as there is data in the wqueue, all subsequent send operations will be directly queued into the wqueue.

3. If there are no events, there is no way to dispatch the wqueue.

[zhangjian3032]

This is essentially the same problem as when mixing synchronous and asynchronous logic.

In our poll-driven model, we seem unable to dispatch the wqueue or rqueue if new data is unexpectedly added while there are no active events.

Specifically:

For sending: the send operation is not fully completed (partial write).
For reading: the event is hijacked by the synchronous poll call.

[gyuqian]

@zhangjian3032 Awesome, you described my problem perfectly.