abfddd90aa44f396e7538f2dc00a9a1faee7448f
[reactos.git] / reactos / ntoskrnl / ke / queue.c
1 /*
2 * COPYRIGHT: See COPYING in the top level directory
3 * PROJECT: ReactOS kernel
4 * FILE: ntoskrnl/ke/queue.c
5 * PURPOSE: Implements kernel queues
6 *
7 * PROGRAMMERS: Alex Ionescu (alex@relsoft.net)
8 * Gunnar Dalsnes
9 * Eric Kohl (ekohl@rz-online.de)
10 */
11
12 /* INCLUDES *****************************************************************/
13
14 #include <ntoskrnl.h>
15 #define NDEBUG
16 #include <internal/debug.h>
17
18 /* FUNCTIONS *****************************************************************/
19
20 LONG STDCALL KiInsertQueue(IN PKQUEUE Queue, IN PLIST_ENTRY Entry, BOOLEAN Head);
21
22 /*
23 * @implemented
24 */
25 VOID
26 STDCALL
27 KeInitializeQueue(IN PKQUEUE Queue,
28 IN ULONG Count OPTIONAL)
29 {
30 DPRINT("KeInitializeQueue %x\n", Queue);
31
32 /* Initialize the Header */
33 KeInitializeDispatcherHeader(&Queue->Header,
34 QueueObject,
35 sizeof(KQUEUE)/sizeof(ULONG),
36 0);
37
38 /* Initialize the Lists */
39 InitializeListHead(&Queue->EntryListHead);
40 InitializeListHead(&Queue->ThreadListHead);
41
42 /* Set the Current and Maximum Count */
43 Queue->CurrentCount = 0;
44 Queue->MaximumCount = (Count == 0) ? (ULONG) KeNumberProcessors : Count;
45 }
46
47 /*
48 * @implemented
49 */
50 LONG
51 STDCALL
52 KeInsertHeadQueue(IN PKQUEUE Queue,
53 IN PLIST_ENTRY Entry)
54 {
55 LONG PreviousState;
56 KIRQL OldIrql;
57
58 DPRINT("KeInsertHeadQueue %x\n", Queue);
59
60 /* Lock the Dispatcher Database */
61 OldIrql = KeAcquireDispatcherDatabaseLock();
62
63 /* Insert the Queue */
64 PreviousState = KiInsertQueue(Queue, Entry, TRUE);
65
66 /* Release the Dispatcher Lock */
67 KeReleaseDispatcherDatabaseLock(OldIrql);
68
69 /* Return previous State */
70 return PreviousState;
71 }
72
73 /*
74 * @implemented
75 */
76 LONG STDCALL
77 KeInsertQueue(IN PKQUEUE Queue,
78 IN PLIST_ENTRY Entry)
79 {
80 LONG PreviousState;
81 KIRQL OldIrql;
82
83 DPRINT("KeInsertQueue %x\n", Queue);
84
85 /* Lock the Dispatcher Database */
86 OldIrql = KeAcquireDispatcherDatabaseLock();
87
88 /* Insert the Queue */
89 PreviousState = KiInsertQueue(Queue, Entry, FALSE);
90
91 /* Release the Dispatcher Lock */
92 KeReleaseDispatcherDatabaseLock(OldIrql);
93
94 /* Return previous State */
95 return PreviousState;
96 }
97
98 /*
99 * @implemented
100 *
101 * Returns number of entries in the queue
102 */
103 LONG
104 STDCALL
105 KeReadStateQueue(IN PKQUEUE Queue)
106 {
107 /* Returns the Signal State */
108 return(Queue->Header.SignalState);
109 }
110
111 /*
112 * @implemented
113 */
114 PLIST_ENTRY
115 STDCALL
116 KeRemoveQueue(IN PKQUEUE Queue,
117 IN KPROCESSOR_MODE WaitMode,
118 IN PLARGE_INTEGER Timeout OPTIONAL)
119 {
120
121 PLIST_ENTRY ListEntry;
122 NTSTATUS Status;
123 PKTHREAD Thread = KeGetCurrentThread();
124 KIRQL OldIrql;
125 PKQUEUE PreviousQueue;
126 PKWAIT_BLOCK WaitBlock;
127 PKWAIT_BLOCK TimerWaitBlock;
128 PKTIMER Timer;
129
130 DPRINT("KeRemoveQueue %x\n", Queue);
131
132 /* Check if the Lock is already held */
133 if (Thread->WaitNext) {
134
135 DPRINT("Lock is already held\n");
136
137 } else {
138
139 /* Lock the Dispatcher Database */
140 DPRINT("Lock not held, acquiring\n");
141 OldIrql = KeAcquireDispatcherDatabaseLock();
142 Thread->WaitIrql = OldIrql;
143 }
144
145 /* This is needed so that we can set the new queue right here, before additional processing */
146 PreviousQueue = Thread->Queue;
147 Thread->Queue = Queue;
148
149 /* Check if this is a different queue */
150 if (Queue != PreviousQueue) {
151
152 /*
153 * INVESTIGATE: What is the Thread->QueueListEntry used for? It's linked it into the
154 * Queue->ThreadListHead when the thread registers with the queue and unlinked when
155 * the thread registers with a new queue. The Thread->Queue already tells us what
156 * queue the thread is registered with.
157 * -Gunnar
158 */
159 DPRINT("Different Queue\n");
160 if (PreviousQueue) {
161
162 /* Remove from this list */
163 DPRINT("Removing Old Queue\n");
164 RemoveEntryList(&Thread->QueueListEntry);
165
166 /* Wake the queue */
167 DPRINT("Activating new thread\n");
168 KiWakeQueue(PreviousQueue);
169 }
170
171 /* Insert in this new Queue */
172 DPRINT("Inserting new Queue!\n");
173 InsertTailList(&Queue->ThreadListHead, &Thread->QueueListEntry);
174
175 } else {
176
177 /* Same queue, decrement waiting threads */
178 DPRINT("Same Queue!\n");
179 Queue->CurrentCount--;
180 }
181
182 /* Loop until the queue is processed */
183 while (TRUE) {
184
185 /* Get the Entry */
186 ListEntry = Queue->EntryListHead.Flink;
187
188 /* Check if the counts are valid and if there is still a queued entry */
189 if ((Queue->CurrentCount < Queue->MaximumCount) &&
190 (ListEntry != &Queue->EntryListHead)) {
191
192 /* Remove the Entry and Save it */
193 DPRINT("Removing Queue Entry. CurrentCount: %d, Maximum Count: %d\n",
194 Queue->CurrentCount, Queue->MaximumCount);
195 ListEntry = RemoveHeadList(&Queue->EntryListHead);
196
197 /* Decrease the number of entries */
198 Queue->Header.SignalState--;
199
200 /* Increase numbef of running threads */
201 Queue->CurrentCount++;
202
203 /* Check if the entry is valid. If not, bugcheck */
204 if (!ListEntry->Flink || !ListEntry->Blink) {
205
206 KEBUGCHECK(INVALID_WORK_QUEUE_ITEM);
207 }
208
209 /* Remove the Entry */
210 RemoveEntryList(ListEntry);
211 ListEntry->Flink = NULL;
212
213 /* Nothing to wait on */
214 break;
215
216 } else {
217
218 /* Do the wait */
219 DPRINT("Waiting on Queue Entry. CurrentCount: %d, Maximum Count: %d\n",
220 Queue->CurrentCount, Queue->MaximumCount);
221
222 /* Use the Thread's Wait Block, it's big enough */
223 Thread->WaitBlockList = &Thread->WaitBlock[0];
224
225 /* Fail if there's an APC Pending */
226 if (WaitMode == UserMode && Thread->ApcState.UserApcPending) {
227
228 /* Return the status and increase the pending threads */
229 ListEntry = (PLIST_ENTRY)STATUS_USER_APC;
230 Queue->CurrentCount++;
231
232 /* Nothing to wait on */
233 break;
234 }
235
236 /* Build the Wait Block */
237 WaitBlock = &Thread->WaitBlock[0];
238 WaitBlock->Object = (PVOID)Queue;
239 WaitBlock->WaitKey = STATUS_SUCCESS;
240 WaitBlock->WaitType = WaitAny;
241 WaitBlock->Thread = Thread;
242 WaitBlock->NextWaitBlock = NULL;
243
244 Thread->WaitStatus = STATUS_SUCCESS;
245
246 /* We need to wait for the object... check if we have a timeout */
247 if (Timeout) {
248
249 /* If it's zero, then don't do any waiting */
250 if (!Timeout->QuadPart) {
251
252 /* Instant Timeout, return the status and increase the pending threads */
253 DPRINT("Queue Wait has timed out\n");
254 ListEntry = (PLIST_ENTRY)STATUS_TIMEOUT;
255 Queue->CurrentCount++;
256
257 /* Nothing to wait on */
258 break;
259 }
260
261 /*
262 * Set up the Timer. We'll use the internal function so that we can
263 * hold on to the dispatcher lock.
264 */
265 Timer = &Thread->Timer;
266 TimerWaitBlock = &Thread->WaitBlock[1];
267
268 /* Set up the Timer Wait Block */
269 TimerWaitBlock->Object = (PVOID)Timer;
270 TimerWaitBlock->Thread = Thread;
271 TimerWaitBlock->WaitKey = STATUS_TIMEOUT;
272 TimerWaitBlock->WaitType = WaitAny;
273 TimerWaitBlock->NextWaitBlock = NULL;
274
275 /* Link the timer to this Wait Block */
276 InitializeListHead(&Timer->Header.WaitListHead);
277 InsertTailList(&Timer->Header.WaitListHead, &TimerWaitBlock->WaitListEntry);
278
279 /* Create Timer */
280 DPRINT("Creating Timer with timeout %I64d\n", *Timeout);
281 KiInsertTimer(Timer, *Timeout);
282 }
283
284 /* Insert the wait block into the Queues's wait list */
285 WaitBlock = Thread->WaitBlockList;
286 InsertTailList(&Queue->Header.WaitListHead, &WaitBlock->WaitListEntry);
287
288 /* Block the Thread */
289 DPRINT("Blocking the Thread: %x %x!\n", KeGetCurrentThread(), Thread);
290 PsBlockThread(&Status,
291 FALSE,
292 WaitMode,
293 WrQueue);
294
295 /* Reset the wait reason */
296 Thread->WaitReason = 0;
297
298 /* Check if we were executing an APC */
299 if (Status != STATUS_KERNEL_APC) {
300
301 /* Done Waiting */
302 DPRINT("Done waking queue. Thread: %x %x!\n", KeGetCurrentThread(), Thread);
303 return (PLIST_ENTRY)Status;
304 }
305
306 /* Acquire again the lock */
307 DPRINT("Looping again\n");
308 OldIrql = KeAcquireDispatcherDatabaseLock();
309
310 /* Save the new IRQL and decrease number of waiting threads */
311 Thread->WaitIrql = OldIrql;
312 Queue->CurrentCount--;
313 }
314 }
315
316 /* Unlock Database and return */
317 KeReleaseDispatcherDatabaseLock(Thread->WaitIrql);
318 DPRINT("Returning. CurrentCount: %d, Maximum Count: %d\n",
319 Queue->CurrentCount, Queue->MaximumCount);
320 return ListEntry;
321 }
322
323 /*
324 * @implemented
325 */
326 PLIST_ENTRY
327 STDCALL
328 KeRundownQueue(IN PKQUEUE Queue)
329 {
330 PLIST_ENTRY EnumEntry;
331 PLIST_ENTRY FirstEntry;
332 PKTHREAD Thread;
333 KIRQL OldIrql;
334
335 DPRINT("KeRundownQueue(Queue %x)\n", Queue);
336
337 /* Get the Dispatcher Lock */
338 OldIrql = KeAcquireDispatcherDatabaseLock();
339
340 /* Get the First Empty Entry */
341 FirstEntry = Queue->EntryListHead.Flink;
342
343 /* Make sure the list is not empty */
344 if (FirstEntry == &Queue->EntryListHead) {
345
346 /* It is, so don't return anything */
347 EnumEntry = NULL;
348
349 } else {
350
351 /* Remove it */
352 RemoveEntryList(&Queue->EntryListHead);
353 }
354
355 /* Unlink threads and clear their Thread->Queue */
356 while (!IsListEmpty(&Queue->ThreadListHead)) {
357
358 /* Get the Entry and Remove it */
359 EnumEntry = RemoveHeadList(&Queue->ThreadListHead);
360
361 /* Get the Entry's Thread */
362 Thread = CONTAINING_RECORD(EnumEntry, KTHREAD, QueueListEntry);
363
364 /* Kill its Queue */
365 Thread->Queue = NULL;
366 }
367
368 /* Release the lock and return */
369 KeReleaseDispatcherDatabaseLock(OldIrql);
370 return FirstEntry;
371 }
372
373 /*
374 * Called when a thread which has a queue entry is entering a wait state
375 */
376 VOID
377 FASTCALL
378 KiWakeQueue(IN PKQUEUE Queue)
379 {
380 PLIST_ENTRY QueueEntry;
381 PLIST_ENTRY WaitEntry;
382 PKWAIT_BLOCK WaitBlock;
383
384 /* Decrement the number of active threads */
385 DPRINT("KiWakeQueue: %x. Thread: %x\n", Queue, KeGetCurrentThread());
386 Queue->CurrentCount--;
387
388 /* Make sure the counts are OK */
389 if (Queue->CurrentCount < Queue->MaximumCount) {
390
391 /* Get the Queue Entry */
392 QueueEntry = Queue->EntryListHead.Flink;
393
394 /* Get the Wait Entry */
395 WaitEntry = Queue->Header.WaitListHead.Blink;
396 DPRINT("Queue Count is ok, Queue entries: %x, %x\n", QueueEntry, WaitEntry);
397
398 /* Make sure that the Queue List isn't empty and that this entry is valid */
399 if (!IsListEmpty(&Queue->Header.WaitListHead) &&
400 (QueueEntry != &Queue->EntryListHead)) {
401
402 /* Remove this entry */
403 DPRINT("Queue in List, removing it\n");
404 RemoveEntryList(QueueEntry);
405 QueueEntry->Flink = NULL;
406
407 /* Decrease the Signal State */
408 Queue->Header.SignalState--;
409
410 /* Unwait the Thread */
411 DPRINT("Unwaiting Thread\n");
412 WaitBlock = CONTAINING_RECORD(WaitEntry, KWAIT_BLOCK, WaitListEntry);
413 KiAbortWaitThread(WaitBlock->Thread, (NTSTATUS)QueueEntry, IO_NO_INCREMENT);
414 }
415 }
416 }
417
418 /*
419 * Returns the previous number of entries in the queue
420 */
421 LONG
422 STDCALL
423 KiInsertQueue(IN PKQUEUE Queue,
424 IN PLIST_ENTRY Entry,
425 BOOLEAN Head)
426 {
427 ULONG InitialState;
428 PKTHREAD Thread = KeGetCurrentThread();
429 PKWAIT_BLOCK WaitBlock;
430 PLIST_ENTRY WaitEntry;
431
432 DPRINT("KiInsertQueue(Queue %x, Entry %x)\n", Queue, Entry);
433
434 /* Save the old state */
435 InitialState = Queue->Header.SignalState;
436
437 /* Get the Entry */
438 WaitEntry = Queue->Header.WaitListHead.Blink;
439 DPRINT("Initial State, WaitEntry: %d, %x\n", InitialState, WaitEntry);
440
441 /*
442 * Why the KeGetCurrentThread()->Queue != Queue?
443 * KiInsertQueue might be called from an APC for the current thread.
444 * -Gunnar
445 */
446 if ((Queue->CurrentCount < Queue->MaximumCount) &&
447 (WaitEntry != &Queue->Header.WaitListHead) &&
448 ((Thread->Queue != Queue) || (Thread->WaitReason != WrQueue))) {
449
450 /* Remove the wait entry */
451 DPRINT("Removing Entry\n");
452 RemoveEntryList(WaitEntry);
453
454 /* Get the Wait Block and Thread */
455 WaitBlock = CONTAINING_RECORD(WaitEntry, KWAIT_BLOCK, WaitListEntry);
456 DPRINT("Got wait block: %x\n", WaitBlock);
457 Thread = WaitBlock->Thread;
458
459 /* Reset the wait reason */
460 Thread->WaitReason = 0;
461
462 /* Increase the waiting threads */
463 Queue->CurrentCount++;
464
465 /* Check if there's a Thread Timer */
466 if (Thread->Timer.Header.Inserted) {
467
468 /* Cancel the Thread Timer with the no-lock fastpath */
469 DPRINT("Removing the Thread's Timer\n");
470 Thread->Timer.Header.Inserted = FALSE;
471 RemoveEntryList(&Thread->Timer.TimerListEntry);
472 }
473
474 /* Reschedule the Thread */
475 DPRINT("Unblocking the Thread\n");
476 PsUnblockThread((PETHREAD)Thread, (PNTSTATUS)&Entry, 0);
477
478 } else {
479
480 /* Increase the Entries */
481 DPRINT("Adding new Queue Entry: %d %d\n", Head, Queue->Header.SignalState);
482 Queue->Header.SignalState++;
483
484 if (Head) {
485
486 InsertHeadList(&Queue->EntryListHead, Entry);
487
488 } else {
489
490 InsertTailList(&Queue->EntryListHead, Entry);
491 }
492 }
493
494 /* Return the previous state */
495 DPRINT("Returning\n");
496 return InitialState;
497 }
498
499 /* EOF */