WorkQueue* pq = iom->workQueue;
WorkItem* work;
int len = 0, fd = 0;
- DWORD errCode;
+ DWORD errCode = 0;
void* complData;
hWaits[0] = (HANDLE)iom->hExitEvent;
*/
iom->workersIdle++;
LeaveCriticalSection(&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
+ return 0;
+ }
+
EnterCriticalSection(&iom->manLock);
/* Signal that the thread is 'non-idle' and about to consume
* a work item.
iom->queueSize--;
LeaveCriticalSection(&iom->manLock);
- if ( WAIT_OBJECT_0 == rc ) {
- /* shutdown */
- return 0;
- } else if ( (WAIT_OBJECT_0 + 1) == rc ) {
+ if ( rc == (WAIT_OBJECT_0 + 1) ) {
/* work item available, fetch it. */
if (FetchWork(pq,(void**)&work)) {
if ( work->workKind & WORKER_READ ) {
errCode = WSAGetLastError();
}
} else {
+ DWORD dw;
+
+ 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);
+ dw = GetLastError();
+ 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 non-EOF condition too far back up, but ignore
+ * it.
+ *
+ * However, we do want to give the RTS an opportunity to deliver the
+ * console event. Take care of this in the low-level console handler
+ * in ConsoleHandler.c which wakes up the RTS thread that's blocked
+ * waiting for I/O results from this worker (and possibly others).
+ * It won't see any I/O, but notices and dispatches the queued up
+ * signals/console events while in the Scheduler.
+ *
+ * The original, and way hackier scheme, was to have the worker
+ * return a special return code representing aborted-due-to-ctrl-C-on-stdin,
+ * which GHC.Conc.asyncRead would look out for and retry the I/O
+ * call if encountered.
+ */
+ if ( dw == ERROR_OPERATION_ABORTED ) {
+ /* Only do the retry when dealing with the standard input handle. */
+ HANDLE h = (HANDLE)GetStdHandle(STD_INPUT_HANDLE);
+ if ( _get_osfhandle(work->workData.ioData.fd) == (long)h ) {
+ Sleep(0);
+ } else {
+ break;
+ }
+ } else {
+ break;
+ }
+ } else {
+ break;
+ }
+ }
if (len == -1) { errCode = errno; }
}
complData = work->workData.ioData.buf;
}
/*
- * Function: AddIORequest()
+ * 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.
*
- * Conduit to underlying WorkQueue's SubmitWork(); adds IO
- * request to work queue, deciding whether or not to augment
- * the thread pool in the process.
*/
+static
int
-AddIORequest ( int fd,
- BOOL forWriting,
- BOOL isSocket,
- int len,
- char* buffer,
- CompletionProc onCompletion)
+depositWorkItem( unsigned int reqID,
+ WorkItem* wItem )
{
- WorkItem* wItem = (WorkItem*)malloc(sizeof(WorkItem));
- unsigned int reqID = ioMan->requestID++;
- if (!ioMan || !wItem) return 0;
-
- /* 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->onCompletion = onCompletion;
- wItem->requestID = reqID;
-
EnterCriticalSection(&ioMan->manLock);
- /* If there are no worker threads available, create one.
- *
- * If this turns out to be too aggressive a policy, refine.
- */
+
#if 0
- fprintf(stderr, "AddIORequest: %d\n", ioMan->workersIdle);
+ 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
* 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, returning without blocking.
- request is simply queued. If addIORequest()
+ * 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
*
*/
ioMan->queueSize++;
- if ( ioMan->workersIdle < ioMan->queueSize ) {
- ioMan->numWorkers++;
+ if ( (ioMan->workersIdle < ioMan->queueSize) ) {
+ /* see if giving up our quantum ferrets out some idle threads.
+ */
LeaveCriticalSection(&ioMan->manLock);
- NewIOWorkerThread(ioMan);
+ Sleep(0);
+ EnterCriticalSection(&ioMan->manLock);
+ if ( (ioMan->workersIdle < ioMan->queueSize) ) {
+ /* No, go ahead and create another. */
+ ioMan->numWorkers++;
+ LeaveCriticalSection(&ioMan->manLock);
+ NewIOWorkerThread(ioMan);
+ } else {
+ LeaveCriticalSection(&ioMan->manLock);
+ }
} else {
LeaveCriticalSection(&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.
} 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,
+ int len,
+ char* buffer,
+ CompletionProc onCompletion)
+{
+ WorkItem* wItem = (WorkItem*)malloc(sizeof(WorkItem));
+ unsigned int reqID = ioMan->requestID++;
+ if (!ioMan || !wItem) return 0;
+
+ /* 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->onCompletion = onCompletion;
+ wItem->requestID = reqID;
+
+ return depositWorkItem(reqID, wItem);
}
/*
wItem->onCompletion = onCompletion;
wItem->requestID = reqID;
- EnterCriticalSection(&ioMan->manLock);
-#if 0
- fprintf(stderr, "AddDelayRequest: %d\n", ioMan->workersIdle);
- fflush(stderr);
-#endif
- /* See AddIORequest() for comments regarding policy
- * for augmenting the worker thread pool.
- */
- ioMan->queueSize++;
- if ( ioMan->workersIdle < ioMan->queueSize ) {
- ioMan->numWorkers++;
- LeaveCriticalSection(&ioMan->manLock);
- NewIOWorkerThread(ioMan);
- } else {
- LeaveCriticalSection(&ioMan->manLock);
- }
-
- if (SubmitWork(ioMan->workQueue,wItem)) {
- /* See AddIORequest() comment */
- return reqID;
- } else {
- return 0;
- }
+ return depositWorkItem(reqID, wItem);
}
/*
wItem->onCompletion = onCompletion;
wItem->requestID = reqID;
- EnterCriticalSection(&ioMan->manLock);
-#if 0
- fprintf(stderr, "AddProcRequest: %d\n", ioMan->workersIdle);
- fflush(stderr);
-#endif
- /* See AddIORequest() for comments regarding policy
- * for augmenting the worker thread pool.
- */
- ioMan->queueSize++;
- if ( ioMan->workersIdle < ioMan->queueSize ) {
- ioMan->numWorkers++;
- LeaveCriticalSection(&ioMan->manLock);
- NewIOWorkerThread(ioMan);
- } else {
- LeaveCriticalSection(&ioMan->manLock);
- }
-
- if (SubmitWork(ioMan->workQueue,wItem)) {
- /* See AddIORequest() comment */
- return reqID;
- } else {
- return 0;
- }
+ return depositWorkItem(reqID, wItem);
}
-void ShutdownIOManager()
+void ShutdownIOManager ( void )
{
SetEvent(ioMan->hExitEvent);
- free(ioMan);
- ioMan = NULL;
+ // ToDo: we can't free this now, because the worker thread(s)
+ // haven't necessarily finished with it yet. Perhaps it should
+ // have a reference count or something.
+ // free(ioMan);
+ // ioMan = NULL;
}
projects / ghc-hetmet.git / blobdiff