AsyncIO.c 8.79 KB
Newer Older
sof's avatar
sof committed
1
2
3
4
5
6
7
/* AsyncIO.c
 *
 * Integrating Win32 asynchronous I/O with the GHC RTS.
 *
 * (c) sof, 2002-2003.
 */
#include "Rts.h"
sof's avatar
sof committed
8
#include "RtsUtils.h"
sof's avatar
sof committed
9
10
11
#include <windows.h>
#include <stdio.h>
#include "Schedule.h"
12
#include "Capability.h"
sof's avatar
sof committed
13
14
15
16
17
18
19
#include "win32/AsyncIO.h"
#include "win32/IOManager.h"

/*
 * Overview:
 *
 * Haskell code issue asynchronous I/O requests via the 
sof's avatar
sof committed
20
 * async{Read,Write,DoOp}# primops. These cause addIORequest()
sof's avatar
sof committed
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
 * to be invoked, which forwards the request to the underlying
 * asynchronous I/O subsystem. Each request is tagged with a unique
 * ID.
 *
 * addIORequest() returns this ID, so that when the blocked CH
 * thread is added onto blocked_queue, its TSO is annotated with
 * it. Upon completion of an I/O request, the async I/O handling
 * code makes a back-call to signal its completion; the local
 * onIOComplete() routine. It adds the IO request ID (along with
 * its result data) to a queue of completed requests before returning. 
 *
 * The queue of completed IO request is read by the thread operating
 * the RTS scheduler. It de-queues the CH threads corresponding
 * to the request IDs, making them runnable again.
 *
 */

typedef struct CompletedReq {
sof's avatar
sof committed
39
40
41
    unsigned int   reqID;
    int            len;
    int            errCode;
sof's avatar
sof committed
42
43
44
45
46
47
} CompletedReq;

#define MAX_REQUESTS 200

static CRITICAL_SECTION queue_lock;
static HANDLE           completed_req_event;
sof's avatar
sof committed
48
49
static HANDLE           abandon_req_wait;
static HANDLE           wait_handles[2];
sof's avatar
sof committed
50
51
52
53
54
55
56
57
static CompletedReq     completedTable[MAX_REQUESTS];
static int              completed_hw;
static int              issued_reqs;

static void
onIOComplete(unsigned int reqID,
	     int   fd STG_UNUSED,
	     int   len,
sof's avatar
sof committed
58
	     void* buf STG_UNUSED,
sof's avatar
sof committed
59
60
	     int   errCode)
{
sof's avatar
sof committed
61
62
63
64
65
66
67
    /* Deposit result of request in queue/table */
    EnterCriticalSection(&queue_lock);
    if (completed_hw == MAX_REQUESTS) {
	/* Not likely */
	fprintf(stderr, "Request table overflow (%d); dropping.\n", reqID);
	fflush(stderr);
    } else {
sof's avatar
sof committed
68
#if 0
sof's avatar
sof committed
69
70
71
	fprintf(stderr, "onCompl: %d %d %d %d %d\n", 
		reqID, len, errCode, issued_reqs, completed_hw); 
	fflush(stderr);
sof's avatar
sof committed
72
#endif
sof's avatar
sof committed
73
74
75
76
77
78
79
	completedTable[completed_hw].reqID   = reqID;
	completedTable[completed_hw].len     = len;
	completedTable[completed_hw].errCode = errCode;
	completed_hw++;
	issued_reqs--;
	if (completed_hw == 1) {
	    /* The event is used to wake up the scheduler thread should it
sof's avatar
sof committed
80
	     * be blocked waiting for requests to complete. The event resets once
sof's avatar
sof committed
81
82
83
84
	     * that thread has cleared out the request queue/table.
	     */
	    SetEvent(completed_req_event);
	}
sof's avatar
sof committed
85
    }
sof's avatar
sof committed
86
    LeaveCriticalSection(&queue_lock);
sof's avatar
sof committed
87
88
89
90
91
92
93
94
95
}

unsigned int
addIORequest(int   fd,
	     int   forWriting,
	     int   isSock,
	     int   len,
	     char* buf)
{
sof's avatar
sof committed
96
97
98
    EnterCriticalSection(&queue_lock);
    issued_reqs++;
    LeaveCriticalSection(&queue_lock);
sof's avatar
sof committed
99
#if 0
sof's avatar
sof committed
100
    fprintf(stderr, "addIOReq: %d %d %d\n", fd, forWriting, len); fflush(stderr);
sof's avatar
sof committed
101
#endif
sof's avatar
sof committed
102
    return AddIORequest(fd,forWriting,isSock,len,buf,onIOComplete);
sof's avatar
sof committed
103
104
}

sof's avatar
sof committed
105
unsigned int
sof's avatar
sof committed
106
addDelayRequest(int msecs)
sof's avatar
sof committed
107
{
sof's avatar
sof committed
108
109
110
    EnterCriticalSection(&queue_lock);
    issued_reqs++;
    LeaveCriticalSection(&queue_lock);
sof's avatar
sof committed
111
#if 0
sof's avatar
sof committed
112
    fprintf(stderr, "addDelayReq: %d\n", msecs); fflush(stderr);
sof's avatar
sof committed
113
#endif
sof's avatar
sof committed
114
    return AddDelayRequest(msecs,onIOComplete);
sof's avatar
sof committed
115
116
}

sof's avatar
sof committed
117
118
119
unsigned int
addDoProcRequest(void* proc, void* param)
{
sof's avatar
sof committed
120
121
122
    EnterCriticalSection(&queue_lock);
    issued_reqs++;
    LeaveCriticalSection(&queue_lock);
sof's avatar
sof committed
123
#if 0
sof's avatar
sof committed
124
    fprintf(stderr, "addProcReq: %p %p\n", proc, param); fflush(stderr);
sof's avatar
sof committed
125
#endif
sof's avatar
sof committed
126
    return AddProcRequest(proc,param,onIOComplete);
sof's avatar
sof committed
127
128
129
}


sof's avatar
sof committed
130
131
132
int
startupAsyncIO()
{
sof's avatar
sof committed
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
    if (!StartIOManager()) {
	return 0;
    }
    InitializeCriticalSection(&queue_lock);
    /* Create a pair of events:
     *
     *    - completed_req_event  -- signals the deposit of request result; manual reset.
     *    - abandon_req_wait     -- external OS thread tells current RTS/Scheduler
     *                              thread to abandon wait for IO request completion.
     *                              Auto reset.
     */
    completed_req_event = CreateEvent (NULL, TRUE,  FALSE, NULL);
    abandon_req_wait    = CreateEvent (NULL, FALSE, FALSE, NULL);
    wait_handles[0] = completed_req_event;
    wait_handles[1] = abandon_req_wait;
    completed_hw = 0;
    return ( completed_req_event != INVALID_HANDLE_VALUE &&
	     abandon_req_wait    != INVALID_HANDLE_VALUE );
sof's avatar
sof committed
151
152
153
154
155
}

void
shutdownAsyncIO()
{
sof's avatar
sof committed
156
157
    CloseHandle(completed_req_event);
    ShutdownIOManager();
sof's avatar
sof committed
158
159
}

sof's avatar
sof committed
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
/*
 * Function: awaitRequests(wait)
 *
 * Check for the completion of external IO work requests. Worker
 * threads signal completion of IO requests by depositing them
 * in a table (completedTable). awaitRequests() matches up 
 * requests in that table with threads on the blocked_queue, 
 * making the threads whose IO requests have completed runnable
 * again.
 * 
 * awaitRequests() is called by the scheduler periodically _or_ if
 * it is out of work, and need to wait for the completion of IO
 * requests to make further progress. In the latter scenario, 
 * awaitRequests() will simply block waiting for worker threads 
 * to complete if the 'completedTable' is empty.
 */
sof's avatar
sof committed
176
177
178
179
180
int
awaitRequests(rtsBool wait)
{
start:
#if 0
sof's avatar
sof committed
181
182
    fprintf(stderr, "awaitRequests(): %d %d %d\n", issued_reqs, completed_hw, wait);
    fflush(stderr);
sof's avatar
sof committed
183
#endif
sof's avatar
sof committed
184
185
    EnterCriticalSection(&queue_lock);
    /* Nothing immediately available & we won't wait */
186
187
188
189
190
191
192
    if ((!wait && completed_hw == 0)
#if 0
	// If we just return when wait==rtsFalse, we'll go into a busy
	// wait loop, so I disabled this condition --SDM 18/12/2003
	(issued_reqs == 0 && completed_hw == 0)
#endif
	) {
sof's avatar
sof committed
193
	LeaveCriticalSection(&queue_lock);
sof's avatar
sof committed
194
	return 0;
sof's avatar
sof committed
195
    }
sof's avatar
sof committed
196
197
198
199
200
201
202
203
204
205
206
    if (completed_hw == 0) {
	/* empty table, drop lock and wait */
	LeaveCriticalSection(&queue_lock);
	if ( wait && !interrupted ) {
	    DWORD dwRes = WaitForMultipleObjects(2, wait_handles, FALSE, INFINITE);
	    switch (dwRes) {
	    case WAIT_OBJECT_0:
		break;
	    case WAIT_OBJECT_0 + 1:
	    case WAIT_TIMEOUT:
		return 0;
sof's avatar
sof committed
207
208
209
210
211
	    case WAIT_FAILED: {
		DWORD dw = GetLastError();
		fprintf(stderr, "awaitRequests: wait failed -- error code: %lu\n", dw); fflush(stderr);
		return 0;
	    }
sof's avatar
sof committed
212
213
214
215
216
	    default:
		fprintf(stderr, "awaitRequests: unexpected wait return code %lu\n", dwRes); fflush(stderr);
		return 0;
	    }
	} else {
217
	    return 0;
sof's avatar
sof committed
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
	}
	goto start;
    } else {
	int i;
	StgTSO *tso, *prev;
	
	for (i=0; i < completed_hw; i++) {
	    /* For each of the completed requests, match up their Ids
	     * with those of the threads on the blocked_queue. If the
	     * thread that made the IO request has been subsequently
	     * killed (and removed from blocked_queue), no match will
	     * be found for that request Id. 
	     *
	     * i.e., killing a Haskell thread doesn't attempt to cancel
	     * the IO request it is blocked on.
	     *
sof's avatar
sof committed
234
	     */
sof's avatar
sof committed
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
	    unsigned int rID = completedTable[i].reqID;
	    
	    prev = NULL;
	    for(tso = blocked_queue_hd ; tso != END_TSO_QUEUE; prev = tso, tso = tso->link) {
	
		switch(tso->why_blocked) {
		case BlockedOnRead:
		case BlockedOnWrite:
		case BlockedOnDoProc:
		    if (tso->block_info.async_result->reqID == rID) {
			/* Found the thread blocked waiting on request; stodgily fill 
			 * in its result block. 
			 */
			tso->block_info.async_result->len = completedTable[i].len;
			tso->block_info.async_result->errCode = completedTable[i].errCode;
			
			/* Drop the matched TSO from blocked_queue */
			if (prev) {
			    prev->link = tso->link;
			} else {
			    blocked_queue_hd = tso->link;
			}
			if (blocked_queue_tl == tso) {
			    blocked_queue_tl = prev;
			}
		    
			/* Terminates the run queue + this inner for-loop. */
			tso->link = END_TSO_QUEUE;
			tso->why_blocked = NotBlocked;
			PUSH_ON_RUN_QUEUE(tso);
			break;
		    }
		    break;
		default:
		    if (tso->why_blocked != NotBlocked) {
			barf("awaitRequests: odd thread state");
		    }
		    break;
sof's avatar
sof committed
273
		}
sof's avatar
sof committed
274
275
	    }
	}
sof's avatar
sof committed
276
277
278
279
	completed_hw = 0;
	ResetEvent(completed_req_event);
	LeaveCriticalSection(&queue_lock);
	return 1;
sof's avatar
sof committed
280
281
    }
}
sof's avatar
sof committed
282

sof's avatar
sof committed
283
284
285
286
287
288
/*
 * Function: abandonRequestWait()
 *
 * Wake up a thread that's blocked waiting for new IO requests
 * to complete (via awaitRequests().)
 */
sof's avatar
sof committed
289
void
290
abandonRequestWait( void )
sof's avatar
sof committed
291
{
sof's avatar
sof committed
292
293
294
    /* the event is auto-reset, but in case there's no thread
     * already waiting on the event, we want to return it to
     * a non-signalled state.
295
296
297
298
299
300
301
302
303
     *
     * Careful!  There is no synchronisation between
     * abandonRequestWait and awaitRequest, which means that
     * abandonRequestWait might be called just before a thread
     * goes into a wait, and we miss the abandon signal.  So we
     * must SetEvent() here rather than PulseEvent() to ensure
     * that the event isn't lost.  We can re-optimise by resetting
     * the event somewhere safe if we know the event has been
     * properly serviced (see resetAbandon() below).  --SDM 18/12/2003
sof's avatar
sof committed
304
     */
305
306
307
308
309
310
311
    SetEvent(abandon_req_wait);
}

void
resetAbandonRequestWait( void )
{
    ResetEvent(abandon_req_wait);
sof's avatar
sof committed
312
}
313