4 files changed, 85 insertions, 2 deletions

diff --git a/src/network/socket/qabstractsocket.cpp b/src/network/socket/qabstractsocket.cpp
index e74f448987..7c48692155 100644
--- a/src/network/socket/qabstractsocket.cpp
+++ b/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();
diff --git a/src/network/socket/qabstractsocket_p.h b/src/network/socket/qabstractsocket_p.h
index 98d74dcfd4..cc5f53179c 100644
--- a/src/network/socket/qabstractsocket_p.h
+++ b/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;
 
diff --git a/src/network/socket/qudpsocket.cpp b/src/network/socket/qudpsocket.cpp
index 2a7982057e..06bc6a4fd5 100644
--- a/src/network/socket/qudpsocket.cpp
+++ b/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) {
diff --git a/tests/auto/network/socket/qudpsocket/tst_qudpsocket.cpp b/tests/auto/network/socket/qudpsocket/tst_qudpsocket.cpp
index 7cf7a49706..689ff452f9 100644
--- a/tests/auto/network/socket/qudpsocket/tst_qudpsocket.cpp
+++ b/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"