path: root/rts/win32/MIOManager.c

/* MIOManager.c
 *
 * Non-blocking / asynchronous I/O for Win32.
 *
 * (c) sof, 2002-2003.
 *
 * NOTE: This is the MIO manager, only used for --io-manager=posix.
 *       For the WINIO manager see base in the GHC.Event modules.
 */

#if !defined(THREADED_RTS)

#include "Rts.h"
#include "RtsUtils.h"
#include "MIOManager.h"
#include "WorkQueue.h"
#include "ConsoleHandler.h"
#include <stdio.h>
#include <stdlib.h>
#include <io.h>
#include <winsock.h>
#include <process.h>
#include <errno.h>

/*
 * Internal state maintained by the IO manager.
 */
typedef struct IOManagerState {
    Mutex            manLock;
    WorkQueue*       workQueue;
    int              queueSize;
    int              numWorkers;
    int              workersIdle;
    HANDLE           hExitEvent;
    unsigned int     requestID;
    /* fields for keeping track of active WorkItems */
    Mutex            active_work_lock;
    WorkItem*        active_work_items;
    UINT             sleepResolution;
} IOManagerState;

/* ToDo: wrap up this state via a IOManager handle instead? */
static IOManagerState* ioMan;

static void RegisterWorkItem  ( IOManagerState* iom, WorkItem* wi);
static void DeregisterWorkItem( IOManagerState* iom, WorkItem* wi);

/*
 * The routine executed by each worker thread.
 */
static
unsigned
WINAPI
IOWorkerProc(PVOID param)
{
    HANDLE  hWaits[2];
    DWORD   rc;
    IOManagerState* iom = (IOManagerState*)param;
    WorkQueue* pq = iom->workQueue;
    WorkItem*  work;
    int        len = 0, fd = 0;
    DWORD      errCode = 0;
    void*      complData;

    hWaits[0] = (HANDLE)iom->hExitEvent;
    hWaits[1] = GetWorkQueueHandle(pq);

    while (1) {
        // The error code is communicated back on completion of request; reset.
        errCode = 0;

        OS_ACQUIRE_LOCK(&iom->manLock);
        /* Signal that the worker is idle.
         *
         * 'workersIdle' is used when determining whether or not to
         * increase the worker thread pool when adding a new request.
         * (see addIORequest().)
         */
        iom->workersIdle++;
        OS_RELEASE_LOCK(&iom->manLock);

        /*
         * A possible future refinement is to make long-term idle threads
         * wake up and decide to shut down should the number of idle threads
         * be above some threshold.
         *
         */
        rc = WaitForMultipleObjects( 2, hWaits, false, INFINITE );

        if (rc == WAIT_OBJECT_0) {
            // we received the exit event
            OS_ACQUIRE_LOCK(&iom->manLock);
            ioMan->numWorkers--;
            OS_RELEASE_LOCK(&iom->manLock);
            return 0;
        }

        OS_ACQUIRE_LOCK(&iom->manLock);
        /* Signal that the thread is 'non-idle' and about to consume
         * a work item.
         */
        iom->workersIdle--;
        iom->queueSize--;
        OS_RELEASE_LOCK(&iom->manLock);

        if ( rc == (WAIT_OBJECT_0 + 1) ) {
            /* work item available, fetch it. */
            if (FetchWork(pq,(void**)&work)) {
                work->abandonOp = 0;
                RegisterWorkItem(iom,work);
                if ( work->workKind & WORKER_READ ) {
                    if ( work->workKind & WORKER_FOR_SOCKET ) {
                        len = recv(work->workData.ioData.fd,
                                   work->workData.ioData.buf,
                                   work->workData.ioData.len,
                                   0);
                        if (len == SOCKET_ERROR) {
                            errCode = WSAGetLastError();
                        }
                    } else {
                        while (1) {
                        // Do the read(), with extra-special handling for Ctrl+C
                        len = read(work->workData.ioData.fd,
                                   work->workData.ioData.buf,
                                   work->workData.ioData.len);
                        if ( len == 0 && work->workData.ioData.len != 0 ) {
                            /* Given the following scenario:
                             * - a console handler has been registered
                             *   that handles Ctrl+C events.
                             * - we've not tweaked the 'console mode'
                             *   settings to turn on
                             *   ENABLE_PROCESSED_INPUT.
                             * - we're blocked waiting on input from
                                 standard input.
                             * - the user hits Ctrl+C.
                             *
                             * The OS will invoke the console handler
                             * (in a separate OS thread), and the
                             * above read() (i.e., under the hood, a
                             * ReadFile() op) returns 0, with the
                             * error set to
                             * ERROR_OPERATION_ABORTED. We don't want
                             * to percolate this error condition back
                             * to the Haskell user.  Do this by
                             * waiting for the completion of the
                             * Haskell console handler. If upon
                             * completion of the console handler
                             * routine, the Haskell thread that issued
                             * the request is found to have been
                             * thrown an exception, the worker
                             * abandons the request (since that's what
                             * the Haskell thread has done.) If the
                             * Haskell thread hasn't been interrupted,
                             * the worker retries the read request as
                             * if nothing happened.
                             */
                            if ( (GetLastError()) == ERROR_OPERATION_ABORTED ) {
                                /* For now, only abort when dealing
                                 * with the standard input handle.
                                 * i.e., for all others, an error is
                                 * raised.
                                 */
                                HANDLE h =
                                    (HANDLE)GetStdHandle(STD_INPUT_HANDLE);
                                int iofd = work->workData.ioData.fd;
                                if ( _get_osfhandle(iofd) == (intptr_t)h ) {
                                    if (rts_waitConsoleHandlerCompletion()) {
                                        /* If the Scheduler has set
                                         * work->abandonOp, the
                                         * Haskell thread has been
                                         * thrown an exception (=> the
                                         * worker must abandon this
                                         * request.)  We test for this
                                         * below before invoking the
                                         * on-completion routine.
                                         */
                                        if (work->abandonOp) {
                                            break;
                                        } else {
                                            continue;
                                        }
                                    }
                                } else {
                                    break; /* Treat it like an error */
                                }
                            } else {
                                break;
                            }
                        } else {
                            break;
                        }
                        }
                        if (len == -1) { errCode = errno; }
                    }
                    complData = work->workData.ioData.buf;
                    fd = work->workData.ioData.fd;
                } else if ( work->workKind & WORKER_WRITE ) {
                    if ( work->workKind & WORKER_FOR_SOCKET ) {
                        len = send(work->workData.ioData.fd,
                                   work->workData.ioData.buf,
                                   work->workData.ioData.len,
                                   0);
                        if (len == SOCKET_ERROR) {
                            errCode = WSAGetLastError();
                        }
                    } else {
                        len = write(work->workData.ioData.fd,
                                    work->workData.ioData.buf,
                                    work->workData.ioData.len);
                        if (len == -1) {
                            errCode = errno;
                            // write() gets errno wrong for
                            // ERROR_NO_DATA, we have to fix it here:
                            if (errCode == EINVAL &&
                                GetLastError() == ERROR_NO_DATA) {
                                errCode = EPIPE;
                            }
                        }
                    }
                    complData = work->workData.ioData.buf;
                    fd = work->workData.ioData.fd;
                } else if ( work->workKind & WORKER_DELAY ) {
                    /* Approximate implementation of threadDelay;
                     *
                     * Note: Sleep() is in milliseconds, not micros.
                     *
                     * MSDN says of Sleep:
                     *   If dwMilliseconds is greater than one tick
                     *   but less than two, the wait can be anywhere
                     *   between one and two ticks, and so on.
                     *
                     * so we need to add (milliseconds-per-tick - 1)
                     * to the amount of time we sleep for.
                     *
                     * test ThreadDelay001 fails if we get this wrong.
                     */
                    Sleep(((work->workData.delayData.usecs + 999) / 1000)
                            + iom->sleepResolution - 1);
                    len = work->workData.delayData.usecs;
                    complData = NULL;
                    fd = 0;
                    errCode = 0;
                } else if ( work->workKind & WORKER_DO_PROC ) {
                    // perform operation/proc on behalf of Haskell thread.
                    if (work->workData.procData.proc) {
                        // The procedure is assumed to encode result +
                        // success/failure via its param.
                        void* param = work->workData.procData.param;
                        errCode=work->workData.procData.proc(param);
                    } else {
                        errCode=1;
                    }
                    complData = work->workData.procData.param;
                } else {
                    fprintf(stderr, "unknown work request type (%d), "
                                    "ignoring.\n", work->workKind);
                    fflush(stderr);
                    continue;
                }
                if (!work->abandonOp) {
                    work->onCompletion(work->requestID,
                                       fd,
                                       len,
                                       complData,
                                       errCode);
                }
                // Free the WorkItem
                DeregisterWorkItem(iom,work);
                stgFree(work);
            } else {
                fprintf(stderr, "unable to fetch work; fatal.\n");
                fflush(stderr);
                OS_ACQUIRE_LOCK(&iom->manLock);
                ioMan->numWorkers--;
                OS_RELEASE_LOCK(&iom->manLock);
                return 1;
            }
        } else {
            fprintf(stderr, "waiting failed (%lu); fatal.\n", rc);
            fflush(stderr);
            OS_ACQUIRE_LOCK(&iom->manLock);
            ioMan->numWorkers--;
            OS_RELEASE_LOCK(&iom->manLock);
            return 1;
        }
    }
    return 0;
}

static
bool
NewIOWorkerThread(IOManagerState* iom)
{
    unsigned threadId;
    return ( 0 != _beginthreadex(NULL,
                                 0,
                                 IOWorkerProc,
                                 (LPVOID)iom,
                                 0,
                                 &threadId) );
}

bool
StartIOManager(void)
{
    HANDLE hExit;
    WorkQueue* wq;
    UINT sleepResolution;
    TIMECAPS timecaps;
    MMRESULT mmresult;

    mmresult = timeGetDevCaps(&timecaps, sizeof(timecaps));
    if (mmresult != MMSYSERR_NOERROR) {
        return false;
    }
    sleepResolution = timecaps.wPeriodMin;
    mmresult = timeBeginPeriod(sleepResolution);
    if (mmresult != MMSYSERR_NOERROR) {
        return false;
    }

    wq = NewWorkQueue();
    if ( !wq ) return false;

    ioMan = (IOManagerState*)stgMallocBytes(sizeof(IOManagerState), "StartIOManager");

    if (!ioMan) {
        FreeWorkQueue(wq);
        return false;
    }

    /* A manual-reset event */
    hExit = CreateEvent ( NULL, true, false, NULL );
    if ( !hExit ) {
        FreeWorkQueue(wq);
        stgFree(ioMan);
        return false;
    }

    ioMan->hExitEvent = hExit;
    OS_INIT_LOCK(&ioMan->manLock);
    ioMan->workQueue   = wq;
    ioMan->numWorkers  = 0;
    ioMan->workersIdle = 0;
    ioMan->queueSize   = 0;
    ioMan->requestID   = 1;
    OS_INIT_LOCK(&ioMan->active_work_lock);
    ioMan->active_work_items = NULL;
    ioMan->sleepResolution = sleepResolution;

    return true;
}

/*
 * Function: depositWorkItem()
 *
 * Local function which deposits a WorkItem onto a work queue,
 * deciding in the process whether or not the thread pool needs
 * to be augmented with another thread to handle the new request.
 *
 */
static
int
depositWorkItem( unsigned int reqID,
                 WorkItem* wItem )
{
    OS_ACQUIRE_LOCK(&ioMan->manLock);

#if 0
    fprintf(stderr, "depositWorkItem: %d/%d\n",
            ioMan->workersIdle, ioMan->numWorkers);
    fflush(stderr);
#endif
    /* A new worker thread is created when there are fewer idle threads
     * than non-consumed queue requests. This ensures that requests will
     * be dealt with in a timely manner.
     *
     * [Long explanation of why the previous thread pool policy lead to
     * trouble]
     *
     * Previously, the thread pool was augmented iff no idle worker threads
     * were available. That strategy runs the risk of repeatedly adding to
     * the request queue without expanding the thread pool to handle this
     * sudden spike in queued requests.
     * [How? Assume workersIdle is 1, and addIORequest() is called. No new
     * thread is created and the request is simply queued. If addIORequest()
     * is called again _before the OS schedules a worker thread to pull the
     * request off the queue_, workersIdle is still 1 and another request is
     * simply added to the queue. Once the worker thread is run, only one
     * request is de-queued, leaving the 2nd request in the queue]
     *
     * Assuming none of the queued requests take an inordinate amount
     * of to complete, the request queue would eventually be
     * drained. But if that's not the case, the later requests will
     * end up languishing in the queue indefinitely. The non-timely
     * handling of requests may cause CH applications to misbehave /
     * hang; bad.
     *
     */
    ioMan->queueSize++;
    if ( (ioMan->workersIdle < ioMan->queueSize) ) {
        /* see if giving up our quantum ferrets out some idle threads.
         */
        OS_RELEASE_LOCK(&ioMan->manLock);
        Sleep(0);
        OS_ACQUIRE_LOCK(&ioMan->manLock);
        if ( (ioMan->workersIdle < ioMan->queueSize) ) {
            /* No, go ahead and create another. */
            ioMan->numWorkers++;
            if (!NewIOWorkerThread(ioMan)) {
                ioMan->numWorkers--;
            }
        }
    }
    OS_RELEASE_LOCK(&ioMan->manLock);

    if (SubmitWork(ioMan->workQueue,wItem)) {
        /* Note: the work item has potentially been consumed by a worker thread
         *       (and freed) at this point, so we cannot use wItem's requestID.
         */
        return reqID;
    } else {
        return 0;
    }
}

/*
 * Function: AddIORequest()
 *
 * Conduit to underlying WorkQueue's SubmitWork(); adds IO
 * request to work queue, deciding whether or not to augment
 * the thread pool in the process.
 */
int
AddIORequest ( int   fd,
               bool  forWriting,
               bool  isSocket,
               HsInt len,
               char* buffer,
               CompletionProc onCompletion)
{
    ASSERT(ioMan);

    WorkItem* wItem    = (WorkItem*)stgMallocBytes(sizeof(WorkItem), "AddIORequest");

    unsigned int reqID = ioMan->requestID++;

    /* Fill in the blanks */
    wItem->workKind     = ( isSocket   ? WORKER_FOR_SOCKET : 0 ) |
                          ( forWriting ? WORKER_WRITE : WORKER_READ );
    wItem->workData.ioData.fd  = fd;
    wItem->workData.ioData.len = len;
    wItem->workData.ioData.buf = buffer;
    wItem->link = NULL;

    wItem->onCompletion        = onCompletion;
    wItem->requestID           = reqID;

    return depositWorkItem(reqID, wItem);
}

/*
 * Function: AddDelayRequest()
 *
 * Like AddIORequest(), but this time adding a delay request to
 * the request queue.
 */
BOOL
AddDelayRequest ( HsInt          usecs,
                  CompletionProc onCompletion)
{
    ASSERT(ioMan);

    WorkItem* wItem = (WorkItem*)stgMallocBytes(sizeof(WorkItem), "AddDelayRequest");

    unsigned int reqID = ioMan->requestID++;

    /* Fill in the blanks */
    wItem->workKind     = WORKER_DELAY;
    wItem->workData.delayData.usecs = usecs;
    wItem->onCompletion = onCompletion;
    wItem->requestID    = reqID;
    wItem->link         = NULL;

    return depositWorkItem(reqID, wItem);
}

/*
 * Function: AddProcRequest()
 *
 * Add an asynchronous procedure request.
 */
BOOL
AddProcRequest ( void* proc,
                 void* param,
                 CompletionProc onCompletion)
{
    ASSERT(ioMan);

    WorkItem* wItem = (WorkItem*)stgMallocBytes(sizeof(WorkItem), "AddProcRequest");
    if (!wItem) return false;

    unsigned int reqID = ioMan->requestID++;

    /* Fill in the blanks */
    wItem->workKind     = WORKER_DO_PROC;
    wItem->workData.procData.proc  = proc;
    wItem->workData.procData.param = param;
    wItem->onCompletion = onCompletion;
    wItem->requestID    = reqID;
    wItem->abandonOp    = 0;
    wItem->link         = NULL;

    return depositWorkItem(reqID, wItem);
}

void ShutdownIOManager ( bool wait_threads )
{
    int num;
    MMRESULT mmresult;

    SetEvent(ioMan->hExitEvent);

    if (wait_threads) {
        /* Wait for all worker threads to die. */
        for (;;) {
            OS_ACQUIRE_LOCK(&ioMan->manLock);
            num = ioMan->numWorkers;
            OS_RELEASE_LOCK(&ioMan->manLock);
            if (num == 0)
                break;
            Sleep(10);
        }
        FreeWorkQueue(ioMan->workQueue);
        CloseHandle(ioMan->hExitEvent);
        OS_CLOSE_LOCK(&ioMan->active_work_lock);
        OS_CLOSE_LOCK(&ioMan->manLock);

        mmresult = timeEndPeriod(ioMan->sleepResolution);
        if (mmresult != MMSYSERR_NOERROR) {
            barf("timeEndPeriod failed");
        }

        stgFree(ioMan);
        ioMan = NULL;
    }
}

/* Keep track of WorkItems currently being serviced. */
static
void
RegisterWorkItem(IOManagerState* ioMan,
                 WorkItem* wi)
{
    OS_ACQUIRE_LOCK(&ioMan->active_work_lock);
    wi->link = ioMan->active_work_items;
    ioMan->active_work_items = wi;
    OS_RELEASE_LOCK(&ioMan->active_work_lock);
}

static
void
DeregisterWorkItem(IOManagerState* ioMan,
                   WorkItem* wi)
{
    WorkItem *ptr, *prev;

    OS_ACQUIRE_LOCK(&ioMan->active_work_lock);
    for(prev=NULL,ptr=ioMan->active_work_items;ptr;prev=ptr,ptr=ptr->link) {
        if (wi->requestID == ptr->requestID) {
            if (prev==NULL) {
                ioMan->active_work_items = ptr->link;
            } else {
                prev->link = ptr->link;
            }
            OS_RELEASE_LOCK(&ioMan->active_work_lock);
            return;
        }
    }
    fprintf(stderr, "DeregisterWorkItem: unable to locate work item %d\n",
            wi->requestID);
    OS_RELEASE_LOCK(&ioMan->active_work_lock);
}


/*
 * Function: abandonWorkRequest()
 *
 * Signal that a work request isn't of interest. Called by the Scheduler
 * if a blocked Haskell thread has an exception thrown to it.
 *
 * Note: we're not aborting the system call that a worker might be blocked on
 * here, just disabling the propagation of its result once its finished. We
 * may have to go the whole hog here and switch to overlapped I/O so that we
 * can abort blocked system calls.
 */
void
abandonWorkRequest ( int reqID )
{
    WorkItem *ptr;
    OS_ACQUIRE_LOCK(&ioMan->active_work_lock);
    for(ptr=ioMan->active_work_items;ptr;ptr=ptr->link) {
        if (ptr->requestID == (unsigned int)reqID ) {
            ptr->abandonOp = 1;
            OS_RELEASE_LOCK(&ioMan->active_work_lock);
            return;
        }
    }
    /* Note: if the request ID isn't present, the worker will have
     * finished sometime since awaitRequests() last drained the completed
     * request table; i.e., not an error.
     */
    OS_RELEASE_LOCK(&ioMan->active_work_lock);
}

#endif