Changeset 196 for trunk/src

Timestamp:

Sep 30, 2009, 6:08:53 AM (16 years ago)

Author:

Dmitry A. Kuminov

Message:

corelib: Improved socket notifiers: don't block other threads by constantly posting "socket ready" messages to the message queue in cases when the target thread isn't fast enough to process them. This fixes spontaneous various odd behavior (partial lines, unexpected child termination due to EPIPE) when redirecting child process output in QProcess.

File:

• trunk/src/corelib/kernel/qeventdispatcher_pm.cpp (modified) (14 diffs)

trunk/src/corelib/kernel/qeventdispatcher_pm.cpp

@@ -284 +284 @@
     static void addSelect(QSocketNotifier *notifier, HWND hwnd);
     static void removeSelect(QSocketNotifier *notifier);
-    static QSocketNotifier *getSocketNotifier(int key);
+    static QSocketNotifier *getSocketNotifier(int key, bool reset);
 
     // timer support is based on QTimerInfoList from the Unix implementation
@@ -300 +300 @@
     ~QSelectThread();
 
+    enum Op { Add, Remove };
+    void updateMaxSockFd(int sockFd, Op op);
+    fd_set *setForType(QSocketNotifier::Type type);
+
     void run();
     void cancelSelectOrIdle();
@@ -322 +326 @@
     // socket stuff
 
-    typedef QHash<int, QSocketNotifier*> Sockets;
+    typedef QPair<QSocketNotifier*, HWND> SockNot;
+    typedef QHash<int, SockNot> Sockets;
     Sockets sockets;
+
+    fd_set readS, writeS, exS;
     int maxSockfd;
-
-    enum Op { Add, Remove };
-    struct PendingSockOp {
-        Op op;
-        int sockfd;
-        Type type;
-        HWND hwnd;
-    };
-
-    typedef QList<PendingSockOp> PendingSockets;
-    PendingSockets pendingSockets;
 
     // timer stuff
@@ -402 +398 @@
 
     }
-    instance->sockets.insert(key, notifier);
-    PendingSockOp op = {Add, sockfd, type, hwnd};
-    instance->pendingSockets.append(op);
+    instance->sockets.insert(key, qMakePair(notifier, hwnd));
+    FD_SET(sockfd, instance->setForType(type));
+    instance->updateMaxSockFd(sockfd, Add);
     instance->cancelSelectOrIdle();
 }
@@ -420 +416 @@
     if (instance->sockets.contains(key)) {
         instance->sockets.remove(key);
-        PendingSockOp op = {Remove, sockfd, type};
-        instance->pendingSockets.append(op);
+        FD_CLR(sockfd, instance->setForType(type));
+        instance->updateMaxSockFd(sockfd, Remove);
         instance->cancelSelectOrIdle();
     }
@@ -436 +432 @@
 */
 // static
-QSocketNotifier *QSelectThread::getSocketNotifier(int key)
+QSocketNotifier *QSelectThread::getSocketNotifier(int key, bool reset)
 {
     QMutexLocker locker(&mutex);
     Q_ASSERT(instance);
 
-    if (instance->sockets.contains(key) &&
-        instance->sockets[key]->thread() == QThread::currentThread())
-        return instance->sockets[key];
+    if (instance->sockets.contains(key)) {
+        QSocketNotifier* notifier = instance->sockets[key].first;
+        if (notifier->thread() == QThread::currentThread()) {
+            if (reset && notifier->isEnabled()) {
+                // add the socket back to the set
+                int sockfd = notifier->socket();
+                fd_set *set = instance->setForType(notifier->type());
+                FD_SET(sockfd, set);
+                instance->updateMaxSockFd(sockfd, Add);
+                // inform the select thread that this socket may be included
+                // in the set and posted again
+                instance->cancelSelectOrIdle();
+            }
+            return notifier;
+        }
+    }
 
     return 0;
@@ -577 +586 @@
             } else {
                 // also wake it up if this timer was skipped when choosing the
-                // shortest wait interval so that a longer once could be chosen
+                // shortest wait interval so that a longer one could be chosen
                 bool haveNonPosted = false;
                 for (TimevalMap::const_iterator it = instance->timersByTimeout.begin();
@@ -647 +656 @@
     for (Sockets::iterator it = instance->sockets.begin();
           it != instance->sockets.end();) {
-        QSocketNotifier *notifier = it.value();
+        QSocketNotifier *notifier = it.value().first;
         if (notifier->thread() == QThread::currentThread()) {
-            PendingSockOp op = {Remove, notifier->socket(), notifier->type()};
-            instance->pendingSockets.append(op);
             it = instance->sockets.erase(it);
+            instance->updateMaxSockFd(notifier->socket(), Remove);
+            FD_CLR(notifier->socket(), instance->setForType(notifier->type()));
         } else {
             ++it;
@@ -694 +703 @@
 {
     // initialize socket stuff
+    FD_ZERO(&readS);
+    FD_ZERO(&writeS);
+    FD_ZERO(&exS);
     maxSockfd = -1;
 
@@ -720 +732 @@
 }
 
+// Finds the new highest socket FD. Must be called from under the mutex.
+void QSelectThread::updateMaxSockFd(int sockFd, Op op)
+{
+    switch (op) {
+    case Add:
+        maxSockfd = qMax(maxSockfd, sockFd);
+        break;
+    case Remove:
+        if (maxSockfd == sockFd) {
+            maxSockfd = -1;
+            if (!sockets.isEmpty()) {
+                for (Sockets::const_iterator it = sockets.constBegin();
+                      it != sockets.constEnd(); ++it) {
+                    maxSockfd = qMax(toSocket(it.key()), maxSockfd);
+                }
+            }
+        }
+        break;
+    }
+}
+
+// Returns the set corresponding to the socket type. Must be called from under
+// the mutex.
+fd_set *QSelectThread::setForType(QSocketNotifier::Type type)
+{
+    fd_set *set = 0;
+    switch (type) {
+        case QSocketNotifier::Read:
+            set = &readS; break;
+        case QSocketNotifier::Write:
+            set = &writeS; break;
+        case QSocketNotifier::Exception:
+            set = &exS; break;
+    }
+    return set;
+}
+
 void QSelectThread::run()
 {
-    // maintain a separate hash for HWNDs to avoid mutex locking every time
-    // select() returns an event that we want to post
-    typedef QHash<int, HWND> Hwnds;
-    Hwnds hwnds;
-
-    fd_set readS, writeS, exS;
-    FD_ZERO(&readS);
-    FD_ZERO(&writeS);
-    FD_ZERO(&exS);
-
     mutex.lock();
 
     do {
-        // process pending socket operations
-        while (!pendingSockets.isEmpty()) {
-            PendingSockOp p = pendingSockets.takeFirst();
-            switch (p.op) {
-                case Add:
-                    switch (p.type) {
-                        case QSocketNotifier::Read:
-                            FD_SET(p.sockfd, &readS); break;
-                        case QSocketNotifier::Write:
-                            FD_SET(p.sockfd, &writeS); break;
-                        case QSocketNotifier::Exception:
-                            FD_SET(p.sockfd, &exS); break;
-                    }
-                    hwnds.insert(toSockKey(p.sockfd, p.type), p.hwnd);
-                    maxSockfd = qMax(maxSockfd, p.sockfd);
-                    break;
-                case Remove:
-                    switch (p.type) {
-                        case QSocketNotifier::Read:
-                            FD_CLR(p.sockfd, &readS); break;
-                        case QSocketNotifier::Write:
-                            FD_CLR(p.sockfd, &writeS); break;
-                        case QSocketNotifier::Exception:
-                            FD_CLR(p.sockfd, &exS); break;
-                    }
-                    hwnds.remove(toSockKey(p.sockfd, p.type));
-                    if (maxSockfd == p.sockfd) {
-                        // find the new hignest socket
-                        maxSockfd = -1;
-                        if (!hwnds.isEmpty()) {
-                            for (Hwnds::const_iterator it = hwnds.constBegin();
-                                  it != hwnds.constEnd(); ++it) {
-                                maxSockfd = qMax(toSocket(it.key()), maxSockfd);
-                            }
-                        }
-                    }
-                    break;
-            }
-        }
-
         // get the maximum time we can wait (for the closest timer)
         timeval *timeout = 0;
@@ -787 +786 @@
             mutex.unlock();
             nsel = ::select(maxSockfd + 1, &tmpRead, &tmpWrite, &tmpEx, timeout);
+            mutex.lock();
             if (nsel > 0) {
-                for (Hwnds::const_iterator it = hwnds.constBegin();
-                      it != hwnds.constEnd(); ++it) {
+                // find out which sockets to post. Note that we remove these
+                // sockets from the main sets to avoid polluting the message
+                // queue with sockett messages if the target window is not fast
+                // enough to process them. They will be put back once processed.
+                for (Sockets::const_iterator it = sockets.constBegin();
+                      it != sockets.constEnd(); ++it) {
                     int sockfd = toSocket(it.key());
+                    Type type = toSockType(it.key());
                     bool isSet = false;
-                    switch (toSockType(it.key())) {
+                    switch (type) {
                         case QSocketNotifier::Read:
                             isSet = FD_ISSET(sockfd, &tmpRead); break;
@@ -800 +805 @@
                             isSet = FD_ISSET(sockfd, &tmpEx); break;
                     }
-                    if (isSet)
-                        WinPostMsg(it.value(), WM_U_SEM_SELECT, MPFROMLONG(it.key()), 0);
+                    if (isSet) {
+                        fd_set *set = setForType(type);
+                        FD_CLR(sockfd, set);
+                        updateMaxSockFd(sockfd, Remove);
+                        WinPostMsg(it.value().second, WM_U_SEM_SELECT, MPFROMLONG(it.key()), 0);
+                    }
                 }
             }
-            mutex.lock();
         } else {
             nsel = -1;
@@ -1035 +1043 @@
     }
     static void removeSelect(QSocketNotifier *notifier) {}
-    static QSocketNotifier *getSocketNotifier(int key); { return 0; }
+    static QSocketNotifier *getSocketNotifier(int key, bool reset); { return 0; }
 
     static void addTimer(int timerId, int interval, QObject *object, HWND hwnd) {
@@ -1147 +1155 @@
         case WM_U_SEM_SELECT: {
             QSocketNotifier *notifier =
-                QSelectThread::getSocketNotifier(LONGFROMMP(mp1));
+                QSelectThread::getSocketNotifier(LONGFROMMP(mp1), true /*reset*/);
             if (notifier) {
                 QEvent event(QEvent::SockAct);