Changeset 44462 in vbox

Timestamp:

Jan 30, 2013 12:44:14 PM (12 years ago)

Author:

vboxsync

svn:sync-xref-src-repo-rev:

83468

Message:

pipe-os2.cpp: Basic implementation passes the test case.

File:

• trunk/src/VBox/Runtime/r3/os2/pipe-os2.cpp (modified) (11 diffs)

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

@@ -5 +5 @@
 
 /*
- * Copyright (C) 2010 Oracle Corporation
+ * Copyright (C) 2010-2013 Oracle Corporation
  *
  * This file is part of VirtualBox Open Source Edition (OSE), as
@@ -29 +29 @@
 *   Header Files                                                               *
 *******************************************************************************/
+#define INCL_ERRORS
+#define INCL_DOSSEMAPHORES
+#include <os2.h>
+
 #include <iprt/pipe.h>
 #include "internal/iprt.h"
 
+#include <iprt/asm.h>
 #include <iprt/assert.h>
+#include <iprt/critsect.h>
 #include <iprt/err.h>
+#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 prefer. */
+#define RTPIPE_OS2_SIZE     _32K
+
+
+/*******************************************************************************
+*   Structures and Typedefs                                                    *
+*******************************************************************************/
+typedef struct RTPIPEINTERNAL
+{
+    /** Magic value (RTPIPE_MAGIC). */
+    uint32_t            u32Magic;
+    /** The pipe handle. */
+    HPIPE               hPipe;
+    /** Set if this is the read end, clear if it's the write end. */
+    bool                fRead;
+    /** Whether the pipe is in blocking or non-blocking mode. */
+    bool                fBlocking;
+    /** Set if the pipe is broken. */
+    bool                fBrokenPipe;
+    /** Usage counter. */
+    uint32_t            cUsers;
+
+    /** The event semaphore associated with the pipe. */
+    HEV                 hev;
+    /** The handle of the poll set currently polling on this pipe.
+     *  We can only have one poller at the time (lazy bird). */
+    RTPOLLSET           hPollSet;
+    /** Critical section protecting the above members.
+     * (Taking the lazy/simple approach.) */
+    RTCRITSECT          CritSect;
+
+} RTPIPEINTERNAL;
+
+
+/**
+ * Ensures that the pipe has a semaphore associated with it.
+ *
+ * @returns VBox status code.
+ * @param   pThis               The pipe.
+ */
+static int rtPipeOs2EnsureSem(RTPIPEINTERNAL *pThis)
+{
+    if (pThis->hev != NULLHANDLE)
+        return VINF_SUCCESS;
+
+    HEV     hev;
+    APIRET  orc = DosCreateEventSem(NULL, &hev, DC_SEM_SHARED, FALSE);
+    if (orc == NO_ERROR)
+    {
+        orc = DosSetNPipeSem(pThis->hPipe, (HSEM)hev, 1);
+        if (orc == NO_ERROR)
+        {
+            pThis->hev = hev;
+            return VINF_SUCCESS;
+        }
+
+        DosCloseEventSem(hev);
+    }
+    return RTErrConvertFromOS2(orc);
+}
 
 
 RTDECL(int)  RTPipeCreate(PRTPIPE phPipeRead, PRTPIPE phPipeWrite, uint32_t fFlags)
 {
-    return VERR_NOT_IMPLEMENTED;
+    AssertPtrReturn(phPipeRead, VERR_INVALID_POINTER);
+    AssertPtrReturn(phPipeWrite, VERR_INVALID_POINTER);
+    AssertReturn(!(fFlags & ~RTPIPE_C_VALID_MASK), VERR_INVALID_PARAMETER);
+
+    /*
+     * Try create and connect a pipe pair.
+     */
+    APIRET  orc;
+    HPIPE   hPipeR;
+    HFILE   hPipeW;
+    int     rc;
+    for (;;)
+    {
+        static volatile uint32_t    g_iNextPipe = 0;
+        char                        szName[128];
+        RTStrPrintf(szName, sizeof(szName), "\\pipe\\iprt-pipe-%u-%u", RTProcSelf(), ASMAtomicIncU32(&g_iNextPipe));
+
+        /*
+         * Create the read end of the pipe.
+         */
+        ULONG fPipeMode = 1 /*instance*/ | NP_TYPE_BYTE | NP_READMODE_BYTE | NP_NOWAIT;
+        ULONG fOpenMode = NP_ACCESS_DUPLEX | NP_WRITEBEHIND;
+        if (fFlags & RTPIPE_C_INHERIT_READ)
+            fOpenMode |= NP_INHERIT;
+        else
+            fOpenMode |= NP_NOINHERIT;
+        orc = DosCreateNPipe((PSZ)szName, &hPipeR, fOpenMode, fPipeMode, RTPIPE_OS2_SIZE, RTPIPE_OS2_SIZE, NP_DEFAULT_WAIT);
+        if (orc == NO_ERROR)
+        {
+            orc = DosConnectNPipe(hPipeR);
+            if (orc == ERROR_PIPE_NOT_CONNECTED || orc == NO_ERROR)
+            {
+                /*
+                 * Connect to the pipe (the write end), attach sem below.
+                 */
+                ULONG ulAction = 0;
+                ULONG fOpenW   = OPEN_ACTION_FAIL_IF_NEW | OPEN_ACTION_OPEN_IF_EXISTS;
+                ULONG fModeW   = OPEN_ACCESS_WRITEONLY | OPEN_SHARE_DENYNONE | OPEN_FLAGS_FAIL_ON_ERROR;
+                if (!(fFlags & RTPIPE_C_INHERIT_WRITE))
+                    fModeW |= OPEN_FLAGS_NOINHERIT;
+                orc = DosOpen((PSZ)szName, &hPipeW, &ulAction, 0 /*cbFile*/, FILE_NORMAL,
+                              fOpenW, fModeW, NULL /*peaop2*/);
+                if (orc == NO_ERROR)
+                    break;
+            }
+            DosClose(hPipeR);
+        }
+        if (   orc != ERROR_PIPE_BUSY     /* already exist - compatible */
+            && orc != ERROR_ACCESS_DENIED /* already exist - incompatible (?) */)
+            return RTErrConvertFromOS2(orc);
+        /* else: try again with a new name */
+    }
+
+    /*
+     * Create the two handles.
+     */
+    RTPIPEINTERNAL *pThisR = (RTPIPEINTERNAL *)RTMemAllocZ(sizeof(RTPIPEINTERNAL));
+    if (pThisR)
+    {
+        RTPIPEINTERNAL *pThisW = (RTPIPEINTERNAL *)RTMemAllocZ(sizeof(RTPIPEINTERNAL));
+        if (pThisW)
+        {
+            /* Crit sects. */
+            rc = RTCritSectInit(&pThisR->CritSect);
+            if (RT_SUCCESS(rc))
+            {
+                rc = RTCritSectInit(&pThisW->CritSect);
+                if (RT_SUCCESS(rc))
+                {
+                    /* Initialize the structures. */
+                    pThisR->u32Magic        = RTPIPE_MAGIC;
+                    pThisW->u32Magic        = RTPIPE_MAGIC;
+                    pThisR->hPipe           = hPipeR;
+                    pThisW->hPipe           = hPipeW;
+                    pThisR->hev             = NULLHANDLE;
+                    pThisW->hev             = NULLHANDLE;
+                    pThisR->fRead           = true;
+                    pThisW->fRead           = false;
+                    pThisR->fBlocking       = false;
+                    pThisW->fBlocking       = true;
+                    //pThisR->fBrokenPipe     = false;
+                    //pThisW->fBrokenPipe     = false;
+                    //pThisR->cUsers          = 0;
+                    //pThisW->cUsers          = 0;
+                    pThisR->hPollSet        = NIL_RTPOLLSET;
+                    pThisW->hPollSet        = NIL_RTPOLLSET;
+
+                    *phPipeRead  = pThisR;
+                    *phPipeWrite = pThisW;
+                    return VINF_SUCCESS;
+                }
+
+                RTCritSectDelete(&pThisR->CritSect);
+            }
+            RTMemFree(pThisW);
+        }
+        else
+            rc = VERR_NO_MEMORY;
+        RTMemFree(pThisR);
+    }
+    else
+        rc = VERR_NO_MEMORY;
+
+    /* Don't call DosDisConnectNPipe! */
+    DosClose(hPipeW);
+    DosClose(hPipeR);
+    return rc;
 }
 
@@ -44 +229 @@
 RTDECL(int)  RTPipeClose(RTPIPE hPipe)
 {
-    return VERR_NOT_IMPLEMENTED;
+    RTPIPEINTERNAL *pThis = hPipe;
+    if (pThis == NIL_RTPIPE)
+        return VINF_SUCCESS;
+    AssertPtrReturn(pThis, VERR_INVALID_PARAMETER);
+    AssertReturn(pThis->u32Magic == RTPIPE_MAGIC, VERR_INVALID_HANDLE);
+
+    /*
+     * Do the cleanup.
+     */
+    AssertReturn(ASMAtomicCmpXchgU32(&pThis->u32Magic, ~RTPIPE_MAGIC, RTPIPE_MAGIC), VERR_INVALID_HANDLE);
+    RTCritSectEnter(&pThis->CritSect);
+    Assert(pThis->cUsers == 0);
+
+    /* Don't call DosDisConnectNPipe! */
+    DosClose(pThis->hPipe);
+    pThis->hPipe = (HPIPE)-1;
+
+    if (pThis->hev != NULLHANDLE)
+    {
+        DosCloseEventSem(pThis->hev);
+        pThis->hev = NULLHANDLE;
+    }
+
+    RTCritSectLeave(&pThis->CritSect);
+    RTCritSectDelete(&pThis->CritSect);
+
+    RTMemFree(pThis);
+
+    return VINF_SUCCESS;
 }
 
@@ -50 +263 @@
 RTDECL(int)  RTPipeFromNative(PRTPIPE phPipe, RTHCINTPTR hNativePipe, uint32_t fFlags)
 {
-    return VERR_NOT_IMPLEMENTED;
-}
-
+    AssertPtrReturn(phPipe, VERR_INVALID_POINTER);
+    AssertReturn(!(fFlags & ~RTPIPE_N_VALID_MASK), VERR_INVALID_PARAMETER);
+    AssertReturn(!!(fFlags & RTPIPE_N_READ) != !!(fFlags & RTPIPE_N_WRITE), VERR_INVALID_PARAMETER);
+
+    /*
+     * Get and validate the pipe handle info.
+     */
+    HPIPE hNative = (HPIPE)hNativePipe;
+    ULONG ulType  = 0;
+    ULONG ulAttr  = 0;
+    APIRET orc = DosQueryHType(hNative, &ulType, &ulAttr);
+    AssertMsgReturn(orc == NO_ERROR, ("%d\n", orc), RTErrConvertFromOS2(orc));
+    AssertReturn((ulType & 0x7) == HANDTYPE_PIPE, VERR_INVALID_HANDLE);
+
+#if 0
+    union
+    {
+        PIPEINFO    PipeInfo;
+        uint8_t     abPadding[sizeof(PIPEINFO) + 127];
+    } Buf;
+    orc = DosQueryNPipeInfo(hNative, 1, &Buf, sizeof(Buf));
+    if (orc != NO_ERROR)
+    {
+        /* Sorry, anonymous pips are not supported. */
+        AssertMsgFailed(("%d\n", orc));
+        return VERR_INVALID_HANDLE;
+    }
+    AssertReturn(Buf.PipeInfo.cbMaxInst == 1, VERR_INVALID_HANDLE);
+#endif
+
+    ULONG fPipeState = 0;
+    orc = DosQueryNPHState(hNative, &fPipeState);
+    if (orc != NO_ERROR)
+    {
+        /* Sorry, anonymous pips are not supported. */
+        AssertMsgFailed(("%d\n", orc));
+        return VERR_INVALID_HANDLE;
+    }
+    AssertReturn(!(fPipeState & NP_TYPE_MESSAGE), VERR_INVALID_HANDLE);
+    AssertReturn(!(fPipeState & NP_READMODE_MESSAGE), VERR_INVALID_HANDLE);
+    AssertReturn((fPipeState & 0xff) == 1, VERR_INVALID_HANDLE);
+
+    ULONG fFileState = 0;
+    orc = DosQueryFHState(hNative, &fFileState);
+    AssertMsgReturn(orc == NO_ERROR, ("%d\n", orc), VERR_INVALID_HANDLE);
+    AssertMsgReturn(   (fFileState & 0x3) == (fFlags & RTPIPE_N_READ ? OPEN_ACCESS_READONLY : OPEN_ACCESS_WRITEONLY)
+                    || (fFileState & 0x3) == OPEN_ACCESS_READWRITE
+                    , ("%#x\n", fFileState), VERR_INVALID_HANDLE);
+
+    /*
+     * Looks kind of OK. Fix the inherit flag.
+     */
+    orc = DosSetFHState(hNative,   (fFileState & (OPEN_FLAGS_WRITE_THROUGH | OPEN_FLAGS_FAIL_ON_ERROR | OPEN_FLAGS_NO_CACHE))
+                                 | (fFlags & RTPIPE_N_INHERIT ? 0 : OPEN_FLAGS_NOINHERIT));
+    AssertMsgReturn(orc == NO_ERROR, ("%d\n", orc), RTErrConvertFromOS2(orc));
+
+
+    /*
+     * Create a handle so we can try rtPipeQueryInfo on it
+     * and see if we need to duplicate it to make that call work.
+     */
+    RTPIPEINTERNAL *pThis = (RTPIPEINTERNAL *)RTMemAllocZ(sizeof(RTPIPEINTERNAL));
+    if (!pThis)
+        return VERR_NO_MEMORY;
+    int rc = RTCritSectInit(&pThis->CritSect);
+    if (RT_SUCCESS(rc))
+    {
+        pThis->u32Magic        = RTPIPE_MAGIC;
+        pThis->hPipe           = hNative;
+        pThis->hev             = NULLHANDLE;
+        pThis->fRead           = !!(fFlags & RTPIPE_N_READ);
+        pThis->fBlocking       = !(fPipeState & NP_NOWAIT);
+        //pThis->fBrokenPipe     = false;
+        //pThis->cUsers          = 0;
+        pThis->hPollSet        = NIL_RTPOLLSET;
+
+        *phPipe = pThis;
+        return VINF_SUCCESS;
+
+        //RTCritSectDelete(&pThis->CritSect);
+    }
+    RTMemFree(pThis);
+    return rc;
+}
 
 RTDECL(RTHCINTPTR) RTPipeToNative(RTPIPE hPipe)
 {
-    return -1;
+    RTPIPEINTERNAL *pThis = hPipe;
+    AssertPtrReturn(pThis, -1);
+    AssertReturn(pThis->u32Magic == RTPIPE_MAGIC, -1);
+
+    return (RTHCINTPTR)pThis->hPipe;
+}
+
+/**
+ * Prepare blocking mode.
+ *
+ * @returns IPRT status code.
+ * @retval  VERR_WRONG_ORDER if simultaneous non-blocking and blocking access is
+ *          attempted.
+ *
+ * @param   pThis               The pipe handle.
+ *
+ * @remarks Caller owns the critical section.
+ */
+static int rtPipeTryBlocking(RTPIPEINTERNAL *pThis)
+{
+    if (!pThis->fBlocking)
+    {
+        if (pThis->cUsers != 0)
+            return VERR_WRONG_ORDER;
+
+        APIRET orc = DosSetNPHState(pThis->hPipe, NP_WAIT | NP_READMODE_BYTE);
+        if (orc != NO_ERROR)
+        {
+            if (orc != ERROR_BROKEN_PIPE && orc != ERROR_PIPE_NOT_CONNECTED)
+                return RTErrConvertFromOS2(orc);
+            pThis->fBrokenPipe = true;
+        }
+        pThis->fBlocking = true;
+    }
+
+    pThis->cUsers++;
+    return VINF_SUCCESS;
+}
+
+
+/**
+ * Prepare non-blocking mode.
+ *
+ * @returns IPRT status code.
+ * @retval  VERR_WRONG_ORDER if simultaneous non-blocking and blocking access is
+ *          attempted.
+ *
+ * @param   pThis               The pipe handle.
+ */
+static int rtPipeTryNonBlocking(RTPIPEINTERNAL *pThis)
+{
+    if (pThis->fBlocking)
+    {
+        if (pThis->cUsers != 0)
+            return VERR_WRONG_ORDER;
+
+        APIRET orc = DosSetNPHState(pThis->hPipe, NP_NOWAIT | NP_READMODE_BYTE);
+        if (orc != NO_ERROR)
+        {
+            if (orc != ERROR_BROKEN_PIPE && orc != ERROR_PIPE_NOT_CONNECTED)
+                return RTErrConvertFromOS2(orc);
+            pThis->fBrokenPipe = true;
+        }
+        pThis->fBlocking = false;
+    }
+
+    pThis->cUsers++;
+    return VINF_SUCCESS;
+}
+
+
+/**
+ * Checks if the read pipe has been broken.
+ *
+ * @returns true if broken, false if no.
+ * @param   pThis               The pipe handle (read).
+ */
+static bool rtPipeOs2IsBroken(RTPIPEINTERNAL *pThis)
+{
+    Assert(pThis->fRead);
+
+#if 0
+    /*
+     * Query it via the semaphore. Not sure how fast this is...
+     */
+    PIPESEMSTATE aStates[3]; RT_ZERO(aStates);
+    APIRET orc = DosQueryNPipeSemState(pThis->hev, &aStates[0], sizeof(aStates));
+    if (orc == NO_ERROR)
+    {
+        if (aStates[0].fStatus == NPSS_CLOSE)
+            return true;
+        if (aStates[0].fStatus == NPSS_RDATA)
+            return false;
+    }
+    AssertMsgFailed(("%d / %d\n", orc, aStates[0].fStatus));
+
+    /*
+     * Fall back / alternative method.
+     */
+#endif
+    ULONG       cbActual = 0;
+    ULONG       ulState  = 0;
+    AVAILDATA   Avail    = { 0, 0 };
+    APIRET orc = DosPeekNPipe(pThis->hPipe, NULL, 0, &cbActual, &Avail, &ulState);
+    if (orc != NO_ERROR)
+    {
+        if (orc != ERROR_PIPE_BUSY)
+            AssertMsgFailed(("%d\n", orc));
+        return false;
+    }
+
+    return ulState != NP_STATE_CONNECTED;
 }
 
@@ -62 +467 @@
 RTDECL(int) RTPipeRead(RTPIPE hPipe, void *pvBuf, size_t cbToRead, size_t *pcbRead)
 {
-    return VERR_NOT_IMPLEMENTED;
+    RTPIPEINTERNAL *pThis = hPipe;
+    AssertPtrReturn(pThis, VERR_INVALID_HANDLE);
+    AssertReturn(pThis->u32Magic == RTPIPE_MAGIC, VERR_INVALID_HANDLE);
+    AssertReturn(pThis->fRead, VERR_ACCESS_DENIED);
+    AssertPtr(pcbRead);
+    AssertPtr(pvBuf);
+
+    int rc = RTCritSectEnter(&pThis->CritSect);
+    if (RT_SUCCESS(rc))
+    {
+        rc = rtPipeTryNonBlocking(pThis);
+        if (RT_SUCCESS(rc))
+        {
+            RTCritSectLeave(&pThis->CritSect);
+
+            ULONG cbActual = 0;
+            APIRET orc = DosRead(pThis->hPipe, pvBuf, cbToRead, &cbActual);
+            if (orc == NO_ERROR)
+            {
+                if (cbActual || !cbToRead || !rtPipeOs2IsBroken(pThis))
+                    *pcbRead = cbActual;
+                else
+                    rc = VERR_BROKEN_PIPE;
+            }
+            else if (orc == ERROR_NO_DATA)
+            {
+                *pcbRead = 0;
+                rc = VINF_TRY_AGAIN;
+            }
+            else
+                rc = RTErrConvertFromOS2(orc);
+
+            RTCritSectEnter(&pThis->CritSect);
+            if (rc == VERR_BROKEN_PIPE)
+                pThis->fBrokenPipe = true;
+            pThis->cUsers--;
+        }
+        else
+            rc = VERR_WRONG_ORDER;
+        RTCritSectLeave(&pThis->CritSect);
+    }
+    return rc;
 }
 
@@ -68 +514 @@
 RTDECL(int) RTPipeReadBlocking(RTPIPE hPipe, void *pvBuf, size_t cbToRead, size_t *pcbRead)
 {
-    return VERR_NOT_IMPLEMENTED;
+    RTPIPEINTERNAL *pThis = hPipe;
+    AssertPtrReturn(pThis, VERR_INVALID_HANDLE);
+    AssertReturn(pThis->u32Magic == RTPIPE_MAGIC, VERR_INVALID_HANDLE);
+    AssertReturn(pThis->fRead, VERR_ACCESS_DENIED);
+    AssertPtr(pvBuf);
+
+    int rc = RTCritSectEnter(&pThis->CritSect);
+    if (RT_SUCCESS(rc))
+    {
+        rc = rtPipeTryBlocking(pThis);
+        if (RT_SUCCESS(rc))
+        {
+            RTCritSectLeave(&pThis->CritSect);
+
+            size_t cbTotalRead = 0;
+            while (cbToRead > 0)
+            {
+                ULONG  cbActual = 0;
+                APIRET orc = DosRead(pThis->hPipe, pvBuf, cbToRead, &cbActual);
+                if (orc != NO_ERROR)
+                {
+                    rc = RTErrConvertFromOS2(orc);
+                    break;
+                }
+                if (!cbActual && rtPipeOs2IsBroken(pThis))
+                {
+                    rc = VERR_BROKEN_PIPE;
+                    break;
+                }
+
+                /* advance */
+                pvBuf        = (char *)pvBuf + cbActual;
+                cbTotalRead += cbActual;
+                cbToRead    -= cbActual;
+            }
+
+            if (pcbRead)
+            {
+                *pcbRead = cbTotalRead;
+                if (   RT_FAILURE(rc)
+                    && cbTotalRead)
+                    rc = VINF_SUCCESS;
+            }
+
+            RTCritSectEnter(&pThis->CritSect);
+            if (rc == VERR_BROKEN_PIPE)
+                pThis->fBrokenPipe = true;
+            pThis->cUsers--;
+        }
+        else
+            rc = VERR_WRONG_ORDER;
+        RTCritSectLeave(&pThis->CritSect);
+    }
+    return rc;
+}
+
+
+/**
+ * Gets the available write buffer size of the pipe.
+ *
+ * @returns Number of bytes, 1 on failure.
+ * @param   pThis               The pipe handle.
+ */
+static ULONG rtPipeOs2GetSpace(RTPIPEINTERNAL *pThis)
+{
+    Assert(!pThis->fRead);
+
+#if 0 /* Not sure which is more efficient, neither are really optimal, I fear. */
+    /*
+     * Query via semaphore state.
+     * This will walk the list of active named pipes...
+     */
+    /** @todo Check how hev and hpipe are associated, if complicated, use the
+     *        alternative method below. */
+    PIPESEMSTATE aStates[3]; RT_ZERO(aStates);
+    APIRET orc = DosQueryNPipeSemState((HSEM)pThis->hev, &aStates[0], sizeof(aStates));
+    if (orc == NO_ERROR)
+    {
+        if (aStates[0].fStatus == NPSS_WSPACE)
+            return aStates[0].usAvail;
+        if (aStates[1].fStatus == NPSS_WSPACE)
+            return aStates[1].usAvail;
+        return 0;
+    }
+    AssertMsgFailed(("%d / %d\n", orc, aStates[0].fStatus));
+
+#else
+    /*
+     * Query via the pipe info.
+     * This will have to lookup and store the pipe name.
+     */
+    union
+    {
+        PIPEINFO    PipeInfo;
+        uint8_t     abPadding[sizeof(PIPEINFO) + 127];
+    } Buf;
+    APIRET orc = DosQueryNPipeInfo(pThis->hPipe, 1, &Buf, sizeof(Buf));
+    if (orc == NO_ERROR)
+        return Buf.PipeInfo.cbOut;
+    AssertMsgFailed(("%d\n", orc));
+#endif
+
+    return 1;
 }
 
@@ -74 +622 @@
 RTDECL(int) RTPipeWrite(RTPIPE hPipe, const void *pvBuf, size_t cbToWrite, size_t *pcbWritten)
 {
-    return VERR_NOT_IMPLEMENTED;
+    RTPIPEINTERNAL *pThis = hPipe;
+    AssertPtrReturn(pThis, VERR_INVALID_HANDLE);
+    AssertReturn(pThis->u32Magic == RTPIPE_MAGIC, VERR_INVALID_HANDLE);
+    AssertReturn(!pThis->fRead, VERR_ACCESS_DENIED);
+    AssertPtr(pcbWritten);
+    AssertPtr(pvBuf);
+
+    int rc = RTCritSectEnter(&pThis->CritSect);
+    if (RT_SUCCESS(rc))
+    {
+        rc = rtPipeTryNonBlocking(pThis);
+        if (RT_SUCCESS(rc))
+        {
+            if (cbToWrite > 0)
+            {
+                ULONG  cbActual = 0;
+                APIRET orc = DosWrite(pThis->hPipe, pvBuf, cbToWrite, &cbActual);
+                if (orc == NO_ERROR && cbActual == 0)
+                {
+                    /* Retry with the request adjusted to the available buffer space. */
+                    ULONG cbAvail = rtPipeOs2GetSpace(pThis);
+                    orc = DosWrite(pThis->hPipe, pvBuf, RT_MIN(cbAvail, cbToWrite), &cbActual);
+                }
+
+                if (orc == NO_ERROR)
+                {
+                    *pcbWritten = cbActual;
+                    if (cbActual == 0)
+                        rc = VINF_TRY_AGAIN;
+                }
+                else
+                {
+                    rc = RTErrConvertFromOS2(orc);
+                    if (rc == VERR_PIPE_NOT_CONNECTED)
+                        rc = VERR_BROKEN_PIPE;
+                }
+            }
+            else
+                *pcbWritten = 0;
+
+            if (rc == VERR_BROKEN_PIPE)
+                pThis->fBrokenPipe = true;
+            pThis->cUsers--;
+        }
+        else
+            rc = VERR_WRONG_ORDER;
+        RTCritSectLeave(&pThis->CritSect);
+    }
+    return rc;
 }
 
@@ -80 +676 @@
 RTDECL(int) RTPipeWriteBlocking(RTPIPE hPipe, const void *pvBuf, size_t cbToWrite, size_t *pcbWritten)
 {
-    return VERR_NOT_IMPLEMENTED;
+    RTPIPEINTERNAL *pThis = hPipe;
+    AssertPtrReturn(pThis, VERR_INVALID_HANDLE);
+    AssertReturn(pThis->u32Magic == RTPIPE_MAGIC, VERR_INVALID_HANDLE);
+    AssertReturn(!pThis->fRead, VERR_ACCESS_DENIED);
+    AssertPtr(pvBuf);
+    AssertPtrNull(pcbWritten);
+
+    int rc = RTCritSectEnter(&pThis->CritSect);
+    if (RT_SUCCESS(rc))
+    {
+        rc = rtPipeTryBlocking(pThis);
+        if (RT_SUCCESS(rc))
+        {
+            RTCritSectLeave(&pThis->CritSect);
+
+            size_t cbTotalWritten = 0;
+            while (cbToWrite > 0)
+            {
+                ULONG cbActual = 0;
+                APIRET orc = DosWrite(pThis->hPipe, pvBuf, cbToWrite, &cbActual);
+                if (orc != NO_ERROR)
+                {
+                    rc = RTErrConvertFromOS2(orc);
+                    if (rc == VERR_PIPE_NOT_CONNECTED)
+                        rc = VERR_BROKEN_PIPE;
+                    break;
+                }
+                pvBuf           = (char const *)pvBuf + cbActual;
+                cbToWrite      -= cbActual;
+                cbTotalWritten += cbActual;
+            }
+
+            if (pcbWritten)
+            {
+                *pcbWritten = cbTotalWritten;
+                if (   RT_FAILURE(rc)
+                    && cbTotalWritten)
+                    rc = VINF_SUCCESS;
+            }
+
+            RTCritSectEnter(&pThis->CritSect);
+            if (rc == VERR_BROKEN_PIPE)
+                pThis->fBrokenPipe = true;
+            pThis->cUsers--;
+        }
+        else
+            rc = VERR_WRONG_ORDER;
+        RTCritSectLeave(&pThis->CritSect);
+    }
+    return rc;
 }
 
@@ -86 +731 @@
 RTDECL(int) RTPipeFlush(RTPIPE hPipe)
 {
-    return VERR_NOT_IMPLEMENTED;
+    RTPIPEINTERNAL *pThis = hPipe;
+    AssertPtrReturn(pThis, VERR_INVALID_HANDLE);
+    AssertReturn(pThis->u32Magic == RTPIPE_MAGIC, VERR_INVALID_HANDLE);
+    AssertReturn(!pThis->fRead, VERR_ACCESS_DENIED);
+
+    APIRET orc = DosResetBuffer(pThis->hPipe);
+    if (orc != NO_ERROR)
+    {
+        int rc = RTErrConvertFromOS2(orc);
+        if (rc == VERR_BROKEN_PIPE)
+        {
+            RTCritSectEnter(&pThis->CritSect);
+            pThis->fBrokenPipe = true;
+            RTCritSectLeave(&pThis->CritSect);
+        }
+        return rc;
+    }
+    return VINF_SUCCESS;
 }
 
@@ -92 +754 @@
 RTDECL(int) RTPipeSelectOne(RTPIPE hPipe, RTMSINTERVAL cMillies)
 {
-    return VERR_NOT_IMPLEMENTED;
+    RTPIPEINTERNAL *pThis = hPipe;
+    AssertPtrReturn(pThis, VERR_INVALID_HANDLE);
+    AssertReturn(pThis->u32Magic == RTPIPE_MAGIC, VERR_INVALID_HANDLE);
+
+    uint64_t const StartMsTS = RTTimeMilliTS();
+
+    int rc = RTCritSectEnter(&pThis->CritSect);
+    if (RT_FAILURE(rc))
+        return rc;
+    rc = rtPipeOs2EnsureSem(pThis);
+    if (RT_SUCCESS(rc))
+    {
+        for (unsigned iLoop = 0;; iLoop++)
+        {
+            /*
+             * Check the handle state.
+             */
+            PIPESEMSTATE aStates[4]; RT_ZERO(aStates);
+            APIRET orc = DosQueryNPipeSemState((HSEM)pThis->hev, &aStates[0], sizeof(aStates));
+            if (orc != NO_ERROR)
+            {
+                rc = RTErrConvertFromOS2(orc);
+                break;
+            }
+            int i = 0;
+            if (pThis->fRead)
+                while (aStates[i].fStatus == NPSS_WSPACE)
+                    i++;
+            else
+                while (aStates[i].fStatus == NPSS_RDATA)
+                    i++;
+            if (aStates[i].fStatus == NPSS_CLOSE)
+                break;
+            Assert(aStates[i].fStatus == NPSS_WSPACE || aStates[i].fStatus == NPSS_RDATA || aStates[i].fStatus == NPSS_EOI);
+            if (   aStates[i].fStatus != NPSS_EOI
+                && aStates[0].usAvail > 0)
+                break;
+
+            /*
+             * Check for timeout.
+             */
+            ULONG cMsMaxWait = SEM_INDEFINITE_WAIT;
+            if (cMillies != RT_INDEFINITE_WAIT)
+            {
+                uint64_t cElapsed = RTTimeMilliTS() - StartMsTS;
+                if (cElapsed >= cMillies)
+                {
+                    rc = VERR_TIMEOUT;
+                    break;
+                }
+                cMsMaxWait = cMillies - (uint32_t)cElapsed;
+            }
+
+            /*
+             * Wait.
+             */
+            RTCritSectLeave(&pThis->CritSect);
+            orc = DosWaitEventSem(pThis->hev, cMsMaxWait);
+            RTCritSectEnter(&pThis->CritSect);
+            if (orc != NO_ERROR && orc != ERROR_TIMEOUT && orc != ERROR_SEM_TIMEOUT )
+            {
+                rc = RTErrConvertFromOS2(orc);
+                break;
+            }
+        }
+
+        if (rc == VERR_BROKEN_PIPE)
+            pThis->fBrokenPipe = true;
+    }
+
+    RTCritSectLeave(&pThis->CritSect);
+    return rc;
 }
 
@@ -98 +831 @@
 RTDECL(int) RTPipeQueryReadable(RTPIPE hPipe, size_t *pcbReadable)
 {
-    return VERR_NOT_IMPLEMENTED;
-}
-
+    RTPIPEINTERNAL *pThis = hPipe;
+    AssertPtrReturn(pThis, VERR_INVALID_HANDLE);
+    AssertReturn(pThis->u32Magic == RTPIPE_MAGIC, VERR_INVALID_HANDLE);
+    AssertReturn(pThis->fRead, VERR_PIPE_NOT_READ);
+    AssertPtrReturn(pcbReadable, VERR_INVALID_POINTER);
+
+    int rc = RTCritSectEnter(&pThis->CritSect);
+    if (RT_FAILURE(rc))
+        return rc;
+
+    ULONG       cbActual = 0;
+    ULONG       ulState  = 0;
+    AVAILDATA   Avail    = { 0, 0 };
+    APIRET orc = DosPeekNPipe(pThis->hPipe, NULL, 0, &cbActual, &Avail, &ulState);
+    if (orc == NO_ERROR)
+    {
+        if (Avail.cbpipe > 0 || ulState == NP_STATE_CONNECTED)
+            *pcbReadable = Avail.cbpipe;
+        else
+            rc = VERR_PIPE_NOT_CONNECTED; /*??*/
+    }
+    else
+        rc = RTErrConvertFromOS2(orc);
+
+    RTCritSectLeave(&pThis->CritSect);
+    return rc;
+}
+
+
+#if 0
+/**
+ * Internal RTPollSetAdd helper that returns the handle 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   ph                  where to put the primary handle.
+ */
+int rtPipePollGetHandle(RTPIPE hPipe, uint32_t fEvents, PHANDLE ph)
+{
+    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);
+
+    /* 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. */
+    *ph = pThis->Overlapped.hEvent;
+    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
+                    && (fEvents & RTPOLL_EVT_WRITE)
+                    && (   Info.WriteQuotaAvailable > 0
+                        || Info.OutboundQuota == 0)
+                    )
+                    fRetEvents |= RTPOLL_EVT_WRITE;
+            }
+            else if (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   fFinalEntry         Set if this is the final entry for this handle
+ *                              in this poll set.  This can be used for dealing
+ *                              with duplicate entries.
+ * @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 fFinalEntry, 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->cUsers   == 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->cUsers++;
+                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.
+ * @param   fFinalEntry         Set if this is the final entry for this handle
+ *                              in this poll set.  This can be used for dealing
+ *                              with duplicate entries.  Only keep in mind that
+ *                              this method is called in reverse order, so the
+ *                              first call will have this set (when the entire
+ *                              set was processed).
+ * @param   fHarvestEvents      Set if we should check for pending events.
+ */
+uint32_t rtPipePollDone(RTPIPE hPipe, uint32_t fEvents, bool fFinalEntry, bool fHarvestEvents)
+{
+    RTPIPEINTERNAL *pThis = hPipe;
+    AssertPtrReturn(pThis, 0);
+    AssertReturn(pThis->u32Magic == RTPIPE_MAGIC, 0);
+
+    int rc = RTCritSectEnter(&pThis->CritSect);
+    AssertRCReturn(rc, 0);
+
+    Assert(pThis->cUsers > 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->cUsers--;
+    if (!pThis->cUsers)
+        pThis->hPollSet = NIL_RTPOLLSET;
+
+    RTCritSectLeave(&pThis->CritSect);
+    return fRetEvents;
+}
+#endif