UDP: don't disable read notification unless we have a datagram

The current logic that we will disable the read notification if we
have any data at all doesn't make sense for users who use the
receiveDatagram functionality, since they will not make any calls
that trigger the read notifier to be re-enabled unless there is a
datagram ready for us to hand back.

Fixes: QTBUG-105871
Pick-to: 6.7 6.6 6.5
Change-Id: I0a1f1f8babb037d923d1124c2603b1cb466cfe18
Reviewed-by: Thiago Macieira <thiago.macieira@intel.com>

src/network/socket/qabstractsocket.cpp

@@ -637,11 +637,15 @@ bool QAbstractSocketPrivate::canReadNotification()
             return !q->isReadable();
         }
     } else {
-        if (hasPendingData) {
+        const bool isUdpSocket = (socketType == QAbstractSocket::UdpSocket);
+        if (hasPendingData && (!isUdpSocket || hasPendingDatagram)) {
             socketEngine->setReadNotificationEnabled(false);
             return true;
         }
-        hasPendingData = true;
+        if (!isUdpSocket || socketEngine->hasPendingDatagrams()) {
+            hasPendingData = true;
+            hasPendingDatagram = isUdpSocket;
+        }
     }
 
     emitReadyRead();

src/network/socket/qabstractsocket_p.h

@@ -110,6 +110,7 @@ public:
     qint64 readBufferMaxSize = 0;
     bool isBuffered = false;
     bool hasPendingData = false;
+    bool hasPendingDatagram = false;
 
     QTimer *connectTimer = nullptr;
 

src/network/socket/qudpsocket.cpp

@@ -430,6 +430,7 @@ QNetworkDatagram QUdpSocket::receiveDatagram(qint64 maxSize)
     qint64 readBytes = d->socketEngine->readDatagram(result.d->data.data(), maxSize, &result.d->header,
                                                      QAbstractSocketEngine::WantAll);
     d->hasPendingData = false;
+    d->hasPendingDatagram = false;
     d->socketEngine->setReadNotificationEnabled(true);
     if (readBytes < 0) {
         d->setErrorAndEmit(d->socketEngine->error(), d->socketEngine->errorString());
@@ -479,6 +480,7 @@ qint64 QUdpSocket::readDatagram(char *data, qint64 maxSize, QHostAddress *addres
     }
 
     d->hasPendingData = false;
+    d->hasPendingDatagram = false;
     d->socketEngine->setReadNotificationEnabled(true);
     if (readBytes < 0) {
         if (readBytes == -2) {

tests/auto/network/socket/qudpsocket/tst_qudpsocket.cpp

@@ -10,6 +10,7 @@
 #endif
 #include <QScopeGuard>
 #include <QVersionNumber>
+#include <QSemaphore>
 
 #include <qcoreapplication.h>
 #include <qfileinfo.h>
@@ -104,6 +105,8 @@ private slots:
     void asyncReadDatagram();
     void writeInHostLookupState();
 
+    void readyReadConnectionThrottling();
+
 protected slots:
     void empty_readyReadSlot();
     void empty_connectedSlot();
@@ -1905,5 +1908,78 @@ void tst_QUdpSocket::writeInHostLookupState()
     QVERIFY(!socket.putChar('0'));
 }
 
+void tst_QUdpSocket::readyReadConnectionThrottling()
+{
+    QFETCH_GLOBAL(bool, setProxy);
+    if (setProxy)
+        return;
+    using namespace std::chrono_literals;
+
+    // QTBUG-105871:
+    // We have some signal/slot connection throttling in QAbstractSocket, but it
+    // was caring about the bytes, not about the datagrams.
+    // Test that we don't disable read notifications until we have at least one
+    // datagram available. Otherwise our good users who use the datagram APIs
+    // can get into scenarios where they no longer get the readyRead signal
+    // unless they call a read function once in a while.
+
+    QUdpSocket receiver;
+    QVERIFY(receiver.bind(QHostAddress(QHostAddress::LocalHost), 0));
+
+    QSemaphore semaphore;
+
+    // Repro-ing deterministically eludes me, so we are bruteforcing it:
+    // The thread acts as a remote sender, flooding the receiver with datagrams,
+    // and at some point the receiver would get into the broken state mentioned
+    // earlier.
+    std::unique_ptr<QThread> thread(QThread::create([&semaphore, port = receiver.localPort()]() {
+        QUdpSocket sender;
+        sender.connectToHost(QHostAddress(QHostAddress::LocalHost), port);
+        QCOMPARE(sender.state(), QUdpSocket::ConnectedState);
+
+        constexpr qsizetype PayloadSize = 242;
+        const QByteArray payload(PayloadSize, 'a');
+
+        semaphore.acquire(); // Wait for main thread to be ready
+        while (true) {
+            // We send 100 datagrams at a time, then sleep.
+            // This is mostly to let the main thread catch up between bursts so
+            // it doesn't get stuck in the loop.
+            for (int i = 0; i < 100; ++i) {
+                [[maybe_unused]]
+                qsizetype sent = sender.write(payload);
+                Q_ASSERT(sent > 0);
+            }
+            if (QThread::currentThread()->isInterruptionRequested())
+                break;
+            QThread::sleep(20ms);
+        }
+    }));
+    thread->start();
+    auto threadStopAndWaitGuard = qScopeGuard([&thread] {
+        thread->requestInterruption();
+        thread->quit();
+        thread->wait();
+    });
+
+    qsizetype count = 0;
+    QObject::connect(&receiver, &QUdpSocket::readyRead, &receiver,
+            [&] {
+                while (receiver.hasPendingDatagrams()) {
+                    receiver.readDatagram(nullptr, 0);
+                    ++count;
+                }
+                // If this prints `false, xxxx` we were pretty much guaranteed
+                // that we would not get called again:
+                // qDebug() << receiver.hasPendingDatagrams() << receiver.bytesAvailable();
+            },
+            Qt::QueuedConnection);
+
+    semaphore.release();
+    constexpr qsizetype MaxCount = 500;
+    QVERIFY2(QTest::qWaitFor([&] { return count >= MaxCount; }, 10s),
+             QByteArray::number(count).constData());
+}
+
 QTEST_MAIN(tst_QUdpSocket)
 #include "tst_qudpsocket.moc"