New icon. Couple of code changes linked to that. Just playing :)
[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 /* Check if the counts are valid and if there is still a queued entry */
186 if ((Queue->CurrentCount < Queue->MaximumCount) &&
187 !IsListEmpty(&Queue->EntryListHead)) {
188
189 /* Remove the Entry and Save it */
190 DPRINT("Removing Queue Entry. CurrentCount: %d, Maximum Count: %d\n",
191 Queue->CurrentCount, Queue->MaximumCount);
192 ListEntry = Queue->EntryListHead.Flink;
193
194 /* Decrease the number of entries */
195 Queue->Header.SignalState--;
196
197 /* Increase numbef of running threads */
198 Queue->CurrentCount++;
199
200 /* Check if the entry is valid. If not, bugcheck */
201 if (!ListEntry->Flink || !ListEntry->Blink) {
202
203 KEBUGCHECK(INVALID_WORK_QUEUE_ITEM);
204 }
205
206 /* Remove the Entry */
207 RemoveEntryList(ListEntry);
208 ListEntry->Flink = NULL;
209
210 /* Nothing to wait on */
211 break;
212
213 } else {
214
215 /* Do the wait */
216 DPRINT("Waiting on Queue Entry. CurrentCount: %d, Maximum Count: %d\n",
217 Queue->CurrentCount, Queue->MaximumCount);
218
219 /* Use the Thread's Wait Block, it's big enough */
220 Thread->WaitBlockList = &Thread->WaitBlock[0];
221
222 /* Fail if there's an APC Pending */
223 if (WaitMode != KernelMode && Thread->ApcState.UserApcPending) {
224
225 /* Return the status and increase the pending threads */
226 ListEntry = (PLIST_ENTRY)STATUS_USER_APC;
227 Queue->CurrentCount++;
228
229 /* Nothing to wait on */
230 break;
231 }
232
233 /* Build the Wait Block */
234 WaitBlock = &Thread->WaitBlock[0];
235 WaitBlock->Object = (PVOID)Queue;
236 WaitBlock->WaitKey = STATUS_SUCCESS;
237 WaitBlock->WaitType = WaitAny;
238 WaitBlock->Thread = Thread;
239 WaitBlock->NextWaitBlock = WaitBlock;
240
241 Thread->WaitStatus = STATUS_SUCCESS;
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 ListEntry = (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 TimerWaitBlock = &Thread->WaitBlock[1];
264
265 /* Set up the Timer Wait Block */
266 TimerWaitBlock->Object = (PVOID)Timer;
267 TimerWaitBlock->Thread = Thread;
268 TimerWaitBlock->WaitKey = STATUS_TIMEOUT;
269 TimerWaitBlock->WaitType = WaitAny;
270 TimerWaitBlock->NextWaitBlock = TimerWaitBlock;
271
272 /* Link the timer to this Wait Block */
273 InitializeListHead(&Timer->Header.WaitListHead);
274 InsertTailList(&Timer->Header.WaitListHead, &TimerWaitBlock->WaitListEntry);
275
276 /* Create Timer */
277 DPRINT("Creating Timer with timeout %I64d\n", *Timeout);
278 KiInsertTimer(Timer, *Timeout);
279 }
280
281 /* Insert the wait block into the Queues's wait list */
282 WaitBlock = Thread->WaitBlockList;
283 InsertTailList(&Queue->Header.WaitListHead, &WaitBlock->WaitListEntry);
284
285 /* Block the Thread */
286 DPRINT("Blocking the Thread: %x %x!\n", KeGetCurrentThread(), Thread);
287 KiBlockThread(&Status,
288 FALSE,
289 WaitMode,
290 WrQueue);
291
292 /* Reset the wait reason */
293 Thread->WaitReason = 0;
294
295 /* Check if we were executing an APC */
296 if (Status != STATUS_KERNEL_APC) {
297
298 /* Done Waiting */
299 DPRINT("Done waking queue. Thread: %x %x!\n", KeGetCurrentThread(), Thread);
300 return (PLIST_ENTRY)Status;
301 }
302
303 /* Acquire again the lock */
304 DPRINT("Looping again\n");
305 OldIrql = KeAcquireDispatcherDatabaseLock();
306
307 /* Save the new IRQL and decrease number of waiting threads */
308 Thread->WaitIrql = OldIrql;
309 Queue->CurrentCount--;
310 }
311 }
312
313 /* Unlock Database and return */
314 KeReleaseDispatcherDatabaseLock(Thread->WaitIrql);
315 DPRINT("Returning. CurrentCount: %d, Maximum Count: %d\n",
316 Queue->CurrentCount, Queue->MaximumCount);
317 return ListEntry;
318 }
319
320 /*
321 * @implemented
322 */
323 PLIST_ENTRY
324 STDCALL
325 KeRundownQueue(IN PKQUEUE Queue)
326 {
327 PLIST_ENTRY EnumEntry;
328 PLIST_ENTRY FirstEntry = NULL;
329 PKTHREAD Thread;
330 KIRQL OldIrql;
331
332 DPRINT("KeRundownQueue(Queue %x)\n", Queue);
333
334 /* Get the Dispatcher Lock */
335 OldIrql = KeAcquireDispatcherDatabaseLock();
336
337 /* Make sure the list is not empty */
338 if (!IsListEmpty(&Queue->EntryListHead))
339 {
340 /* Remove it */
341 FirstEntry = RemoveHeadList(&Queue->EntryListHead);
342 }
343
344 /* Unlink threads and clear their Thread->Queue */
345 while (!IsListEmpty(&Queue->ThreadListHead)) {
346
347 /* Get the Entry and Remove it */
348 EnumEntry = RemoveHeadList(&Queue->ThreadListHead);
349
350 /* Get the Entry's Thread */
351 Thread = CONTAINING_RECORD(EnumEntry, KTHREAD, QueueListEntry);
352
353 /* Kill its Queue */
354 Thread->Queue = NULL;
355 }
356
357 /* Release the lock and return */
358 KeReleaseDispatcherDatabaseLock(OldIrql);
359 return FirstEntry;
360 }
361
362 /*
363 * Called when a thread which has a queue entry is entering a wait state
364 */
365 VOID
366 FASTCALL
367 KiWakeQueue(IN PKQUEUE Queue)
368 {
369 PLIST_ENTRY QueueEntry;
370 PLIST_ENTRY WaitEntry;
371 PKWAIT_BLOCK WaitBlock;
372
373 /* Decrement the number of active threads */
374 DPRINT("KiWakeQueue: %x. Thread: %x\n", Queue, KeGetCurrentThread());
375 Queue->CurrentCount--;
376
377 /* Make sure the counts are OK */
378 if (Queue->CurrentCount < Queue->MaximumCount) {
379
380 /* Get the Queue Entry */
381 QueueEntry = Queue->EntryListHead.Flink;
382
383 /* Get the Wait Entry */
384 WaitEntry = Queue->Header.WaitListHead.Blink;
385 DPRINT("Queue Count is ok, Queue entries: %x, %x\n", QueueEntry, WaitEntry);
386
387 /* Make sure that the Queue List isn't empty and that this entry is valid */
388 if (!IsListEmpty(&Queue->Header.WaitListHead) &&
389 (QueueEntry != &Queue->EntryListHead)) {
390
391 /* Remove this entry */
392 DPRINT("Queue in List, removing it\n");
393 RemoveEntryList(QueueEntry);
394 QueueEntry->Flink = NULL;
395
396 /* Decrease the Signal State */
397 Queue->Header.SignalState--;
398
399 /* Unwait the Thread */
400 DPRINT("Unwaiting Thread\n");
401 WaitBlock = CONTAINING_RECORD(WaitEntry, KWAIT_BLOCK, WaitListEntry);
402 KiAbortWaitThread(WaitBlock->Thread, (NTSTATUS)QueueEntry, IO_NO_INCREMENT);
403 }
404 }
405 }
406
407 /*
408 * Returns the previous number of entries in the queue
409 */
410 LONG
411 STDCALL
412 KiInsertQueue(IN PKQUEUE Queue,
413 IN PLIST_ENTRY Entry,
414 BOOLEAN Head)
415 {
416 ULONG InitialState;
417 PKTHREAD Thread = KeGetCurrentThread();
418 PKWAIT_BLOCK WaitBlock;
419 PLIST_ENTRY WaitEntry;
420
421 DPRINT("KiInsertQueue(Queue %x, Entry %x)\n", Queue, Entry);
422
423 /* Save the old state */
424 InitialState = Queue->Header.SignalState;
425
426 /* Get the Entry */
427 WaitEntry = Queue->Header.WaitListHead.Blink;
428 DPRINT("Initial State, WaitEntry: %d, %x\n", InitialState, WaitEntry);
429
430 /*
431 * Why the KeGetCurrentThread()->Queue != Queue?
432 * KiInsertQueue might be called from an APC for the current thread.
433 * -Gunnar
434 */
435 if ((Queue->CurrentCount < Queue->MaximumCount) &&
436 (WaitEntry != &Queue->Header.WaitListHead) &&
437 ((Thread->Queue != Queue) || (Thread->WaitReason != WrQueue))) {
438
439 /* Remove the wait entry */
440 DPRINT("Removing Entry\n");
441 RemoveEntryList(WaitEntry);
442
443 /* Get the Wait Block and Thread */
444 WaitBlock = CONTAINING_RECORD(WaitEntry, KWAIT_BLOCK, WaitListEntry);
445 DPRINT("Got wait block: %x\n", WaitBlock);
446 Thread = WaitBlock->Thread;
447
448 /* Reset the wait reason */
449 Thread->WaitReason = 0;
450
451 /* Increase the waiting threads */
452 Queue->CurrentCount++;
453
454 /* Check if there's a Thread Timer */
455 if (Thread->Timer.Header.Inserted) {
456
457 /* Cancel the Thread Timer with the no-lock fastpath */
458 DPRINT("Removing the Thread's Timer\n");
459 Thread->Timer.Header.Inserted = FALSE;
460 RemoveEntryList(&Thread->Timer.TimerListEntry);
461 }
462
463 /* Reschedule the Thread */
464 DPRINT("Unblocking the Thread\n");
465 KiUnblockThread(Thread, (PNTSTATUS)&Entry, 0);
466
467 } else {
468
469 /* Increase the Entries */
470 DPRINT("Adding new Queue Entry: %d %d\n", Head, Queue->Header.SignalState);
471 Queue->Header.SignalState++;
472
473 if (Head) {
474
475 InsertHeadList(&Queue->EntryListHead, Entry);
476
477 } else {
478
479 InsertTailList(&Queue->EntryListHead, Entry);
480 }
481 }
482
483 /* Return the previous state */
484 DPRINT("Returning\n");
485 return InitialState;
486 }
487
488 /* EOF */