Fix for race condition in signal activation

There was a race condition between QObject::disconnect() and
QMetaObject::activate() which can occur if there are multiple
BlockingQueued connections to one signal from different threads and
they connect/disconnect their connections often.

What can happen in this case is:
T1 is in activate() method and T2 is in disconnect() method

T1                          T2
locks sender mutex
selects next connection
unlocks sender mutex
                            locks sender mutex
                            sets isSlotObject to false
creates QMetaCallEvent      derefs connection
posts event

Two things can happen here:
1. Connection can still be valid, but it will have isSlotObject==false
and callFunction will be used instead of slotObj
2. Connection can already be invalid

To fix it mutex unlock should be moved after QMetaCallEvent creation.

Also there is another case, when we don't disconnect but delete the
receiver object. In this case it can already be invalid during
postEvent, so we need to move mutex unlock after postEvent.

Change-Id: I8103798324140ee11de5b4e10906562ba878ff8b
Reviewed-by: Olivier Goffart (Woboq GmbH) <ogoffart@woboq.com>
Reviewed-by: Marc Mutz <marc.mutz@kdab.com>

src/corelib/kernel/qobject.cpp

@@ -3683,7 +3683,6 @@ void QMetaObject::activate(QObject *sender, int signalOffset, int local_signal_i
                 continue;
 #ifndef QT_NO_THREAD
             } else if (c->connectionType == Qt::BlockingQueuedConnection) {
-                locker.unlock();
                 if (receiverInSameThread) {
                     qWarning("Qt: Dead lock detected while activating a BlockingQueuedConnection: "
                     "Sender is %s(%p), receiver is %s(%p)",
@@ -3695,6 +3694,7 @@ void QMetaObject::activate(QObject *sender, int signalOffset, int local_signal_i
                     new QMetaCallEvent(c->slotObj, sender, signal_index, 0, 0, argv ? argv : empty_argv, &semaphore) :
                     new QMetaCallEvent(c->method_offset, c->method_relative, c->callFunction, sender, signal_index, 0, 0, argv ? argv : empty_argv, &semaphore);
                 QCoreApplication::postEvent(receiver, ev);
+                locker.unlock();
                 semaphore.acquire();
                 locker.relock();
                 continue;

tests/auto/other/qobjectrace/tst_qobjectrace.cpp

@@ -51,6 +51,7 @@ class tst_QObjectRace: public QObject
 private slots:
     void moveToThreadRace();
     void destroyRace();
+    void disconnectRace();
 };
 
 class RaceObject : public QObject
@@ -298,6 +299,180 @@ void tst_QObjectRace::destroyRace()
         delete threads[i];
 }
 
+static QAtomicInteger<unsigned> countedStructObjectsCount;
+struct CountedFunctor
+{
+    CountedFunctor() : destroyed(false) { countedStructObjectsCount.fetchAndAddRelaxed(1); }
+    CountedFunctor(const CountedFunctor &) : destroyed(false) { countedStructObjectsCount.fetchAndAddRelaxed(1); }
+    CountedFunctor &operator=(const CountedFunctor &) { return *this; }
+    ~CountedFunctor() { destroyed = true; countedStructObjectsCount.fetchAndAddRelaxed(-1);}
+    void operator()() const {QCOMPARE(destroyed, false);}
+
+private:
+    bool destroyed;
+};
+
+class DisconnectRaceSenderObject : public QObject
+{
+    Q_OBJECT
+signals:
+    void theSignal();
+};
+
+class DisconnectRaceThread : public QThread
+{
+    Q_OBJECT
+
+    DisconnectRaceSenderObject *sender;
+    bool emitSignal;
+public:
+    DisconnectRaceThread(DisconnectRaceSenderObject *s, bool emitIt)
+        : QThread(), sender(s), emitSignal(emitIt)
+    {
+    }
+
+    void run()
+    {
+        while (!isInterruptionRequested()) {
+            QMetaObject::Connection conn = connect(sender, &DisconnectRaceSenderObject::theSignal,
+                                                   sender, CountedFunctor(), Qt::BlockingQueuedConnection);
+            if (emitSignal)
+                emit sender->theSignal();
+            disconnect(conn);
+            yieldCurrentThread();
+        }
+    }
+};
+
+class DeleteReceiverRaceSenderThread : public QThread
+{
+    Q_OBJECT
+
+    DisconnectRaceSenderObject *sender;
+public:
+    DeleteReceiverRaceSenderThread(DisconnectRaceSenderObject *s)
+        : QThread(), sender(s)
+    {
+    }
+
+    void run()
+    {
+        while (!isInterruptionRequested()) {
+            emit sender->theSignal();
+            yieldCurrentThread();
+        }
+    }
+};
+
+class DeleteReceiverRaceReceiver : public QObject
+{
+    Q_OBJECT
+
+    DisconnectRaceSenderObject *sender;
+    QObject *receiver;
+    QTimer *timer;
+public:
+    DeleteReceiverRaceReceiver(DisconnectRaceSenderObject *s)
+        : QObject(), sender(s), receiver(0)
+    {
+        timer = new QTimer(this);
+        connect(timer, &QTimer::timeout, this, &DeleteReceiverRaceReceiver::onTimeout);
+        timer->start(1);
+    }
+
+    void onTimeout()
+    {
+        if (receiver)
+            delete receiver;
+        receiver = new QObject;
+        connect(sender, &DisconnectRaceSenderObject::theSignal, receiver, CountedFunctor(), Qt::BlockingQueuedConnection);
+    }
+};
+
+class DeleteReceiverRaceReceiverThread : public QThread
+{
+    Q_OBJECT
+
+    DisconnectRaceSenderObject *sender;
+public:
+    DeleteReceiverRaceReceiverThread(DisconnectRaceSenderObject *s)
+        : QThread(), sender(s)
+    {
+    }
+
+    void run()
+    {
+        QScopedPointer<DeleteReceiverRaceReceiver> receiver(new DeleteReceiverRaceReceiver(sender));
+        exec();
+    }
+};
+
+void tst_QObjectRace::disconnectRace()
+{
+    enum { ThreadCount = 20, TimeLimit = 3000 };
+
+    QCOMPARE(countedStructObjectsCount.load(), 0u);
+
+    {
+        QScopedPointer<DisconnectRaceSenderObject> sender(new DisconnectRaceSenderObject());
+        QScopedPointer<QThread> senderThread(new QThread());
+        senderThread->start();
+        sender->moveToThread(senderThread.data());
+
+        DisconnectRaceThread *threads[ThreadCount];
+        for (int i = 0; i < ThreadCount; ++i) {
+            threads[i] = new DisconnectRaceThread(sender.data(), !(i % 10));
+            threads[i]->start();
+        }
+
+        QTime timeLimiter;
+        timeLimiter.start();
+
+        while (timeLimiter.elapsed() < TimeLimit)
+            QTest::qWait(10);
+
+        for (int i = 0; i < ThreadCount; ++i) {
+            threads[i]->requestInterruption();
+            QVERIFY(threads[i]->wait(300));
+            delete threads[i];
+        }
+
+        senderThread->quit();
+        QVERIFY(senderThread->wait(300));
+    }
+
+    QCOMPARE(countedStructObjectsCount.load(), 0u);
+
+    {
+        QScopedPointer<DisconnectRaceSenderObject> sender(new DisconnectRaceSenderObject());
+        QScopedPointer<DeleteReceiverRaceSenderThread> senderThread(new DeleteReceiverRaceSenderThread(sender.data()));
+        senderThread->start();
+        sender->moveToThread(senderThread.data());
+
+        DeleteReceiverRaceReceiverThread *threads[ThreadCount];
+        for (int i = 0; i < ThreadCount; ++i) {
+            threads[i] = new DeleteReceiverRaceReceiverThread(sender.data());
+            threads[i]->start();
+        }
+
+        QTime timeLimiter;
+        timeLimiter.start();
+
+        while (timeLimiter.elapsed() < TimeLimit)
+            QTest::qWait(10);
+
+        senderThread->requestInterruption();
+        QVERIFY(senderThread->wait(300));
+
+        for (int i = 0; i < ThreadCount; ++i) {
+            threads[i]->quit();
+            QVERIFY(threads[i]->wait(300));
+            delete threads[i];
+        }
+    }
+
+    QCOMPARE(countedStructObjectsCount.load(), 0u);
+}
 
 QTEST_MAIN(tst_QObjectRace)
 #include "tst_qobjectrace.moc"