Changeset 27431 in vbox

Timestamp:

Mar 17, 2010 1:33:58 AM (15 years ago)

Author:

vboxsync

Message:

iprt: Implemented RTPollSet on windows (pipes only). Optimized the RTPipeWrite (nonblocking) and RTPipeSelectOne implementations on windows so that we're down to one failing unit test in tstRTPipe.

Location:

trunk

Files:

• include/iprt/err.h (modified) (1 diff)
• src/VBox/Runtime/r3/win/pipe-win.cpp (modified) (23 diffs)
• src/VBox/Runtime/r3/win/poll-win.cpp (modified) (6 diffs)

trunk/include/iprt/err.h

@@ -1251 +1251 @@
 /** The handle ID was not found in the set. */
 #define VERR_POLL_HANDLE_ID_NOT_FOUND           (-952)
+/** The poll set is full. */
+#define VERR_POLL_SET_IS_FULL                   (-953)
 /** @} */
 

trunk/src/VBox/Runtime/r3/win/pipe-win.cpp

@@ -44 +44 @@
 #include <iprt/mem.h>
 #include <iprt/string.h>
+#include <iprt/poll.h>
 #include <iprt/process.h>
 #include <iprt/thread.h>
 #include <iprt/time.h>
+#include "internal/pipe.h"
 #include "internal/magics.h"
+
+
+/*******************************************************************************
+*   Defined Constants And Macros                                               *
+*******************************************************************************/
+/** The pipe buffer size we prefere. */
+#define RTPIPE_NT_SIZE      _64K
 
 
@@ -63 +72 @@
     /** Set if there is already pending I/O. */
     bool                fIOPending;
+    /** Set if the zero byte read that the poll code using is pending. */
+    bool                fZeroByteRead;
+    /** Set if the pipe is broken. */
+    bool                fBrokenPipe;
+    /** Set if we've promised that the handle is writable. */
+    bool                fPromisedWritable;
     /** The number of users of the current mode. */
     uint32_t            cModeUsers;
@@ -73 +88 @@
     /** Amount of allocated buffer space. */
     size_t              cbBounceBufAlloc;
+    /** The handle of the poll set currently polling on this pipe.
+     *  We can only have one poller at the time (lazy bird). */
+    RTPOLLSET           hPollSet;
+    /** The number of references to the handle in hPollSet. */
+    uint32_t            cPolls;
     /** Critical section protecting the above members.
      * (Taking the lazy/simple approach.) */
     RTCRITSECT          CritSect;
+    /** Buffer for the zero byte read. */
+    uint8_t             abBuf[8];
 } RTPIPEINTERNAL;
+
+
+/* from ntdef.h */
+typedef LONG NTSTATUS;
+
+/* from ntddk.h */
+typedef struct _IO_STATUS_BLOCK {
+    union {
+        NTSTATUS Status;
+        PVOID Pointer;
+    };
+    ULONG_PTR Information;
+} IO_STATUS_BLOCK, *PIO_STATUS_BLOCK;
+
+typedef enum _FILE_INFORMATION_CLASS {
+    FilePipeInformation = 23,
+    FilePipeLocalInformation = 24,
+    FilePipeRemoteInformation = 25,
+} FILE_INFORMATION_CLASS, *PFILE_INFORMATION_CLASS;
+
+/* from ntifs.h */
+typedef struct _FILE_PIPE_LOCAL_INFORMATION {
+     ULONG NamedPipeType;
+     ULONG NamedPipeConfiguration;
+     ULONG MaximumInstances;
+     ULONG CurrentInstances;
+     ULONG InboundQuota;
+     ULONG ReadDataAvailable;
+     ULONG OutboundQuota;
+     ULONG WriteQuotaAvailable;
+     ULONG NamedPipeState;
+     ULONG NamedPipeEnd;
+} FILE_PIPE_LOCAL_INFORMATION, *PFILE_PIPE_LOCAL_INFORMATION;
+
+#define FILE_PIPE_DISCONNECTED_STATE    0x00000001
+#define FILE_PIPE_LISTENING_STATE       0x00000002
+#define FILE_PIPE_CONNECTED_STATE       0x00000003
+#define FILE_PIPE_CLOSING_STATE         0x00000004
+
+
+extern "C" NTSYSAPI NTSTATUS WINAPI NtQueryInformationFile(HANDLE, PIO_STATUS_BLOCK, PVOID, LONG, FILE_INFORMATION_CLASS);
+
+
+/**
+ * Wrapper for getting FILE_PIPE_LOCAL_INFORMATION via the NT API.
+ *
+ * @returns Success inidicator (true/false).
+ * @param   pThis               The pipe.
+ * @param   pInfo               The info structure.
+ */
+static bool rtPipeQueryInfo(RTPIPEINTERNAL *pThis, FILE_PIPE_LOCAL_INFORMATION *pInfo)
+{
+    IO_STATUS_BLOCK Ios;
+    RT_ZERO(Ios);
+    RT_ZERO(*pInfo);
+    NTSTATUS rcNt = NtQueryInformationFile(pThis->hPipe, &Ios, pInfo, sizeof(*pInfo), FilePipeLocalInformation);
+    return rcNt >= 0;
+}
 
 
@@ -118 +198 @@
 #endif
 
-        hPipeR = CreateNamedPipeA(szName, dwOpenMode, dwPipeMode, 1 /*nMaxInstances*/, _64K,  _64K,
+        hPipeR = CreateNamedPipeA(szName, dwOpenMode, dwPipeMode, 1 /*nMaxInstances*/, RTPIPE_NT_SIZE, RTPIPE_NT_SIZE,
                                   NMPWAIT_USE_DEFAULT_WAIT, pSecurityAttributes);
 #ifdef PIPE_REJECT_REMOTE_CLIENTS
@@ -124 +204 @@
         {
             dwPipeMode &= ~PIPE_REJECT_REMOTE_CLIENTS;
-            hPipeR = CreateNamedPipeA(szName, dwOpenMode, dwPipeMode, 1 /*nMaxInstances*/, _64K,  _64K,
+            hPipeR = CreateNamedPipeA(szName, dwOpenMode, dwPipeMode, 1 /*nMaxInstances*/, RTPIPE_NT_SIZE, RTPIPE_NT_SIZE,
                                       NMPWAIT_USE_DEFAULT_WAIT, pSecurityAttributes);
         }
@@ -132 +212 @@
         {
             dwOpenMode &= ~FILE_FLAG_FIRST_PIPE_INSTANCE;
-            hPipeR = CreateNamedPipeA(szName, dwOpenMode, dwPipeMode, 1 /*nMaxInstances*/, _64K,  _64K,
+            hPipeR = CreateNamedPipeA(szName, dwOpenMode, dwPipeMode, 1 /*nMaxInstances*/, RTPIPE_NT_SIZE, RTPIPE_NT_SIZE,
                                       NMPWAIT_USE_DEFAULT_WAIT, pSecurityAttributes);
         }
@@ -201 +281 @@
                             pThisR->fRead           = true;
                             pThisW->fRead           = false;
-                            pThisR->fIOPending      = false;
-                            pThisW->fIOPending      = false;
+                            //pThisR->fIOPending      = false;
+                            //pThisW->fIOPending      = false;
+                            //pThisR->fZeroByteRead   = false;
+                            //pThisW->fZeroByteRead   = false;
+                            //pThisR->fBrokenPipe     = false;
+                            //pThisW->fBrokenPipe     = false;
+                            //pThisW->fPromisedWritable= false;
+                            //pThisR->fPromisedWritable= false;
                             //pThisR->cModeUsers      = 0;
                             //pThisW->cModeUsers      = 0;
@@ -211 +297 @@
                             //pThisR->cbBounceBufAlloc= 0;
                             //pThisW->cbBounceBufAlloc= 0;
+                            pThisR->hPollSet        = NIL_RTPOLLSET;
+                            pThisW->hPollSet        = NIL_RTPOLLSET;
+                            //pThisW->cPolls          = 0;
+                            //pThisR->cPolls          = 0;
 
                             *phPipeRead  = pThisR;
@@ -278 +368 @@
                         else
                             rc = RTErrConvertFromWin32(GetLastError());
+                        if (rc == VERR_BROKEN_PIPE)
+                            pThis->fBrokenPipe = true;
                     }
                     break;
@@ -408 +500 @@
             else
                 rc = RTErrConvertFromWin32(GetLastError());
+            if (rc == VERR_BROKEN_PIPE)
+                pThis->fBrokenPipe = true;
 
             pThis->cModeUsers--;
@@ -474 +568 @@
             }
 
+            if (rc == VERR_BROKEN_PIPE)
+                pThis->fBrokenPipe = true;
+
             if (pcbRead)
             {
@@ -519 +616 @@
                 Assert(!pThis->fIOPending);
 
+                /* Adjust the number of bytes to write to fit into the current
+                   buffer quota, unless we've promissed stuff in RTPipeSelectOne.
+                   WriteQuotaAvailable better not be zero when it shouldn't!! */
+                FILE_PIPE_LOCAL_INFORMATION Info;
+                if (   !pThis->fPromisedWritable
+                    && cbToWrite > 0
+                    && rtPipeQueryInfo(pThis, &Info))
+                {
+                    if (Info.NamedPipeState == FILE_PIPE_CLOSING_STATE)
+                        rc = VERR_BROKEN_PIPE;
+                    else if (   cbToWrite >= Info.WriteQuotaAvailable
+                             && Info.OutboundQuota != 0)
+                    {
+                        cbToWrite = Info.WriteQuotaAvailable;
+                        if (!cbToWrite)
+                            rc = VINF_TRY_AGAIN;
+                    }
+                }
+                pThis->fPromisedWritable = false;
+
                 /* Do the bounce buffering. */
                 if (    pThis->cbBounceBufAlloc < cbToWrite
-                    &&  pThis->cbBounceBufAlloc < _64K)
-                {
-                    if (cbToWrite > _64K)
-                        cbToWrite = _64K;
+                    &&  pThis->cbBounceBufAlloc < RTPIPE_NT_SIZE)
+                {
+                    if (cbToWrite > RTPIPE_NT_SIZE)
+                        cbToWrite = RTPIPE_NT_SIZE;
                     void *pv = RTMemRealloc(pThis->pbBounceBuf, RT_ALIGN_Z(cbToWrite, _1K));
                     if (pv)
@@ -534 +651 @@
                         rc = VERR_NO_MEMORY;
                 }
-                else if (cbToWrite > _64K)
-                    cbToWrite = _64K;
+                else if (cbToWrite > RTPIPE_NT_SIZE)
+                    cbToWrite = RTPIPE_NT_SIZE;
                 if (RT_SUCCESS(rc) && cbToWrite)
                 {
@@ -567 +684 @@
                 *pcbWritten = 0;
 
+            if (rc == VERR_BROKEN_PIPE)
+                pThis->fBrokenPipe = true;
+
             pThis->cModeUsers--;
         }
@@ -612 +732 @@
             {
                 Assert(!pThis->fIOPending);
+                pThis->fPromisedWritable = false;
 
                 /*
@@ -663 +784 @@
             }
 
+            if (rc == VERR_BROKEN_PIPE)
+                pThis->fBrokenPipe = true;
+
             pThis->cModeUsers--;
         }
@@ -717 +841 @@
 
     if (!FlushFileBuffers(pThis->hPipe))
-        return RTErrConvertFromWin32(GetLastError());
+    {
+        int rc = RTErrConvertFromWin32(GetLastError());
+        if (rc == VERR_BROKEN_PIPE)
+            pThis->fBrokenPipe = true;
+        return rc;
+    }
     return VINF_SUCCESS;
 }
@@ -735 +864 @@
     for (unsigned iLoop = 0;; iLoop++)
     {
-        uint8_t abBuf[4];
-        bool    fPendingRead  = false;
-        HANDLE  hWait         = INVALID_HANDLE_VALUE;
+        HANDLE  hWait = INVALID_HANDLE_VALUE;
         if (pThis->fRead)
         {
@@ -762 +889 @@
                 rc = ResetEvent(pThis->Overlapped.hEvent); Assert(rc == TRUE);
                 DWORD cbRead = 0;
-                if (ReadFile(pThis->hPipe, &abBuf[0], 0, &cbRead, &pThis->Overlapped))
+                if (ReadFile(pThis->hPipe, pThis->abBuf, 0, &cbRead, &pThis->Overlapped))
                 {
                     rc = VINF_SUCCESS;
@@ -772 +899 @@
                     pThis->cModeUsers++;
                     pThis->fIOPending = true;
-                    fPendingRead = true;
+                    pThis->fZeroByteRead = true;
                     hWait = pThis->Overlapped.hEvent;
                 }
@@ -781 +908 @@
         else
         {
+            if (pThis->fIOPending)
+            {
+                rc = rtPipeWriteCheckCompletion(pThis);
+                if (RT_FAILURE(rc))
+                    break;
+            }
             if (pThis->fIOPending)
                 hWait = pThis->Overlapped.hEvent;
             else
             {
-                /* If nothing pending, the next write will succeed because
-                   we buffer it and pretend that it does... */
-                rc = VINF_SUCCESS;
-                break;
+                FILE_PIPE_LOCAL_INFORMATION Info;
+                if (rtPipeQueryInfo(pThis, &Info))
+                {
+                    /* Check for broken pipe. */
+                    if (Info.NamedPipeState == FILE_PIPE_CLOSING_STATE)
+                    {
+                        rc = VERR_BROKEN_PIPE;
+                        break;
+                    }
+                    /* Check for available write buffer space. */
+                    else if (Info.WriteQuotaAvailable > 0)
+                    {
+                        pThis->fPromisedWritable = false;
+                        rc = VINF_SUCCESS;
+                        break;
+                    }
+                    /* delayed buffer alloc or timeout: phony promise
+                       later: See if we still can associate a semaphore with
+                              the pipe, like on OS/2. */
+                    else if (   Info.OutboundQuota == 0
+                             || cMillies)
+                    {
+                        pThis->fPromisedWritable = true;
+                        rc = VINF_SUCCESS;
+                        break;
+                    }
+                }
+                else
+                {
+                    pThis->fPromisedWritable = true;
+                    rc = VINF_SUCCESS;
+                    break;
+                }
             }
         }
@@ -833 +995 @@
 
             RTCritSectEnter(&pThis->CritSect);
-            if (fPendingRead)
+            if (pThis->fZeroByteRead)
             {
                 pThis->cModeUsers--;
@@ -847 +1009 @@
     }
 
+    if (rc == VERR_BROKEN_PIPE)
+        pThis->fBrokenPipe = true;
+
     RTCritSectLeave(&pThis->CritSect);
     return rc;
 }
 
+
+/**
+ * Internal RTPollSetAdd helper that returns one or two handles that should be
+ * added to the pollset.
+ *
+ * @returns Valid handle on success, INVALID_HANDLE_VALUE on failure.
+ * @param   hPipe               The pipe handle.
+ * @param   fEvents             The events we're polling for.
+ * @param   ph1                 wher to put the primary handle.
+ * @param   ph2                 Where to optionally return a 2nd handle.
+ */
+int rtPipePollGetHandles(RTPIPE hPipe, uint32_t fEvents, PHANDLE ph1, PHANDLE ph2)
+{
+    RTPIPEINTERNAL *pThis = hPipe;
+    AssertPtrReturn(pThis, VERR_INVALID_HANDLE);
+    AssertReturn(pThis->u32Magic == RTPIPE_MAGIC, VERR_INVALID_HANDLE);
+
+    AssertReturn(!(fEvents & RTPOLL_EVT_READ) || pThis->fRead, VERR_INVALID_PARAMETER);
+    AssertReturn(!(fEvents & RTPOLL_EVT_WRITE) || !pThis->fRead, VERR_INVALID_PARAMETER);
+
+    *ph1 = pThis->Overlapped.hEvent;
+    /* Later: Try register an event handle with the pipe like on OS/2, there is
+       a file control for doing this obviously intended for the OS/2 subsys.
+       The question is whether this still exists on Vista and W7. */
+    *ph2 = INVALID_HANDLE_VALUE;
+    return VINF_SUCCESS;
+}
+
+
+/**
+ * Checks for pending events.
+ *
+ * @returns Event mask or 0.
+ * @param   pThis               The pipe handle.
+ * @param   fEvents             The desired events.
+ */
+static uint32_t rtPipePollCheck(RTPIPEINTERNAL *pThis, uint32_t fEvents)
+{
+    uint32_t fRetEvents = 0;
+    if (pThis->fBrokenPipe)
+        fRetEvents |= RTPOLL_EVT_ERROR;
+    else if (pThis->fRead)
+    {
+        if (!pThis->fIOPending)
+        {
+            DWORD cbAvailable;
+            if (PeekNamedPipe(pThis->hPipe, NULL, 0, NULL, &cbAvailable, NULL))
+            {
+                if (   (fEvents & RTPOLL_EVT_READ)
+                    && cbAvailable > 0)
+                    fRetEvents |= RTPOLL_EVT_READ;
+            }
+            else
+            {
+                if (GetLastError() == ERROR_BROKEN_PIPE)
+                    pThis->fBrokenPipe = true;
+                fRetEvents |= RTPOLL_EVT_ERROR;
+            }
+        }
+    }
+    else
+    {
+        if (pThis->fIOPending)
+        {
+            rtPipeWriteCheckCompletion(pThis);
+            if (pThis->fBrokenPipe)
+                fRetEvents |= RTPOLL_EVT_ERROR;
+        }
+        if (   !pThis->fIOPending
+            && !fRetEvents)
+        {
+            FILE_PIPE_LOCAL_INFORMATION Info;
+            if (rtPipeQueryInfo(pThis, &Info))
+            {
+                /* Check for broken pipe. */
+                if (Info.NamedPipeState == FILE_PIPE_CLOSING_STATE)
+                {
+                    fRetEvents = RTPOLL_EVT_ERROR;
+                    pThis->fBrokenPipe = true;
+                }
+
+                /* Check if there is available buffer space. */
+                if (   !fRetEvents
+                    && (   Info.WriteQuotaAvailable > 0
+                        || Info.OutboundQuota == 0)
+                    )
+                    fRetEvents |= RTPOLL_EVT_WRITE;
+            }
+            else if (   !fRetEvents
+                && (fEvents & RTPOLL_EVT_WRITE))
+                fRetEvents |= RTPOLL_EVT_WRITE;
+        }
+    }
+
+    return fRetEvents;
+}
+
+
+/**
+ * Internal RTPoll helper that polls the pipe handle and, if @a fNoWait is
+ * clear, starts whatever actions we've got running during the poll call.
+ *
+ * @returns 0 if no pending events, actions initiated if @a fNoWait is clear.
+ *          Event mask (in @a fEvents) and no actions if the handle is ready
+ *          already.
+ *          UINT32_MAX (asserted) if the pipe handle is busy in I/O or a
+ *          different poll set.
+ *
+ * @param   hPipe               The pipe handle.
+ * @param   hPollSet            The poll set handle (for access checks).
+ * @param   fEvents             The events we're polling for.
+ * @param   fNoWait             Set if it's a zero-wait poll call.  Clear if
+ *                              we'll wait for an event to occur.
+ */
+uint32_t rtPipePollStart(RTPIPE hPipe, RTPOLLSET hPollSet, uint32_t fEvents, bool fNoWait)
+{
+    /** @todo All this polling code could be optimized to make fewer system
+     *        calls; like for instance the ResetEvent calls. */
+    RTPIPEINTERNAL *pThis = hPipe;
+    AssertPtrReturn(pThis, UINT32_MAX);
+    AssertReturn(pThis->u32Magic == RTPIPE_MAGIC, UINT32_MAX);
+
+    int rc = RTCritSectEnter(&pThis->CritSect);
+    AssertRCReturn(rc, UINT32_MAX);
+
+    /* Check that this is the only current use of this pipe. */
+    uint32_t fRetEvents;
+    if (   (   pThis->cPolls == 0
+            && pThis->cModeUsers == 0)
+        || pThis->hPollSet == hPollSet
+       )
+    {
+        /* Check what the current events are. */
+        fRetEvents = rtPipePollCheck(pThis, fEvents);
+        if (   !fRetEvents
+            && !fNoWait)
+        {
+            /* Make sure the event semaphore has been reset. */
+            if (!pThis->fIOPending)
+            {
+                rc = ResetEvent(pThis->Overlapped.hEvent);
+                Assert(rc == TRUE);
+            }
+
+            /* Kick off the zero byte read thing if applicable. */
+            if (   !pThis->fIOPending
+                && pThis->fRead
+                && (fEvents & RTPOLL_EVT_READ)
+               )
+            {
+                DWORD cbRead = 0;
+                if (ReadFile(pThis->hPipe, pThis->abBuf, 0, &cbRead, &pThis->Overlapped))
+                    fRetEvents = rtPipePollCheck(pThis, fEvents);
+                else if (GetLastError() == ERROR_IO_PENDING)
+                {
+                    pThis->fIOPending    = true;
+                    pThis->fZeroByteRead = true;
+                }
+                else
+                    fRetEvents = RTPOLL_EVT_ERROR;
+            }
+
+            /* If we're still set for the waiting, record the poll set and
+               mark the pipe used. */
+            if (!fRetEvents)
+            {
+                pThis->cPolls++;
+                pThis->cModeUsers++;
+                pThis->hPollSet = hPollSet;
+            }
+        }
+    }
+    else
+    {
+        AssertFailed();
+        fRetEvents = UINT32_MAX;
+    }
+
+    RTCritSectLeave(&pThis->CritSect);
+    return fRetEvents;
+}
+
+
+/**
+ * Called after a WaitForMultipleObjects returned in order to check for pending
+ * events and stop whatever actions that rtPipePollStart() initiated.
+ *
+ * @returns Event mask or 0.
+ *
+ * @param   hPipe               The pipe handle.
+ * @param   fEvents             The events we're polling for.
+ */
+uint32_t rtPipePollDone(RTPIPE hPipe, uint32_t fEvents)
+{
+    RTPIPEINTERNAL *pThis = hPipe;
+    AssertPtrReturn(pThis, 0);
+    AssertReturn(pThis->u32Magic == RTPIPE_MAGIC, 0);
+
+    int rc = RTCritSectEnter(&pThis->CritSect);
+    AssertRCReturn(rc, 0);
+
+    Assert(pThis->cPolls > 0);
+    Assert(pThis->cModeUsers > 0);
+
+
+    /* Cancel the zero byte read. */
+    uint32_t fRetEvents = 0;
+    if (pThis->fZeroByteRead)
+    {
+        CancelIo(pThis->hPipe);
+        DWORD cbRead = 0;
+        if (   !GetOverlappedResult(pThis->hPipe, &pThis->Overlapped, &cbRead, TRUE /*fWait*/)
+            && GetLastError() != ERROR_OPERATION_ABORTED)
+            fRetEvents = RTPOLL_EVT_ERROR;
+
+        pThis->fIOPending    = false;
+        pThis->fZeroByteRead = false;
+    }
+
+    /* harvest events. */
+    fRetEvents |= rtPipePollCheck(pThis, fEvents);
+
+    /* update counters. */
+    pThis->cPolls--;
+    if (!pThis->cPolls)
+        pThis->hPollSet = NIL_RTPOLLSET;
+    pThis->cModeUsers--;
+
+    RTCritSectLeave(&pThis->CritSect);
+    return fRetEvents;
+}
+

trunk/src/VBox/Runtime/r3/win/poll-win.cpp

@@ -33 +33 @@
 *   Header Files                                                               *
 *******************************************************************************/
+#include <Windows.h>
+
 #include <iprt/poll.h>
 #include "internal/iprt.h"
 
+#include <iprt/asm.h>
 #include <iprt/assert.h>
 #include <iprt/err.h>
+#include <iprt/mem.h>
+#include <iprt/pipe.h>
+#include <iprt/string.h>
+#include <iprt/thread.h>
+#include <iprt/time.h>
+#include "internal/pipe.h"
 #include "internal/magics.h"
 
 
+/*******************************************************************************
+*   Structures and Typedefs                                                    *
+*******************************************************************************/
+/**
+ * Handle entry in a poll set.
+ */
+typedef struct RTPOLLSETHNDENT
+{
+    /** The handle type. */
+    RTHANDLETYPE    enmType;
+    /** The handle ID. */
+    uint32_t        id;
+    /** The events we're waiting for here. */
+    uint32_t        fEvents;
+    /** The handle union. */
+    RTHANDLEUNION   u;
+} RTPOLLSETHNDENT;
+/** Pointer to a handle entry. */
+typedef RTPOLLSETHNDENT *PRTPOLLSETHNDENT;
+
+
+/**
+ * Poll set data, Windows.
+ */
+typedef struct RTPOLLSETINTERNAL
+{
+    /** The magic value (RTPOLLSET_MAGIC). */
+    uint32_t            u32Magic;
+    /** Set when someone is polling or making changes. */
+    bool volatile       fBusy;
+
+    /** The number of valid handles in the set. */
+    uint32_t            cHandles;
+    /** The native handles. */
+    HANDLE              ahNative[MAXIMUM_WAIT_OBJECTS];
+    /** Array of handles and IDs. */
+    RTPOLLSETHNDENT     aHandles[MAXIMUM_WAIT_OBJECTS];
+} RTPOLLSETINTERNAL;
+
+
+/**
+ * Common worker for RTPoll and RTPollNoResume
+ */
+static int rtPollNoResumeWorker(RTPOLLSETINTERNAL *pThis, RTMSINTERVAL cMillies, uint32_t *pfEvents, uint32_t *pid)
+{
+    int rc;
+
+    if (RT_UNLIKELY(pThis->cHandles == 0 && cMillies == RT_INDEFINITE_WAIT))
+        return VERR_DEADLOCK;
+
+    /*
+     * Check for special case, RTThreadSleep...
+     */
+    uint32_t const  cHandles = pThis->cHandles;
+    if (cHandles == 0)
+    {
+        rc = RTThreadSleep(cMillies);
+        if (RT_SUCCESS(rc))
+            rc = VERR_TIMEOUT;
+        return rc;
+    }
+
+    /*
+     * Check + prepare the handles before waiting.
+     */
+    uint32_t        fEvents  = 0;
+    bool const      fNoWait  = cMillies == 0;
+    uint32_t        i;
+    for (i = 0; i < cHandles; i++)
+    {
+        switch (pThis->aHandles[i].enmType)
+        {
+            case RTHANDLETYPE_PIPE:
+                fEvents = rtPipePollStart(pThis->aHandles[i].u.hPipe, pThis, pThis->aHandles[i].fEvents, fNoWait);
+                break;
+
+            default:
+                AssertFailed();
+                fEvents = UINT32_MAX;
+                break;
+        }
+        if (fEvents)
+            break;
+    }
+    if (   fEvents
+        || fNoWait)
+    {
+
+        if (pid)
+            *pid = pThis->aHandles[i].id;
+        if (pfEvents)
+            *pfEvents = fEvents;
+        rc = !fEvents
+           ? VERR_TIMEOUT
+           : fEvents != UINT32_MAX
+           ? VINF_SUCCESS
+           : VERR_INTERNAL_ERROR_4;
+
+        /* clean up */
+        if (!fNoWait)
+            while (i-- > 0)
+            {
+                switch (pThis->aHandles[i].enmType)
+                {
+                    case RTHANDLETYPE_PIPE:
+                        rtPipePollDone(pThis->aHandles[i].u.hPipe, pThis->aHandles[i].fEvents);
+                        break;
+                    default:
+                        AssertFailed();
+                        break;
+                }
+            }
+
+        return rc;
+    }
+
+    /*
+     * Wait.
+     */
+    DWORD dwRc = WaitForMultipleObjectsEx(cHandles, &pThis->ahNative[0],
+                                          FALSE /*fWaitAll */,
+                                          cMillies == RT_INDEFINITE_WAIT ? INFINITE : cMillies,
+                                          TRUE /*fAlertable*/);
+    if (    dwRc >= WAIT_OBJECT_0
+        &&  dwRc <  WAIT_OBJECT_0 + cHandles)
+        rc = VINF_SUCCESS;
+    else if (dwRc == WAIT_TIMEOUT)
+        rc = VERR_TIMEOUT;
+    else if (dwRc == WAIT_IO_COMPLETION)
+        rc = VERR_INTERRUPTED;
+    else if (dwRc == WAIT_FAILED)
+        rc = RTErrConvertFromWin32(GetLastError());
+    else
+    {
+        AssertMsgFailed(("%u (%#x)\n", dwRc, dwRc));
+        rc = VERR_INTERNAL_ERROR_5;
+    }
+
+    /*
+     * Get event (if pending) and do wait cleanup.
+     */
+    i = cHandles;
+    while (i-- > 0)
+    {
+        fEvents = 0;
+        switch (pThis->aHandles[i].enmType)
+        {
+            case RTHANDLETYPE_PIPE:
+                fEvents = rtPipePollDone(pThis->aHandles[i].u.hPipe, pThis->aHandles[i].fEvents);
+                break;
+            default:
+                AssertFailed();
+                break;
+        }
+        if (fEvents)
+        {
+            Assert(fEvents != UINT32_MAX);
+            if (pfEvents)
+                *pfEvents = fEvents;
+            if (pid)
+                *pid = pThis->aHandles[i].id;
+            rc = VINF_SUCCESS;
+        }
+    }
+
+    return rc;
+}
+
+
 RTDECL(int) RTPoll(RTPOLLSET hPollSet, RTMSINTERVAL cMillies, uint32_t *pfEvents, uint32_t *pid)
 {
-    return VERR_NOT_IMPLEMENTED;
+    RTPOLLSETINTERNAL *pThis = hPollSet;
+    AssertPtrReturn(pThis, VERR_INVALID_HANDLE);
+    AssertReturn(pThis->u32Magic == RTPOLLSET_MAGIC, VERR_INVALID_HANDLE);
+    AssertPtrNull(pfEvents);
+    AssertPtrNull(pid);
+
+    /*
+     * Set the busy flag and do the job.
+     */
+    AssertReturn(ASMAtomicCmpXchgBool(&pThis->fBusy, true,  false), VERR_WRONG_ORDER);
+
+    int rc;
+    if (cMillies == RT_INDEFINITE_WAIT || cMillies == 0)
+    {
+        do rc = rtPollNoResumeWorker(pThis, cMillies, pfEvents, pid);
+        while (rc == VERR_INTERRUPTED);
+    }
+    else
+    {
+        uint64_t MsStart = RTTimeMilliTS();
+        rc = rtPollNoResumeWorker(pThis, cMillies, pfEvents, pid);
+        while (RT_UNLIKELY(rc == VERR_INTERRUPTED))
+        {
+            if (RTTimeMilliTS() - MsStart >= cMillies)
+            {
+                rc = VERR_TIMEOUT;
+                break;
+            }
+            rc = rtPollNoResumeWorker(pThis, cMillies, pfEvents, pid);
+        }
+    }
+
+    ASMAtomicWriteBool(&pThis->fBusy, false);
+
+    return rc;
 }
 
@@ -49 +261 @@
 RTDECL(int) RTPollNoResume(RTPOLLSET hPollSet, RTMSINTERVAL cMillies, uint32_t *pfEvents, uint32_t *pid)
 {
-    return VERR_NOT_IMPLEMENTED;
-}
-
-
-RTDECL(int)  RTPollSetCreate(PRTPOLLSET hPollSet)
-{
-    return VERR_NOT_IMPLEMENTED;
-}
-
-
-RTDECL(int)  RTPollSetDestroy(RTPOLLSET hPollSet)
-{
-    return VERR_NOT_IMPLEMENTED;
+    RTPOLLSETINTERNAL *pThis = hPollSet;
+    AssertPtrReturn(pThis, VERR_INVALID_HANDLE);
+    AssertReturn(pThis->u32Magic == RTPOLLSET_MAGIC, VERR_INVALID_HANDLE);
+    AssertPtrNull(pfEvents);
+    AssertPtrNull(pid);
+
+    /*
+     * Set the busy flag and do the job.
+     */
+    AssertReturn(ASMAtomicCmpXchgBool(&pThis->fBusy, true,  false), VERR_WRONG_ORDER);
+
+    int rc = rtPollNoResumeWorker(pThis, cMillies, pfEvents, pid);
+
+    ASMAtomicWriteBool(&pThis->fBusy, false);
+
+    return rc;
+}
+
+
+RTDECL(int) RTPollSetCreate(PRTPOLLSET phPollSet)
+{
+    AssertPtrReturn(phPollSet, VERR_INVALID_POINTER);
+    RTPOLLSETINTERNAL *pThis = (RTPOLLSETINTERNAL *)RTMemAllocZ(sizeof(RTPOLLSETINTERNAL));
+    if (!pThis)
+        return VERR_NO_MEMORY;
+
+    pThis->u32Magic             = RTPOLLSET_MAGIC;
+    pThis->fBusy                = false;
+    pThis->cHandles             = 0;
+    for (size_t i = 0; i < RT_ELEMENTS(pThis->ahNative); i++)
+        pThis->ahNative[i]     = INVALID_HANDLE_VALUE;
+
+    *phPollSet = pThis;
+    return VINF_SUCCESS;
+}
+
+
+RTDECL(int) RTPollSetDestroy(RTPOLLSET hPollSet)
+{
+    RTPOLLSETINTERNAL *pThis = hPollSet;
+    if (pThis == NIL_RTPOLLSET)
+        return VINF_SUCCESS;
+    AssertPtrReturn(pThis, VERR_INVALID_HANDLE);
+    AssertReturn(pThis->u32Magic == RTPOLLSET_MAGIC, VERR_INVALID_HANDLE);
+    AssertReturn(ASMAtomicCmpXchgBool(&pThis->fBusy, true,  false), VERR_WRONG_ORDER);
+
+    ASMAtomicWriteU32(&pThis->u32Magic, ~RTPOLLSET_MAGIC);
+    RTMemFree(pThis);
+
+    return VINF_SUCCESS;
 }
 
@@ -67 +316 @@
 RTDECL(int) RTPollSetAdd(RTPOLLSET hPollSet, PCRTHANDLE pHandle, uint32_t fEvents, uint32_t id)
 {
-    return VERR_NOT_IMPLEMENTED;
+    /*
+     * Validate the input (tedious).
+     */
+    RTPOLLSETINTERNAL *pThis = hPollSet;
+    AssertPtrReturn(pThis, VERR_INVALID_HANDLE);
+    AssertReturn(pThis->u32Magic == RTPOLLSET_MAGIC, VERR_INVALID_HANDLE);
+    AssertReturn(!(fEvents & ~RTPOLL_EVT_VALID_MASK), VERR_INVALID_PARAMETER);
+    AssertReturn(fEvents, VERR_INVALID_PARAMETER);
+    AssertReturn(id != UINT32_MAX, VERR_INVALID_PARAMETER);
+
+    if (!pHandle)
+        return VINF_SUCCESS;
+    AssertPtrReturn(pHandle, VERR_INVALID_POINTER);
+    AssertReturn(pHandle->enmType > RTHANDLETYPE_INVALID && pHandle->enmType < RTHANDLETYPE_END, VERR_INVALID_PARAMETER);
+
+    /*
+     * Set the busy flag and do the job.
+     */
+    AssertReturn(ASMAtomicCmpXchgBool(&pThis->fBusy, true,  false), VERR_WRONG_ORDER);
+
+    int     rc       = VINF_SUCCESS;
+    HANDLE  hNative  = INVALID_HANDLE_VALUE;
+    HANDLE  hNative2 = INVALID_HANDLE_VALUE;
+    switch (pHandle->enmType)
+    {
+        case RTHANDLETYPE_PIPE:
+            if (pHandle->u.hPipe != NIL_RTPIPE)
+                rc = rtPipePollGetHandles(pHandle->u.hPipe, fEvents, &hNative, &hNative2);
+            break;
+
+        case RTHANDLETYPE_SOCKET:
+            if (pHandle->u.hSocket != NIL_RTSOCKET)
+                rc = VERR_NOT_IMPLEMENTED;
+            break;
+
+        case RTHANDLETYPE_FILE:
+            AssertMsgFailed(("Files are always ready for reading/writing and thus not pollable. Use native APIs for special devices.\n"));
+            rc = VERR_POLL_HANDLE_NOT_POLLABLE;
+            break;
+
+        case RTHANDLETYPE_THREAD:
+            AssertMsgFailed(("Thread handles are currently not pollable\n"));
+            rc = VERR_POLL_HANDLE_NOT_POLLABLE;
+            break;
+
+        default:
+            AssertMsgFailed(("\n"));
+            rc = VERR_POLL_HANDLE_NOT_POLLABLE;
+            break;
+    }
+    if (   RT_SUCCESS(rc)
+        && hNative != INVALID_HANDLE_VALUE)
+    {
+        uint32_t const i = pThis->cHandles;
+
+        /* Check that the handle ID doesn't exist already. */
+        uint32_t j = i;
+        while (j-- > 0)
+            if (pThis->aHandles[j].id == id)
+            {
+                rc = VERR_POLL_HANDLE_ID_EXISTS;
+                break;
+            }
+
+        /* Check that we won't overflow the poll set now. */
+        if (    RT_SUCCESS(rc)
+            &&  i + 1 + (hNative2 != INVALID_HANDLE_VALUE) > RT_ELEMENTS(pThis->ahNative))
+            rc = VERR_POLL_SET_IS_FULL;
+        if (RT_SUCCESS(rc))
+        {
+            /* Add the handles to the two parallel arrays. */
+            pThis->ahNative[i]         = hNative;
+            pThis->aHandles[i].enmType = pHandle->enmType;
+            pThis->aHandles[i].u       = pHandle->u;
+            pThis->aHandles[i].id      = id;
+            pThis->aHandles[i].fEvents = fEvents;
+            if (hNative2 == INVALID_HANDLE_VALUE)
+                pThis->cHandles = i + 1;
+            else
+            {
+                pThis->ahNative[i + 1] = hNative2;
+                pThis->aHandles[i + 1] = pThis->aHandles[i];
+                pThis->cHandles = i + 2;
+            }
+
+            rc = VINF_SUCCESS;
+        }
+    }
+
+    ASMAtomicWriteBool(&pThis->fBusy, false);
+    return rc;
 }
 
@@ -73 +412 @@
 RTDECL(int) RTPollSetRemove(RTPOLLSET hPollSet, uint32_t id)
 {
-    return VERR_NOT_IMPLEMENTED;
+    /*
+     * Validate the input.
+     */
+    RTPOLLSETINTERNAL *pThis = hPollSet;
+    AssertPtrReturn(pThis, VERR_INVALID_HANDLE);
+    AssertReturn(pThis->u32Magic == RTPOLLSET_MAGIC, VERR_INVALID_HANDLE);
+    AssertReturn(id != UINT32_MAX, VERR_INVALID_PARAMETER);
+
+    /*
+     * Set the busy flag and do the job.
+     */
+    AssertReturn(ASMAtomicCmpXchgBool(&pThis->fBusy, true,  false), VERR_WRONG_ORDER);
+
+    int         rc = VERR_POLL_HANDLE_ID_NOT_FOUND;
+    uint32_t    i  = pThis->cHandles;
+    while (i-- > 0)
+        if (pThis->aHandles[i].id == id)
+        {
+            pThis->cHandles--;
+            size_t const cToMove = pThis->cHandles - i;
+            if (cToMove)
+            {
+                memmove(&pThis->aHandles[i], &pThis->aHandles[i + 1], cToMove * sizeof(pThis->aHandles[i]));
+                memmove(&pThis->ahNative[i], &pThis->ahNative[i + 1], cToMove * sizeof(pThis->ahNative[i]));
+            }
+            rc = VINF_SUCCESS;
+        }
+
+    ASMAtomicWriteBool(&pThis->fBusy, false);
+    return rc;
 }
 
@@ -79 +447 @@
 RTDECL(int) RTPollSetQueryHandle(RTPOLLSET hPollSet, uint32_t id, PRTHANDLE pHandle)
 {
-    return VERR_NOT_IMPLEMENTED;
+    /*
+     * Validate the input.
+     */
+    RTPOLLSETINTERNAL *pThis = hPollSet;
+    AssertPtrReturn(pThis, VERR_INVALID_HANDLE);
+    AssertReturn(pThis->u32Magic == RTPOLLSET_MAGIC, VERR_INVALID_HANDLE);
+    AssertReturn(id != UINT32_MAX, VERR_INVALID_PARAMETER);
+    AssertPtrNullReturn(pHandle, VERR_INVALID_POINTER);
+
+    /*
+     * Set the busy flag and do the job.
+     */
+    AssertReturn(ASMAtomicCmpXchgBool(&pThis->fBusy, true,  false), VERR_WRONG_ORDER);
+
+    int         rc = VERR_POLL_HANDLE_ID_NOT_FOUND;
+    uint32_t    i  = pThis->cHandles;
+    while (i-- > 0)
+        if (pThis->aHandles[i].id == id)
+        {
+            if (pHandle)
+            {
+                pHandle->enmType = pThis->aHandles[i].enmType;
+                pHandle->u       = pThis->aHandles[i].u;
+            }
+            rc = VINF_SUCCESS;
+            break;
+        }
+
+    ASMAtomicWriteBool(&pThis->fBusy, false);
+    return rc;
 }
 
@@ -85 +482 @@
 RTDECL(uint32_t) RTPollSetCount(RTPOLLSET hPollSet)
 {
-    return UINT32_MAX;
-}
-
+    /*
+     * Validate the input.
+     */
+    RTPOLLSETINTERNAL *pThis = hPollSet;
+    AssertPtrReturn(pThis, UINT32_MAX);
+    AssertReturn(pThis->u32Magic == RTPOLLSET_MAGIC, UINT32_MAX);
+
+    /*
+     * Set the busy flag and do the job.
+     */
+    AssertReturn(ASMAtomicCmpXchgBool(&pThis->fBusy, true,  false), UINT32_MAX);
+    uint32_t cHandles = pThis->cHandles;
+    ASMAtomicWriteBool(&pThis->fBusy, false);
+
+    return cHandles;
+}
+
+