QSerialPort内存不断增长现象复现

利用QSerialPort开启串口通信。

int main(int argc, char *argv[])
{
    QApplication a(argc, argv);

    QSerialPort* serialPort = new QSerialPort;
    serialPort->setPortName("COM3");
    if (!serialPort->open(QIODevice::ReadWrite))
    {
        qDebug() << "Open error";
    }
    //打开成功,设置波特率和读写方向
    serialPort->setBaudRate(static_cast<QSerialPort::BaudRate>(115200), QSerialPort::AllDirections);

    return a.exec();
}

使用虚拟串口进行串口间的配对。

利用串口调试工具进行模拟通信。串口调试助手每隔20ms像Demo程序发送一段数据。

30s后内存从4.6MB涨到5.0MB。

QSerialPort不断增长原因

先说QSerialPort内存泄漏原因，是Qt实现readyRead和bytesWritten的信号导致。

原因剖析

从源码的角度分析QSerialPort工作的原理。

// 打开串口
bool QSerialPortPrivate::open(QIODevice::OpenMode mode)
{
    DWORD desiredAccess = 0;

    if (mode & QIODevice::ReadOnly)
        desiredAccess |= GENERIC_READ;
    if (mode & QIODevice::WriteOnly)
        desiredAccess |= GENERIC_WRITE;

    handle = ::CreateFile(reinterpret_cast<const wchar_t*>(systemLocation.utf16()),
                              desiredAccess, 0, nullptr, OPEN_EXISTING, FILE_FLAG_OVERLAPPED, nullptr);

    if (handle == INVALID_HANDLE_VALUE) {
        setError(getSystemError());
        return false;
    }

    if (initialize(mode))
        return true;

    ::CloseHandle(handle);
    return false;
}

// 初始化串口
inline bool QSerialPortPrivate::initialize(QIODevice::OpenMode mode)
{
    Q_Q(QSerialPort);

    DCB dcb;
    if (!getDcb(&dcb))
        return false;

    restoredDcb = dcb;

    qt_set_common_props(&dcb);
    qt_set_baudrate(&dcb, inputBaudRate);
    qt_set_databits(&dcb, dataBits);
    qt_set_parity(&dcb, parity);
    qt_set_stopbits(&dcb, stopBits);
    qt_set_flowcontrol(&dcb, flowControl);

    if (!setDcb(&dcb))
        return false;

    if (!::GetCommTimeouts(handle, &restoredCommTimeouts)) {
        setError(getSystemError());
        return false;
    }

    ::ZeroMemory(&currentCommTimeouts, sizeof(currentCommTimeouts));
    currentCommTimeouts.ReadIntervalTimeout = MAXDWORD;

    if (!::SetCommTimeouts(handle, &currentCommTimeouts)) {
        setError(getSystemError());
        return false;
    }

    const DWORD eventMask = (mode & QIODevice::ReadOnly) ? EV_RXCHAR : 0;
    if (!::SetCommMask(handle, eventMask)) {
        setError(getSystemError());
        return false;
    }
   
    // 内存泄漏关键代码
    notifier = new QWinOverlappedIoNotifier(q);
    QObjectPrivate::connect(notifier, &QWinOverlappedIoNotifier::notified,
               this, &QSerialPortPrivate::_q_notified);
    notifier->setHandle(handle);
    notifier->setEnabled(true);

    if ((eventMask & EV_RXCHAR) && !startAsyncCommunication()) {
        delete notifier;
        notifier = nullptr;
        return false;
    }

    return true;
}

下面查看QWinOverlappedIoNotifier的源码：

QWinOverlappedIoNotifier::QWinOverlappedIoNotifier(QObject *parent)
    : QObject(*new QWinOverlappedIoNotifierPrivate, parent)
{
    Q_D(QWinOverlappedIoNotifier);
    WaitForSingleObject(d->iocpInstanceLock, INFINITE);
    if (!d->iocp)
        d->iocp = new QWinIoCompletionPort;
    d->iocpInstanceRefCount++;
    ReleaseMutex(d->iocpInstanceLock);

    d->hSemaphore = CreateSemaphore(NULL, 0, 255, NULL);
    d->hResultsMutex = CreateMutex(NULL, FALSE, NULL);
    connect(this, SIGNAL(_q_notify()), this, SLOT(_q_notified()), Qt::QueuedConnection); // 注意连接方式
}

构造QWinOverlappedIoNotifier时创建QWinIoCompletionPort，下面是QWinIoCompletionPort的源码：

class QWinIoCompletionPort : protected QThread
{
public:
    QWinIoCompletionPort()
        : finishThreadKey(reinterpret_cast<ULONG_PTR>(this)),
          drainQueueKey(reinterpret_cast<ULONG_PTR>(this + 1))
    {
        setObjectName(QLatin1String("I/O completion port thread"));
        HANDLE hIOCP = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0);
        if (!hIOCP) {
            qErrnoWarning("CreateIoCompletionPort failed.");
            return;
        }
        hPort = hIOCP;
        hQueueDrainedEvent = CreateEvent(NULL, TRUE, FALSE, NULL);
        if (!hQueueDrainedEvent) {
            qErrnoWarning("CreateEvent failed.");
            return;
        }
    }

    ~QWinIoCompletionPort()
    {
        PostQueuedCompletionStatus(hPort, 0, finishThreadKey, NULL);
        QThread::wait();
        CloseHandle(hPort);
        CloseHandle(hQueueDrainedEvent);
    }

    void registerNotifier(QWinOverlappedIoNotifierPrivate *notifier)
    {
        const HANDLE hHandle = notifier->hHandle;
        HANDLE hIOCP = CreateIoCompletionPort(hHandle, hPort,
                                              reinterpret_cast<ULONG_PTR>(notifier), 0);
        if (!hIOCP) {
            qErrnoWarning("Can't associate file handle %x with I/O completion port.", hHandle);
            return;
        }
        mutex.lock();
        notifiers += notifier;
        mutex.unlock();
        if (!QThread::isRunning())
            QThread::start();
    }

    void unregisterNotifier(QWinOverlappedIoNotifierPrivate *notifier)
    {
        mutex.lock();
        notifiers.remove(notifier);
        mutex.unlock();
    }

    void drainQueue()
    {
        QMutexLocker locker(&drainQueueMutex);
        ResetEvent(hQueueDrainedEvent);
        PostQueuedCompletionStatus(hPort, 0, drainQueueKey, NULL);
        WaitForSingleObject(hQueueDrainedEvent, INFINITE);
    }

    using QThread::isRunning;

protected:
    void run()
    {
        DWORD dwBytesRead = 0;
        ULONG_PTR pulCompletionKey = 0;
        OVERLAPPED *overlapped = nullptr;
        DWORD msecs = INFINITE;

        forever {
            BOOL success = GetQueuedCompletionStatus(hPort,
                                                     &dwBytesRead,
                                                     &pulCompletionKey,
                                                     &overlapped,
                                                     msecs);

            DWORD errorCode = success ? ERROR_SUCCESS : GetLastError();
            if (!success && !overlapped) {
                if (!msecs) {
                    // Time out in drain mode. The completion status queue is empty.
                    msecs = INFINITE;
                    SetEvent(hQueueDrainedEvent);
                    continue;
                }
                qErrnoWarning(errorCode, "GetQueuedCompletionStatus failed.");
                return;
            }

            if (pulCompletionKey == finishThreadKey)
                return;
            if (pulCompletionKey == drainQueueKey) {
                // Enter drain mode.
                Q_ASSERT(msecs == INFINITE);
                msecs = 0;
                continue;
            }

            QWinOverlappedIoNotifierPrivate *notifier
                    = reinterpret_cast<QWinOverlappedIoNotifierPrivate *>(pulCompletionKey);
            mutex.lock();
            if (notifiers.contains(notifier))
                notifier->notify(dwBytesRead, errorCode, overlapped);
            mutex.unlock();
        }
    }

private:
    const ULONG_PTR finishThreadKey;
    const ULONG_PTR drainQueueKey;
    HANDLE hPort = INVALID_HANDLE_VALUE;
    QSet<QWinOverlappedIoNotifierPrivate *> notifiers;
    QMutex mutex;
    QMutex drainQueueMutex;
    HANDLE hQueueDrainedEvent = nullptr;
};

创建QSerialPort时候就会开始QWinIoCompletionPort的线程。用实现readyRead和bytesWritten的信号。
所以当连接的串口不断向上发送数据时候，waitForReadyRead调用waitForNotify调用WaitForSingleObject 会导致GetQueuedCompletionStatus会被不断触发，因为数据不断触发队列状态。而导致notifier->notify(dwBytesRead, errorCode, overlapped); 这段代码不断被调用。再来看下notifier->notify的实现。

void QWinOverlappedIoNotifierPrivate::notify(DWORD numberOfBytes, DWORD errorCode,
        OVERLAPPED *overlapped)
{
    Q_Q(QWinOverlappedIoNotifier);
    WaitForSingleObject(hResultsMutex, INFINITE);
    results.enqueue(IOResult(numberOfBytes, errorCode, overlapped));
    ReleaseMutex(hResultsMutex);
    ReleaseSemaphore(hSemaphore, 1, NULL);
    if (!waiting)
        emit q->_q_notify(); // 关键信号，导致内存线程的诱因
}

由于该信号的连接方式为connect(this, SIGNAL(_q_notify()), this, SLOT(_q_notified()), Qt::QueuedConnection); 的队列形式，所以信号队列会不断膨胀，在消费效率跟不上的时候就会导致内存不断增长。

Qt QueuedConnection的调用逻辑

首先Qt的信号槽连接最常用的连接方式为DirectConnection和QueuedConnection。DirectConnection是在信号被调用时候，马上进行调用槽函数。而QueuedConnection则是利用eventLoop的事件循环机制，将其加入连接对象线程的事件循环里，等待连接对象线程进行事件调度时候调用相应的槽函数。所以跨线程信号槽机制是依赖eventLoop的，也就不能使用std::thread进行跨线程的槽调用，而一定要在QThread里才能进行跨线程槽调用。

if (connectionType == Qt::DirectConnection) {
    if (callFunction) {
        callFunction(object, QMetaObject::InvokeMetaMethod, idx_relative, param);
        return true;
    } else {
        return QMetaObject::metacall(object, QMetaObject::InvokeMetaMethod, idx_relative + idx_offset, param) < 0;
    }
} else if (connectionType == Qt::QueuedConnection) {
    if (returnValue.data()) {
        qWarning("QMetaMethod::invoke: Unable to invoke methods with return values in "
                 "queued connections");
        return false;
    }

    int nargs = 1; // include return type
    void **args = (void **) malloc(paramCount * sizeof(void *));
    Q_CHECK_PTR(args);
    int *types = (int *) malloc(paramCount * sizeof(int));
    Q_CHECK_PTR(types);
    types[0] = 0; // return type
    args[0] = 0;

    for (int i = 1; i < paramCount; ++i) {
        types[i] = QMetaType::type(typeNames[i]);
        if (types[i] == QMetaType::UnknownType && param[i]) {
            // Try to register the type and try again before reporting an error.
            int index = nargs - 1;
            void *argv[] = { &types[i], &index };
            QMetaObject::metacall(object, QMetaObject::RegisterMethodArgumentMetaType,
                                  idx_relative + idx_offset, argv);
            if (types[i] == -1) {
                qWarning("QMetaMethod::invoke: Unable to handle unregistered datatype '%s'",
                        typeNames[i]);
                for (int x = 1; x < i; ++x) {
                    if (types[x] && args[x])
                        QMetaType::destroy(types[x], args[x]);
                }
                free(types);
                free(args);
                return false;
            }
        }
        if (types[i] != QMetaType::UnknownType) {
            args[i] = QMetaType::create(types[i], param[i]);
            ++nargs;
        }
    }

    QCoreApplication::postEvent(object, new QMetaCallEvent(idx_offset, idx_relative, callFunction,
                                                    0, -1, nargs, types, args));
}

postEvent将QueuedConnection触发的信号加入到postEventList中。

void QCoreApplication::postEvent(QObject *receiver, QEvent *event, int priority)
{
    Q_TRACE_SCOPE(QCoreApplication_postEvent, receiver, event, event->type());

    if (receiver == 0) {
        qWarning("QCoreApplication::postEvent: Unexpected null receiver");
        delete event;
        return;
    }

    QThreadData * volatile * pdata = &receiver->d_func()->threadData;
    QThreadData *data = *pdata;
    if (!data) {
        // posting during destruction? just delete the event to prevent a leak
        delete event;
        return;
    }

    // lock the post event mutex
    data->postEventList.mutex.lock();

    // if object has moved to another thread, follow it
    while (data != *pdata) {
        data->postEventList.mutex.unlock();

        data = *pdata;
        if (!data) {
            // posting during destruction? just delete the event to prevent a leak
            delete event;
            return;
        }

        data->postEventList.mutex.lock();
    }

    QMutexUnlocker locker(&data->postEventList.mutex);

    // if this is one of the compressible events, do compression
    if (receiver->d_func()->postedEvents
        && self && self->compressEvent(event, receiver, &data->postEventList)) {
        Q_TRACE(QCoreApplication_postEvent_event_compressed, receiver, event);
        return;
    }

    if (event->type() == QEvent::DeferredDelete)
        receiver->d_ptr->deleteLaterCalled = true;

    if (event->type() == QEvent::DeferredDelete && data == QThreadData::current()) {
        // remember the current running eventloop for DeferredDelete
        // events posted in the receiver's thread.

        // Events sent by non-Qt event handlers (such as glib) may not
        // have the scopeLevel set correctly. The scope level makes sure that
        // code like this:
        //     foo->deleteLater();
        //     qApp->processEvents(); // without passing QEvent::DeferredDelete
        // will not cause "foo" to be deleted before returning to the event loop.

        // If the scope level is 0 while loopLevel != 0, we are called from a
        // non-conformant code path, and our best guess is that the scope level
        // should be 1. (Loop level 0 is special: it means that no event loops
        // are running.)
        int loopLevel = data->loopLevel;
        int scopeLevel = data->scopeLevel;
        if (scopeLevel == 0 && loopLevel != 0)
            scopeLevel = 1;
        static_cast<QDeferredDeleteEvent *>(event)->level = loopLevel + scopeLevel;
    }

    // delete the event on exceptions to protect against memory leaks till the event is
    // properly owned in the postEventList
    QScopedPointer<QEvent> eventDeleter(event);
    Q_TRACE(QCoreApplication_postEvent_event_posted, receiver, event, event->type());
    data->postEventList.addEvent(QPostEvent(receiver, event, priority)); // 事件队列容器
    eventDeleter.take();
    event->posted = true;
    ++receiver->d_func()->postedEvents;
    data->canWait = false;
    locker.unlock();

    QAbstractEventDispatcher* dispatcher = data->eventDispatcher.loadAcquire();
    if (dispatcher)
        dispatcher->wakeUp();
}

QPostEventList 是继承于QVector的类，代码如下：

// This class holds the list of posted events.
//  The list has to be kept sorted by priority
class QPostEventList : public QVector<QPostEvent>
{
public:
    // recursion == recursion count for sendPostedEvents()
    int recursion;

    // sendOffset == the current event to start sending
    int startOffset;
    // insertionOffset == set by sendPostedEvents to tell postEvent() where to start insertions
    int insertionOffset;

    QMutex mutex;

    inline QPostEventList()
        : QVector<QPostEvent>(), recursion(0), startOffset(0), insertionOffset(0)
    { }

    void addEvent(const QPostEvent &ev) {
        int priority = ev.priority;
        if (isEmpty() ||
            constLast().priority >= priority ||
            insertionOffset >= size()) {
            // optimization: we can simply append if the last event in
            // the queue has higher or equal priority
            append(ev);
        } else {
            // insert event in descending priority order, using upper
            // bound for a given priority (to ensure proper ordering
            // of events with the same priority)
            QPostEventList::iterator at = std::upper_bound(begin() + insertionOffset, end(), ev);
            insert(at, ev);
        }
    }
private:
    //hides because they do not keep that list sorted. addEvent must be used
    using QVector<QPostEvent>::append;
    using QVector<QPostEvent>::insert;
};

QWinIoCompletionPort的线程实现readyRead和bytesWritten的信号

/*
 * Note: This function runs in the I/O completion port thread.
 */
void QWinOverlappedIoNotifierPrivate::notify(DWORD numberOfBytes, DWORD errorCode,
        OVERLAPPED *overlapped)
{
    Q_Q(QWinOverlappedIoNotifier);
    WaitForSingleObject(hResultsMutex, INFINITE);
    results.enqueue(IOResult(numberOfBytes, errorCode, overlapped));
    ReleaseMutex(hResultsMutex);
    ReleaseSemaphore(hSemaphore, 1, NULL);
    if (!waiting)
        emit q->_q_notify();
}

void QWinOverlappedIoNotifierPrivate::_q_notified()
{
    if (WaitForSingleObject(hSemaphore, 0) == WAIT_OBJECT_0)
        dispatchNextIoResult();
}

OVERLAPPED *QWinOverlappedIoNotifierPrivate::dispatchNextIoResult()
{
    Q_Q(QWinOverlappedIoNotifier);
    WaitForSingleObject(hResultsMutex, INFINITE);
    IOResult ioresult = results.dequeue();
    ReleaseMutex(hResultsMutex);
    emit q->notified(ioresult.numberOfBytes, ioresult.errorCode, ioresult.overlapped);
    return ioresult.overlapped;
}

在QSerialPort初始化时候调用了QObjectPrivate::connect(notifier, &QWinOverlappedIoNotifier::notified, this, &QSerialPortPrivate::_q_notified); 的连接。

void QSerialPortPrivate::_q_notified(DWORD numberOfBytes, DWORD errorCode, OVERLAPPED *overlapped)
{
    const QSerialPortErrorInfo error = getSystemError(errorCode);
    if (error.errorCode != QSerialPort::NoError) {
        setError(error);
        return;
    }

    if (overlapped == &communicationOverlapped)
        completeAsyncCommunication(numberOfBytes);
    else if (overlapped == &readCompletionOverlapped)
        completeAsyncRead(numberOfBytes);
    else if (overlapped == &writeCompletionOverlapped)
        completeAsyncWrite(numberOfBytes);
    else
        Q_ASSERT(!"Unknown OVERLAPPED activated");
}

bool QSerialPortPrivate::completeAsyncCommunication(qint64 bytesTransferred)
{
    communicationStarted = false;

    if (bytesTransferred == qint64(-1))
        return false;

    return startAsyncRead();
}

bool QSerialPortPrivate::completeAsyncRead(qint64 bytesTransferred)
{
    if (bytesTransferred == qint64(-1)) {
        readStarted = false;
        return false;
    }
    if (bytesTransferred > 0)
        buffer.append(readChunkBuffer.constData(), bytesTransferred);

    readStarted = false;

    bool result = true;
    if (bytesTransferred == QSERIALPORT_BUFFERSIZE
            || queuedBytesCount(QSerialPort::Input) > 0) {
        result = startAsyncRead();
    } else {
        result = startAsyncCommunication();
    }

    if (bytesTransferred > 0)
        emitReadyRead();

    return result;
}

bool QSerialPortPrivate::completeAsyncWrite(qint64 bytesTransferred)
{
    Q_Q(QSerialPort);

    if (writeStarted) {
        if (bytesTransferred == qint64(-1)) {
            writeChunkBuffer.clear();
            writeStarted = false;
            return false;
        }
        Q_ASSERT(bytesTransferred == writeChunkBuffer.size());
        writeChunkBuffer.clear();
        emit q->bytesWritten(bytesTransferred);
        writeStarted = false;
    }

    return _q_startAsyncWrite();
}

结论

1. QSerialPort内部存在QThread线程以及QTimer的使用，所以QSerialPort只能在QThread的带QEventLoop的线程内运行，不然部分功能将会失效。
2. QSerialPort使用时候，需要注意通信间隔，不然会导致内存不断增长。注意不是内存泄漏，是因为QPostEventList的不断增长导致vector的长度不断增长而导致内存增加。
3. 解决内存增长的问题，可以从通信间隔减少吞吐量的方式。也可以在通信过程中加入在使用QSerialPort的线程手动调用processEvents，让事件循环对postEvent进行的处理，将QPostEventList里面的事件队列处理掉，从而减少内存的增加。