Changeset 83 for trunk/src

Timestamp:

Jul 24, 2009, 2:31:38 AM (16 years ago)

Author:

Dmitry A. Kuminov

Message:

corelib: Implemented socket notifiers (#24).

File:

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

trunk/src/corelib/kernel/qeventdispatcher_pm.cpp

@@ -46 +46 @@
 #include "qcoreapplication.h"
 #include "qhash.h"
-#include "qlibrary.h"
-#include "qpair.h"
-#include "qset.h"
 #include "qsocketnotifier.h"
-#include "qvarlengtharray.h"
+#include "qmutex.h"
+#include "qwaitcondition.h"
 
 #include "qabstracteventdispatcher_p.h"
@@ -59 +57 @@
 
 #include <Qt/qwindowdefs_pm.h> // for QPMObjectWindow declaration
+
+#include <sys/socket.h>
 
 QT_BEGIN_NAMESPACE
@@ -152 +152 @@
 */
 
-/// @todo remove?
-// returns HMQ of the current thread or NULL if no event queue has been created
-static HMQ qt_WinQueryQueue(HAB hab)
-{
-    PTIB ptib;
-    PPIB ppib;
-    DosGetInfoBlocks(&ptib, &ppib);
-    return WinQueueFromID(hab, ppib->pib_ulpid, ptib->tib_ptib2->tib2_ultid);
-}
-
 /*!
     Constructs a new object window for the current thread.
@@ -263 +253 @@
 
 /*!
-  \fn QPMObjectWindow::send(ULONG msg, MPARAM mp1, MPARAM mp2) const
-
-  Synchronously sends a message \a msg with the given parameters \a mp1 and
-  \a mp2 to this window handle and returns a reply from the message() function.
-
-  \note Must be called on the same thread that cosnstructed this instance.
+    \fn QPMObjectWindow::send(ULONG msg, MPARAM mp1, MPARAM mp2) const
+
+    Synchronously sends a message \a msg with the given parameters \a mp1 and
+    \a mp2 to this window handle and returns a reply from the message() function.
+
+    \note Must be called on the same thread that cosnstructed this instance.
 */
 
 /*!
-  \fn QPMObjectWindow::post(ULONG msg, MPARAM mp1, MPARAM mp2) const
-
-  Asynchronously posts a message \a msg with the given parameters \a mp1 and
-  \a mp2 to this window handle. Returns \c TRUE on success and \c FALSE
-  otherwise.
-
-  \note Can be called on any thread.
+    \fn QPMObjectWindow::post(ULONG msg, MPARAM mp1, MPARAM mp2) const
+
+    Asynchronously posts a message \a msg with the given parameters \a mp1 and
+    \a mp2 to this window handle. Returns \c true on success and \c false
+    otherwise.
+
+    \note Can be called on any thread.
 */
+
 
 // socket select notification (highest priority)
@@ -285 +276 @@
 // zero timer notification (lowest priority)
 #define WM_U_SEM_ZEROTIMER  WM_SEM4
+
+/*****************************************************************************
+ socket select() thread
+ *****************************************************************************/
+
+#if !defined(QT_NO_THREAD)
+
+class QSockSelectThread : public QThread
+{
+public:
+    typedef QSocketNotifier::Type Type;
+
+    static void addSelect(QSocketNotifier *notifier, HWND hwnd);
+    static void removeSelect(QSocketNotifier *notifier);
+    static QSocketNotifier *getSocketNotifier(int key);
+
+    static void attachThread();
+    static void detachThread();
+
+private:
+    QSockSelectThread() : finish(false), refcnt(0), maxSockfd(-1) {};
+
+    void run();
+    void cancelSelectOrIdle();
+
+    bool finish;
+    int refcnt;
+    QWaitCondition cond;
+
+    typedef QHash<int, QSocketNotifier*> Sockets;
+    Sockets sockets;
+    int maxSockfd;
+
+    enum Op { Add, Remove };
+    struct PendingOp {
+        Op op;
+        int sockfd;
+        Type type;
+        HWND hwnd;
+    };
+    typedef QList<PendingOp> Pending;
+    Pending pending;
+
+    static int toKey(int sockfd, Type type) {
+        // as a hash key, we use first two bits of int for the type and the rest
+        // for the sockfd which should be enough in the real world
+        Q_ASSERT(((sockfd << 2) >> 2) == sockfd);
+        return (sockfd << 2) + type;
+    }
+    // opposite to toKey()
+    static int toSocket(int key) { return key >> 2; }
+    static Type toType(int key) { return Type(key & 0x3); }
+
+    static QSockSelectThread *instance;
+    static QMutex mutex;
+};
+
+// static
+QSockSelectThread *QSockSelectThread::instance = 0;
+QMutex QSockSelectThread::mutex;
+
+// static
+void QSockSelectThread::addSelect(QSocketNotifier *notifier, HWND hwnd)
+{
+    Q_ASSERT(hwnd != NULLHANDLE);
+    if (hwnd == NULLHANDLE)
+        return;
+
+    int sockfd = notifier->socket();
+    Type type = notifier->type();
+    int key = toKey(sockfd, type);
+
+    QMutexLocker locker(&mutex);
+    Q_ASSERT(instance);
+
+    instance->start();
+
+    if (instance->sockets.contains(key)) {
+        static const char *t[] = { "Read", "Write", "Exception" };
+        qWarning("QSocketNotifier: Multiple socket notifiers for "
+                 "same socket %d and type %s", sockfd, t[type]);
+
+    }
+    instance->sockets.insert(key, notifier);
+    PendingOp op = {Add, sockfd, type, hwnd};
+    instance->pending.append(op);
+    instance->cancelSelectOrIdle();
+}
+
+// static
+void QSockSelectThread::removeSelect(QSocketNotifier *notifier)
+{
+    QMutexLocker locker(&mutex);
+    Q_ASSERT(instance);
+
+    int sockfd = notifier->socket();
+    Type type = notifier->type();
+    int key = toKey(sockfd, notifier->type());
+
+    if (instance->sockets.contains(key)) {
+        instance->sockets.remove(key);
+        PendingOp op = {Remove, sockfd, type};
+        instance->pending.append(op);
+        instance->cancelSelectOrIdle();
+    }
+}
+
+/*!
+    Returns the socket notifier object corresponding to the given key from the
+    WM_U_SEM_SELECT message. The return value is only correct for the thread
+    that owns QSocketNotifier (creates/registers/unregisters it).
+
+    May return 0 if the socket notifier object is disabled/deleted after
+    WM_U_SEM_SELECT is issued for it but before this message gets processed by
+    the owning thread.
+*/
+// static
+QSocketNotifier *QSockSelectThread::getSocketNotifier(int key)
+{
+    QMutexLocker locker(&mutex);
+    Q_ASSERT(instance);
+
+    if (instance->sockets.contains(key) &&
+        instance->sockets[key]->thread() == QThread::currentThread())
+        return instance->sockets[key];
+
+    return 0;
+}
+
+/*!
+    Incrreases the usage count of the QSockSelectThread instance by one. If no
+    QSockSelectThread instance exists yet, creates it. Note that the thread is
+    started on demand by addSelect(), not by this method.
+
+    Must be called once per thread before any other call and must be completed
+    by the detachThread() call when socket select functionality is no more
+    necassary.
+*/
+// static
+void QSockSelectThread::attachThread()
+{
+    QMutexLocker locker(&mutex);
+
+    if (instance == 0) {
+        instance = new QSockSelectThread();
+        instance->start();
+    }
+
+    ++instance->refcnt;
+}
+
+/*!
+    Removes all socket notifiers owned by the current thread and decreases the
+    usage count of the QSockSelectThread instance by one. When the usage count
+    goes to zero, the socket select thread is stopped and the instance is
+    deleted.
+
+    May only be called once per thread and only if attachThread() was called on
+    this thread before.
+*/
+// static
+void QSockSelectThread::detachThread()
+{
+    QMutexLocker locker(&mutex);
+    Q_ASSERT(instance);
+
+    for (Sockets::iterator it = instance->sockets.begin();
+          it != instance->sockets.end();) {
+        QSocketNotifier *notifier = it.value();
+        if (notifier->thread() == QThread::currentThread()) {
+            PendingOp op = {Remove, notifier->socket(), notifier->type()};
+            instance->pending.append(op);
+            it = instance->sockets.erase(it);
+        } else {
+            ++it;
+        }
+    }
+
+    if (--instance->refcnt == 0) {
+        instance->finish = true;
+        instance->cancelSelectOrIdle();
+        instance->wait();
+        delete instance;
+        instance = 0;
+    } else {
+        instance->cancelSelectOrIdle();
+    }
+}
+void QSockSelectThread::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);
+
+    do {
+        mutex.lock();
+
+        while (!finish && sockets.isEmpty()) {
+            cond.wait(&mutex);
+        }
+
+        if (finish) {
+            mutex.unlock();
+            break;
+        }
+
+        while (!pending.isEmpty()) {
+            PendingOp p = pending.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(toKey(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(toKey(p.sockfd, p.type));
+                    if (maxSockfd == p.sockfd) {
+                        // find the new hignest socket
+                        if (hwnds.isEmpty()) {
+                            maxSockfd = -1;
+                        } else {
+                            for (Hwnds::const_iterator it = hwnds.constBegin();
+                                  it != hwnds.constEnd(); ++it) {
+                                maxSockfd = qMax(toSocket(it.key()), maxSockfd);
+                            }
+                        }
+                    }
+                    break;
+            }
+        }
+
+        // do select
+        mutex.unlock();
+        int nsel = ::select(maxSockfd + 1, &readS, &writeS, &exS, NULL);
+        if (nsel > 0) {
+            for (Hwnds::const_iterator it = hwnds.constBegin();
+                  it != hwnds.constEnd(); ++it) {
+                int sockfd = toSocket(it.key());
+                bool isSet = false;
+                switch (toType(it.key())) {
+                    case QSocketNotifier::Read:
+                        isSet = FD_ISSET(sockfd, &readS); break;
+                    case QSocketNotifier::Write:
+                        isSet = FD_ISSET(sockfd, &writeS); break;
+                    case QSocketNotifier::Exception:
+                        isSet = FD_ISSET(sockfd, &exS); break;
+                }
+                if (isSet)
+                    WinPostMsg(it.value(), WM_U_SEM_SELECT, MPFROMLONG(it.key()), 0);
+            }
+        }
+    } while(true);
+}
+
+// Must be called from under QSockSelectThread::mutex
+void QSockSelectThread::cancelSelectOrIdle()
+{
+    if (maxSockfd >= 0) {
+        // terminate select() execution
+        ::so_cancel(maxSockfd);
+    } else {
+        // terminate the idle state
+        cond.wakeOne();
+    }
+}
+
+#else
+
+class QSockSelectThread
+{
+public:
+    static void addSelect(QSocketNotifier *notifier, HWND hwnd) {
+#ifndef QT_NO_DEBUG
+        qWarning("QSocketNotifier: socket notifiers require thread support but"
+                 "QT_NO_THREAD was defined");
+#endif
+    }
+    static void removeSelect(QSocketNotifier *notifier) {}
+    static QSocketNotifier *getSocketNotifier(int key); { return 0; }
+
+    static void attachThread() {}
+    static void detachThread() {}
+};
+
+#endif
 
 class QEventDispatcherPMPrivate : public QAbstractEventDispatcherPrivate
@@ -318 +615 @@
 
 // @todo later
-//  DWORD threadId;
-//
-//  bool interrupt;
-//
-//  // internal window handle used for socketnotifiers/timers/etc
-//  HWND internalHwnd;
 //
 //  // timers
@@ -331 +622 @@
 //  void unregisterTimer(WinTimerInfo *t);
 //  void sendTimerEvent(int timerId);
-//
-//  // socket notifiers
-//  QSNDict sn_read;
-//  QSNDict sn_write;
-//  QSNDict sn_except;
-//  void doWsaAsyncSelect(int socket);
 //
 //  QList<QMSG> queuedUserInputEvents;
@@ -399 +684 @@
     switch (msg) {
         case WM_U_SEM_SELECT: {
-// @todo later
-//          if (eventLoop)
-//              eventLoop->activateSocketNotifiers();
+            QSocketNotifier *notifier =
+                QSockSelectThread::getSocketNotifier(LONGFROMMP(mp1));
+                if (notifier) {
+                    QEvent event(QEvent::SockAct);
+                    QCoreApplication::sendEvent(notifier, &event);
+                }
             break;
         }
@@ -575 +863 @@
 }
 
-void QEventDispatcherPMPrivate::doWsaAsyncSelect(int socket)
-{
-    Q_ASSERT(internalHwnd);
-    int sn_event = 0;
-    if (sn_read.contains(socket))
-        sn_event |= FD_READ | FD_CLOSE | FD_ACCEPT;
-    if (sn_write.contains(socket))
-        sn_event |= FD_WRITE | FD_CONNECT;
-    if (sn_except.contains(socket))
-        sn_event |= FD_OOB;
-    // BoundsChecker may emit a warning for WSAAsyncSelect when sn_event == 0
-    // This is a BoundsChecker bug and not a Qt bug
-    WSAAsyncSelect(socket, internalHwnd, sn_event ? WM_USER : 0, sn_event);
-}
-
 void QEventDispatcherPM::createInternalHwnd()
 {
@@ -624 +897 @@
 bool QEventDispatcherPM::processEvents(QEventLoop::ProcessEventsFlags flags)
 {
-    Q_D(QEventDispatcherPM);
-
 // @todo later
 #if 0
+    Q_D(QEventDispatcherPM);
+
     if (!d->internalHwnd)
         createInternalHwnd();
@@ -732 +1005 @@
 void QEventDispatcherPM::registerSocketNotifier(QSocketNotifier *notifier)
 {
-// @todo later
-#if 0
     Q_ASSERT(notifier);
+#ifndef QT_NO_DEBUG
     int sockfd = notifier->socket();
-    int type = notifier->type();
-#ifndef QT_NO_DEBUG
-    if (sockfd < 0) {
+    if (sockfd < 0
+        || unsigned(sockfd) >= FD_SETSIZE) {
         qWarning("QSocketNotifier: Internal error");
         return;
-    } else if (notifier->thread() != thread() || thread() != QThread::currentThread()) {
+    } else if (notifier->thread() != thread()
+               || thread() != QThread::currentThread()) {
         qWarning("QSocketNotifier: socket notifiers cannot be enabled from another thread");
         return;
@@ -748 +1020 @@
 
     Q_D(QEventDispatcherPM);
-    QSNDict *sn_vec[3] = { &d->sn_read, &d->sn_write, &d->sn_except };
-    QSNDict *dict = sn_vec[type];
-
-    if (QCoreApplication::closingDown()) // ### d->exitloop?
-        return; // after sn_cleanup, don't reinitialize.
-
-    if (dict->contains(sockfd)) {
-        const char *t[] = { "Read", "Write", "Exception" };
-    /* Variable "socket" below is a function pointer. */
-        qWarning("QSocketNotifier: Multiple socket notifiers for "
-                 "same socket %d and type %s", sockfd, t[type]);
-    }
-
-    QSockNot *sn = new QSockNot;
-    sn->obj = notifier;
-    sn->fd  = sockfd;
-    dict->insert(sn->fd, sn);
-
-    if (d->internalHwnd)
-        d->doWsaAsyncSelect(sockfd);
-#endif
+    d->createAuxWnd();
+
+    QSockSelectThread::addSelect(notifier, d->auxWnd.hwnd());
 }
 
 void QEventDispatcherPM::unregisterSocketNotifier(QSocketNotifier *notifier)
 {
-// @todo later
-#if 0
     Q_ASSERT(notifier);
+#ifndef QT_NO_DEBUG
     int sockfd = notifier->socket();
-    int type = notifier->type();
-#ifndef QT_NO_DEBUG
-    if (sockfd < 0) {
+    if (sockfd < 0
+        || unsigned(sockfd) >= FD_SETSIZE) {
         qWarning("QSocketNotifier: Internal error");
         return;
-    } else if (notifier->thread() != thread() || thread() != QThread::currentThread()) {
+    } else if (notifier->thread() != thread()
+               || thread() != QThread::currentThread()) {
         qWarning("QSocketNotifier: socket notifiers cannot be disabled from another thread");
         return;
@@ -788 +1041 @@
 #endif
 
-    Q_D(QEventDispatcherPM);
-    QSNDict *sn_vec[3] = { &d->sn_read, &d->sn_write, &d->sn_except };
-    QSNDict *dict = sn_vec[type];
-    QSockNot *sn = dict->value(sockfd);
-    if (!sn)
-        return;
-
-    dict->remove(sockfd);
-    delete sn;
-
-    if (d->internalHwnd)
-        d->doWsaAsyncSelect(sockfd);
-#endif
+    QSockSelectThread::removeSelect(notifier);
 }
 
@@ -944 +1185 @@
 void QEventDispatcherPM::startingUp()
 {
+    QSockSelectThread::attachThread();
 }
 
 void QEventDispatcherPM::closingDown()
 {
-    Q_D(QEventDispatcherPM);
+    QSockSelectThread::detachThread();
 
 // @todo remove later
 #if 0
-    // clean up any socketnotifiers
-    while (!d->sn_read.isEmpty())
-        unregisterSocketNotifier((*(d->sn_read.begin()))->obj);
-    while (!d->sn_write.isEmpty())
-        unregisterSocketNotifier((*(d->sn_write.begin()))->obj);
-    while (!d->sn_except.isEmpty())
-        unregisterSocketNotifier((*(d->sn_except.begin()))->obj);
+    Q_D(QEventDispatcherPM);
 
     // clean up any timers