Mailing-List: contact dev-help@qpid.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@qpid.apache.org
Date: Tue, 8 Oct 2013 11:46:42 +0000 (UTC)
From: "Pavel Moravec (JIRA)" <jira@apache.org>
To: dev@qpid.apache.org
Message-ID: <JIRA.12672796.1381232795548.31430.1381232802090@arcas>
In-Reply-To: <JIRA.12672796.1381232795548@arcas>
References: <JIRA.12672796.1381232795548@arcas>
Subject: [jira] [Created] (QPID-5214) [C++ broker] Memory leak in
 legacystore when raising RHM_IORES_ENQCAPTHRESH
MIME-Version: 1.0
Content-Type: text/plain; charset=utf-8
Content-Transfer-Encoding: 7bit

Pavel Moravec created QPID-5214:
-----------------------------------

             Summary: [C++ broker] Memory leak in legacystore when raising RHM_IORES_ENQCAPTHRESH
                 Key: QPID-5214
                 URL: https://issues.apache.org/jira/browse/QPID-5214
             Project: Qpid
          Issue Type: Bug
          Components: C++ Broker
    Affects Versions: 0.24
            Reporter: Pavel Moravec
            Priority: Minor


There is a memory leak when legacystore raises RHM_IORES_ENQCAPTHRESH: "Enqueue capacity threshold exceeded on queue ..". For reproducer, let try to send durable messages to a tiny journal queue in a loop.

Valgrind showed me:

==632== 2,288 (208 direct, 2,080 indirect) bytes in 2 blocks are definitely lost in loss record 115 of 116
==632==    at 0x4A075BC: operator new(unsigned long) (vg_replace_malloc.c:298)
==632==    by 0x60D76AB: mrg::msgstore::MessageStoreImpl::store(qpid::broker::PersistableQueue const*, mrg::msgstore::TxnCtxt*, boost::intrusive_ptr<qpid::broker::PersistableMessage> const&, bool) (in /data_xfs/qpid-trunk/cpp/BLD/src/legacystore.so)
==632==    by 0x60D7165: mrg::msgstore::MessageStoreImpl::enqueue(qpid::broker::TransactionContext*, boost::intrusive_ptr<qpid::broker::PersistableMessage> const&, qpid::broker::PersistableQueue const&) (in /data_xfs/qpid-trunk/cpp/BLD/src/legacystore.so)
==632==    by 0x5023568: qpid::broker::MessageStoreModule::enqueue(qpid::broker::TransactionContext*, boost::intrusive_ptr<qpid::broker::PersistableMessage> const&, qpid::broker::PersistableQueue const&) (in /data_xfs/qpid-trunk/cpp/BLD/src/libqpidbroker.so.2.0.0)
==632==    by 0x4F9BAC8: qpid::broker::Queue::enqueue(qpid::broker::TransactionContext*, qpid::broker::Message&) (in /data_xfs/qpid-trunk/cpp/BLD/src/libqpidbroker.so.2.0.0)

Some further debugging showed the line with "new" call is:

void MessageStoreImpl::store(..
..
        if (queue) {
            boost::intrusive_ptr<DataTokenImpl> dtokp(new DataTokenImpl);
            dtokp->addRef();
..


I tried to fix the leak, but I see nothing wrong in code that could trigger it. As:
1) dtokp is a local variable declared there, while its content is not copied or referenced anywhere later on
2) even catching StoreException and explicitly calling "dtokp->reset(); dtokp = boost::intrusive_ptr<DataTokenImpl>();" does not prevent the mem.leak

What exactly is executed at the time RHM_IORES_ENQCAPTHRESH to be raised within MessageStoreImpl::store call:

1) ./lib/MessageStoreImpl.cpp:
MessageStoreImpl::store

            boost::intrusive_ptr<DataTokenImpl> dtokp(new DataTokenImpl);
            dtokp->addRef();
            dtokp->setSourceMessage(message);
            dtokp->set_external_rid(true);
            dtokp->set_rid(message->getPersistenceId()); // set the messageID into the Journal header (record-id)

            JournalImpl* jc = static_cast<JournalImpl*>(queue->getExternalQueueStore());
            if (txn->getXid().empty()) {
                if (message->isContentReleased()) {
                    jc->enqueue_extern_data_record(size, dtokp.get(), !message->isPersistent());
                } else {
                    jc->enqueue_data_record(&buff[0], size, size, dtokp.get(), !message->isPersistent());
                }


2) enqueue_data_record called from:
./lib/JournalImpl.cpp
JournalImpl::enqueue_data_record

JournalImpl::enqueue_data_record(const void* const data_buff, const size_t tot_data_len,
        const size_t this_data_len, data_tok* dtokp, const bool transient)
{
    handleIoResult(jcntl::enqueue_data_record(data_buff, tot_data_len, this_data_len, dtokp, transient));


3) nested enqueue_data_record called from:
./lib/jrnl/jcntl.cpp:
jcntl::enqueue_data_record(const void* const data_buff, const std::size_t tot_data_len,
        const std::size_t this_data_len, data_tok* dtokp, const bool transient)

        while (handle_aio_wait(_wmgr.enqueue(data_buff, tot_data_len, this_data_len, dtokp, 0, 0, transient, false), r,
                        dtokp)) ;


4) _wmgr.enqueue called from:
./lib/jrnl/wmgr.cpp:
wmgr::enqueue(const void* const data_buff, const std::size_t tot_data_len,
        const std::size_t this_data_len, data_tok* dtokp, const void* const xid_ptr,
        const std::size_t xid_len, const bool transient, const bool external)

    iores res = pre_write_check(WMGR_ENQUEUE, dtokp, xid_len, tot_data_len, external);
    if (res != RHM_IORES_SUCCESS)
        return res;


5) pre_write_check called from ./lib/jrnl/wmgr.cpp as well:

wmgr::pre_write_check(const _op_type op, const data_tok* const dtokp,
        const std::size_t xidsize, const std::size_t dsize, const bool external
        ) const

    if (!_wrfc.is_wr_reset())
    {
        if (!_wrfc.wr_reset())
            return RHM_IORES_FULL;
    }

    // Check status of current page is ok for writing
    if (_page_cb_arr[_pg_index]._state != IN_USE)
    {
        if (_page_cb_arr[_pg_index]._state == UNUSED)
            _page_cb_arr[_pg_index]._state = IN_USE;
    }

    switch (op)
    {
        case WMGR_ENQUEUE:
            {
                // Check for enqueue reaching cutoff threshold
                u_int32_t size_dblks = jrec::size_dblks(enq_rec::rec_size(xidsize, dsize,
                        external));
                if (!_enq_busy && _wrfc.enq_threshold(_cached_offset_dblks + size_dblks))
                    return RHM_IORES_ENQCAPTHRESH;

6) return to 4, there return to 3, and execute: handle_aio_wait:
./lib/jrnl/jcntl.cpp:
jcntl::handle_aio_wait(const iores res, iores& resout, const data_tok* dtp)

(return false)

7) return to 2, there return value used in:
./lib/JournalImpl.cpp:
JournalImpl::handleIoResult(const iores r):
    writeActivityFlag = true;
    switch (r)
    {
        case mrg::journal::RHM_IORES_SUCCESS:
            return;
        case mrg::journal::RHM_IORES_ENQCAPTHRESH:
            {
                std::ostringstream oss;
                oss << "Enqueue capacity threshold exceeded on queue \"" << _jid << "\".";
                log(LOG_WARN, oss.str());
                if (_agent != 0)
                    _agent->raiseEvent(qmf::com::redhat::rhm::store::EventEnqThresholdExceeded(_jid, "Journal enqueue capacity threshold exceeded"),
                                       qpid::management::ManagementAgent::SEV_WARN);
                THROW_STORE_FULL_EXCEPTION(oss.str());
            }


Simply, no data_tok* object used is copied/referenced/whatever in a manner it could prevent freeing the memory.

Anyway, the memory leak is there, the trivial reproducer causes the broker to consume more and more memory (checked by "ps" and "qpid-stat -m" commands outputs).


--
This message was sent by Atlassian JIRA
(v6.1#6144)

---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@qpid.apache.org
For additional commands, e-mail: dev-help@qpid.apache.org