1bccc288b1c24f0b9d0d47953c029ba4239ceb52
[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 PLIST_ENTRY QueueEntry;
121 NTSTATUS Status;
122 PKTHREAD Thread = KeGetCurrentThread();
123 KIRQL OldIrql;
124 PKQUEUE PreviousQueue;
125 PKWAIT_BLOCK WaitBlock;
126 PKTIMER Timer;
127 DPRINT("KeRemoveQueue %x\n", Queue);
128
129 /* Check if the Lock is already held */
130 if (Thread->WaitNext)
131 {
132 DPRINT("Lock is already held\n");
133 Thread->WaitNext = FALSE;
134 }
135 else
136 {
137 /* Lock the Dispatcher Database */
138 DPRINT("Lock not held, acquiring\n");
139 OldIrql = KeAcquireDispatcherDatabaseLock();
140 Thread->WaitIrql = OldIrql;
141 }
142
143 /* This is needed so that we can set the new queue right here, before additional processing */
144 PreviousQueue = Thread->Queue;
145 Thread->Queue = Queue;
146
147 /* Check if this is a different queue */
148 if (Queue != PreviousQueue)
149 {
150 DPRINT("Different Queue\n");
151 QueueEntry = &Thread->QueueListEntry;
152 if (PreviousQueue)
153 {
154 /* Remove from this list */
155 DPRINT("Removing Old Queue\n");
156 RemoveEntryList(QueueEntry);
157
158 /* Wake the queue */
159 DPRINT("Activating new thread\n");
160 KiWakeQueue(PreviousQueue);
161 }
162
163 /* Insert in this new Queue */
164 DPRINT("Inserting new Queue!\n");
165 InsertTailList(&Queue->ThreadListHead, QueueEntry);
166 }
167 else
168 {
169 /* Same queue, decrement waiting threads */
170 DPRINT("Same Queue!\n");
171 Queue->CurrentCount--;
172 }
173
174 /* Loop until the queue is processed */
175 while (TRUE)
176 {
177 /* Check if the counts are valid and if there is still a queued entry */
178 QueueEntry = Queue->EntryListHead.Flink;
179 if ((Queue->CurrentCount < Queue->MaximumCount) &&
180 (QueueEntry != &Queue->EntryListHead))
181 {
182 /* Remove the Entry and Save it */
183 DPRINT("Removing Queue Entry. CurrentCount: %d, Maximum Count: %d\n",
184 Queue->CurrentCount, Queue->MaximumCount);
185
186 /* Decrease the number of entries */
187 Queue->Header.SignalState--;
188
189 /* Increase numbef of running threads */
190 Queue->CurrentCount++;
191
192 /* Check if the entry is valid. If not, bugcheck */
193 if (!(QueueEntry->Flink) || !(QueueEntry->Blink))
194 {
195 KEBUGCHECK(INVALID_WORK_QUEUE_ITEM);
196 }
197
198 /* Remove the Entry */
199 RemoveEntryList(QueueEntry);
200 QueueEntry->Flink = NULL;
201
202 /* Nothing to wait on */
203 break;
204 }
205 else
206 {
207 /* Do the wait */
208 DPRINT("Waiting on Queue Entry. CurrentCount: %d, Maximum Count: %d\n",
209 Queue->CurrentCount, Queue->MaximumCount);
210
211 /* Use the Thread's Wait Block, it's big enough */
212 Thread->WaitBlockList = &Thread->WaitBlock[0];
213
214 /* Check if a kernel APC is pending and we were below APC_LEVEL */
215 if ((Thread->ApcState.KernelApcPending) &&
216 (Thread->WaitIrql < APC_LEVEL))
217 {
218 /* Increment the count and unlock the dispatcher */
219 Queue->CurrentCount++;
220 KeReleaseDispatcherDatabaseLock(Thread->WaitIrql);
221 goto SkipWait;
222 }
223
224 /* Fail if there's a User APC Pending */
225 if ((WaitMode != KernelMode) && (Thread->ApcState.UserApcPending))
226 {
227 /* Return the status and increase the pending threads */
228 QueueEntry = (PLIST_ENTRY)STATUS_USER_APC;
229 Queue->CurrentCount++;
230
231 /* Nothing to wait on */
232 break;
233 }
234
235 /* Build the Wait Block */
236 WaitBlock = &Thread->WaitBlock[0];
237 WaitBlock->Object = (PVOID)Queue;
238 WaitBlock->WaitKey = STATUS_SUCCESS;
239 WaitBlock->WaitType = WaitAny;
240 WaitBlock->Thread = Thread;
241 Thread->WaitStatus = STATUS_WAIT_0;
242
243 /* We need to wait for the object... check if we have a timeout */
244 if (Timeout)
245 {
246 /* If it's zero, then don't do any waiting */
247 if (!Timeout->QuadPart)
248 {
249 /* Instant Timeout, return the status and increase the pending threads */
250 DPRINT("Queue Wait has timed out\n");
251 QueueEntry = (PLIST_ENTRY)STATUS_TIMEOUT;
252 Queue->CurrentCount++;
253
254 /* Nothing to wait on */
255 break;
256 }
257
258 /*
259 * Set up the Timer. We'll use the internal function so that we can
260 * hold on to the dispatcher lock.
261 */
262 Timer = &Thread->Timer;
263 WaitBlock->NextWaitBlock = &Thread->WaitBlock[1];
264 WaitBlock = &Thread->WaitBlock[1];
265
266 /* Set up the Timer Wait Block */
267 WaitBlock->Object = (PVOID)Timer;
268 WaitBlock->Thread = Thread;
269 WaitBlock->WaitKey = STATUS_TIMEOUT;
270 WaitBlock->WaitType = WaitAny;
271
272 /* Link the timer to this Wait Block */
273 Timer->Header.WaitListHead.Flink = &WaitBlock->WaitListEntry;
274 Timer->Header.WaitListHead.Blink = &WaitBlock->WaitListEntry;
275
276 /* Create Timer */
277 DPRINT("Creating Timer with timeout %I64d\n", *Timeout);
278 KiInsertTimer(Timer, *Timeout);
279 }
280
281 /* Close the loop */
282 WaitBlock->NextWaitBlock = &Thread->WaitBlock[0];
283
284 /* Insert the wait block into the Queues's wait list */
285 WaitBlock = &Thread->WaitBlock[0];
286 InsertTailList(&Queue->Header.WaitListHead,
287 &WaitBlock->WaitListEntry);
288
289 /* Setup the wait information */
290 Thread->WaitMode = WaitMode;
291 Thread->WaitReason = WrQueue;
292 Thread->Alertable = FALSE;
293 Thread->WaitTime = ((PLARGE_INTEGER)&KeTickCount)->LowPart;
294 Thread->State = Waiting;
295
296 /* Find a new thread to run */
297 DPRINT("Swapping threads\n");
298 Status = KiSwapThread();
299
300 /* Reset the wait reason */
301 Thread->WaitReason = 0;
302
303 /* Check if we were executing an APC */
304 if (Status != STATUS_KERNEL_APC)
305 {
306 /* Done Waiting */
307 DPRINT("Done waking queue. Thread: %x %x!\n", KeGetCurrentThread(), Thread);
308 return (PLIST_ENTRY)Status;
309 }
310
311 /* Check if we had a timeout */
312 if (Timeout)
313 {
314 /* FIXME: Fixup interval */
315 }
316
317 /* Acquire again the lock */
318 SkipWait:
319 DPRINT("Looping again\n");
320 OldIrql = KeAcquireDispatcherDatabaseLock();
321
322 /* Save the new IRQL and decrease number of waiting threads */
323 Thread->WaitIrql = OldIrql;
324 Queue->CurrentCount--;
325 }
326 }
327
328 /* Unlock Database and return */
329 KeReleaseDispatcherDatabaseLock(Thread->WaitIrql);
330 DPRINT("Returning. CurrentCount: %d, Maximum Count: %d\n",
331 Queue->CurrentCount, Queue->MaximumCount);
332 return QueueEntry;
333 }
334
335 /*
336 * @implemented
337 */
338 PLIST_ENTRY
339 STDCALL
340 KeRundownQueue(IN PKQUEUE Queue)
341 {
342 PLIST_ENTRY EnumEntry;
343 PLIST_ENTRY FirstEntry = NULL;
344 PKTHREAD Thread;
345 KIRQL OldIrql;
346
347 DPRINT("KeRundownQueue(Queue %x)\n", Queue);
348
349 /* Get the Dispatcher Lock */
350 OldIrql = KeAcquireDispatcherDatabaseLock();
351
352 /* Make sure the list is not empty */
353 if (!IsListEmpty(&Queue->EntryListHead))
354 {
355 /* Remove it */
356 FirstEntry = RemoveHeadList(&Queue->EntryListHead);
357 }
358
359 /* Unlink threads and clear their Thread->Queue */
360 while (!IsListEmpty(&Queue->ThreadListHead))
361 {
362 /* Get the Entry and Remove it */
363 EnumEntry = RemoveHeadList(&Queue->ThreadListHead);
364
365 /* Get the Entry's Thread */
366 Thread = CONTAINING_RECORD(EnumEntry, KTHREAD, QueueListEntry);
367
368 /* Kill its Queue */
369 Thread->Queue = NULL;
370 }
371
372 /* Release the lock and return */
373 KeReleaseDispatcherDatabaseLock(OldIrql);
374 return FirstEntry;
375 }
376
377 /*
378 * Called when a thread which has a queue entry is entering a wait state
379 */
380 VOID
381 FASTCALL
382 KiWakeQueue(IN PKQUEUE Queue)
383 {
384 PLIST_ENTRY QueueEntry;
385 PLIST_ENTRY WaitEntry;
386 PKWAIT_BLOCK WaitBlock;
387 PKTHREAD Thread;
388
389 /* Decrement the number of active threads */
390 DPRINT("KiWakeQueue: %x. Thread: %x\n", Queue, KeGetCurrentThread());
391 Queue->CurrentCount--;
392
393 /* Make sure the counts are OK */
394 if (Queue->CurrentCount < Queue->MaximumCount)
395 {
396 /* Get the Queue Entry */
397 QueueEntry = Queue->EntryListHead.Flink;
398
399 /* Get the Wait Entry */
400 WaitEntry = Queue->Header.WaitListHead.Blink;
401 DPRINT("Queue Count is ok; entries: %p, %p\n", QueueEntry, WaitEntry);
402
403 /* Make sure that the Queue entries are not part of empty lists */
404 if ((WaitEntry != &Queue->Header.WaitListHead) &&
405 (QueueEntry != &Queue->EntryListHead))
406 {
407 /* Remove this entry */
408 DPRINT("Queue in List, removing it\n");
409 RemoveEntryList(QueueEntry);
410 QueueEntry->Flink = NULL;
411
412 /* Decrease the Signal State */
413 Queue->Header.SignalState--;
414
415 /* Unwait the Thread */
416 WaitBlock = CONTAINING_RECORD(WaitEntry,
417 KWAIT_BLOCK,
418 WaitListEntry);
419 Thread = WaitBlock->Thread;
420 DPRINT1("Unwaiting Thread: %d\n", Thread->State);
421 KiAbortWaitThread(Thread, (NTSTATUS)QueueEntry, IO_NO_INCREMENT);
422 }
423 }
424 }
425
426 /*
427 * Returns the previous number of entries in the queue
428 */
429 LONG
430 STDCALL
431 KiInsertQueue(IN PKQUEUE Queue,
432 IN PLIST_ENTRY Entry,
433 BOOLEAN Head)
434 {
435 ULONG InitialState;
436 PKTHREAD Thread = KeGetCurrentThread();
437 PKWAIT_BLOCK WaitBlock;
438 PLIST_ENTRY WaitEntry;
439 DPRINT("KiInsertQueue(Queue %x, Entry %x)\n", Queue, Entry);
440
441 /* Save the old state */
442 InitialState = Queue->Header.SignalState;
443
444 /* Get the Entry */
445 WaitEntry = Queue->Header.WaitListHead.Blink;
446 DPRINT("Initial State, WaitEntry: %d, %x\n", InitialState, WaitEntry);
447
448 /*
449 * Why the KeGetCurrentThread()->Queue != Queue?
450 * KiInsertQueue might be called from an APC for the current thread.
451 * -Gunnar
452 */
453 if ((Queue->CurrentCount < Queue->MaximumCount) &&
454 (WaitEntry != &Queue->Header.WaitListHead) &&
455 ((Thread->Queue != Queue) || (Thread->WaitReason != WrQueue)))
456 {
457 /* Remove the wait entry */
458 DPRINT("Removing Entry\n");
459 RemoveEntryList(WaitEntry);
460
461 /* Get the Wait Block and Thread */
462 WaitBlock = CONTAINING_RECORD(WaitEntry, KWAIT_BLOCK, WaitListEntry);
463 DPRINT("Got wait block: %x\n", WaitBlock);
464 Thread = WaitBlock->Thread;
465
466 /* Reset the wait reason */
467 Thread->WaitReason = 0;
468
469 /* Increase the active threads and set the status*/
470 Queue->CurrentCount++;
471 Thread->WaitStatus = (NTSTATUS)Entry;
472
473 /* Remove the thread from its wait list */
474 RemoveEntryList(&Thread->WaitListEntry);
475
476 /* Check if there's a Thread Timer */
477 if (Thread->Timer.Header.Inserted)
478 {
479 /* Cancel the Thread Timer with the no-lock fastpath */
480 DPRINT("Removing the Thread's Timer\n");
481 Thread->Timer.Header.Inserted = FALSE;
482 RemoveEntryList(&Thread->Timer.TimerListEntry);
483 }
484
485 /* Reschedule the Thread */
486 DPRINT("Unblocking the Thread\n");
487 KiUnblockThread(Thread, (PNTSTATUS)&Entry, 0);
488 }
489 else
490 {
491 /* Increase the Entries */
492 DPRINT("Adding new Queue Entry: %d %d\n", Head, Queue->Header.SignalState);
493 Queue->Header.SignalState++;
494
495 /* Check which mode we're using */
496 if (Head)
497 {
498 /* Insert in the head */
499 InsertHeadList(&Queue->EntryListHead, Entry);
500 }
501 else
502 {
503 /* Insert at the end */
504 InsertTailList(&Queue->EntryListHead, Entry);
505 }
506 }
507
508 /* Return the previous state */
509 DPRINT("Returning\n");
510 return InitialState;
511 }
512
513 /* EOF */