Something went wrong. Try again.
Reactos
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488/* * PROJECT: ReactOS Kernel * LICENSE: GPL - See COPYING in the top level directory * FILE: ntoskrnl/ke/queue.c * PURPOSE: Implements kernel queues * PROGRAMMERS: Alex Ionescu (alex.ionescu@reactos.org) * Gunnar Dalsnes * Eric Kohl */
/* INCLUDES ******************************************************************/
#include <ntoskrnl.h>#define NDEBUG#include <debug.h>
/* PRIVATE FUNCTIONS *********************************************************/
/* * Called when a thread which has a queue entry is entering a wait state */VOIDFASTCALLKiActivateWaiterQueue(IN PKQUEUE Queue){ PLIST_ENTRY QueueEntry; PLIST_ENTRY WaitEntry; PKWAIT_BLOCK WaitBlock; PKTHREAD Thread; ASSERT_QUEUE(Queue);
/* Decrement the number of active threads */ Queue->CurrentCount--;
/* Make sure the counts are OK */ if (Queue->CurrentCount < Queue->MaximumCount) { /* Get the Queue Entry */ QueueEntry = Queue->EntryListHead.Flink;
/* Get the Wait Entry */ WaitEntry = Queue->Header.WaitListHead.Blink;
/* Make sure that the Queue entries are not part of empty lists */ if ((WaitEntry != &Queue->Header.WaitListHead) && (QueueEntry != &Queue->EntryListHead)) { /* Remove this entry */ RemoveEntryList(QueueEntry); QueueEntry->Flink = NULL;
/* Decrease the Signal State */ Queue->Header.SignalState--;
/* Unwait the Thread */ WaitBlock = CONTAINING_RECORD(WaitEntry, KWAIT_BLOCK, WaitListEntry); Thread = WaitBlock->Thread; KiUnwaitThread(Thread, (LONG_PTR)QueueEntry, IO_NO_INCREMENT); } }}
/* * Returns the previous number of entries in the queue */LONGNTAPIKiInsertQueue(IN PKQUEUE Queue, IN PLIST_ENTRY Entry, IN BOOLEAN Head){ ULONG InitialState; PKTHREAD Thread = KeGetCurrentThread(); PKWAIT_BLOCK WaitBlock; PLIST_ENTRY WaitEntry; PKTIMER Timer; ASSERT_QUEUE(Queue);
/* Save the old state */ InitialState = Queue->Header.SignalState;
/* Get the Entry */ WaitEntry = Queue->Header.WaitListHead.Blink;
/* * Why the KeGetCurrentThread()->Queue != Queue? * KiInsertQueue might be called from an APC for the current thread. * -Gunnar */ if ((Queue->CurrentCount < Queue->MaximumCount) && (WaitEntry != &Queue->Header.WaitListHead) && ((Thread->Queue != Queue) || (Thread->WaitReason != WrQueue))) { /* Remove the wait entry */ RemoveEntryList(WaitEntry);
/* Get the Wait Block and Thread */ WaitBlock = CONTAINING_RECORD(WaitEntry, KWAIT_BLOCK, WaitListEntry); Thread = WaitBlock->Thread;
/* Remove the queue from the thread's wait list */ Thread->WaitStatus = (LONG_PTR)Entry; if (Thread->WaitListEntry.Flink) RemoveEntryList(&Thread->WaitListEntry);
/* Increase the active threads and remove any wait reason */ Queue->CurrentCount++; Thread->WaitReason = 0;
/* Check if there's a Thread Timer */ Timer = &Thread->Timer; if (Timer->Header.Inserted) KxRemoveTreeTimer(Timer);
/* Reschedule the Thread */ KiReadyThread(Thread); } else { /* Increase the Entries */ Queue->Header.SignalState++;
/* Check which mode we're using */ if (Head) { /* Insert in the head */ InsertHeadList(&Queue->EntryListHead, Entry); } else { /* Insert at the end */ InsertTailList(&Queue->EntryListHead, Entry); } }
/* Return the previous state */ return InitialState;}
/* PUBLIC FUNCTIONS **********************************************************/
/* * @implemented */VOIDNTAPIKeInitializeQueue(IN PKQUEUE Queue, IN ULONG Count OPTIONAL){ /* Initialize the Header */ Queue->Header.Type = QueueObject; Queue->Header.Abandoned = 0; Queue->Header.Size = sizeof(KQUEUE) / sizeof(ULONG); Queue->Header.SignalState = 0; InitializeListHead(&(Queue->Header.WaitListHead));
/* Initialize the Lists */ InitializeListHead(&Queue->EntryListHead); InitializeListHead(&Queue->ThreadListHead);
/* Set the Current and Maximum Count */ Queue->CurrentCount = 0; Queue->MaximumCount = (Count == 0) ? (ULONG) KeNumberProcessors : Count;}
/* * @implemented */LONGNTAPIKeInsertHeadQueue(IN PKQUEUE Queue, IN PLIST_ENTRY Entry){ LONG PreviousState; KIRQL OldIrql; ASSERT_QUEUE(Queue); ASSERT_IRQL_LESS_OR_EQUAL(DISPATCH_LEVEL);
/* Lock the Dispatcher Database */ OldIrql = KiAcquireDispatcherLock();
/* Insert the Queue */ PreviousState = KiInsertQueue(Queue, Entry, TRUE);
/* Release the Dispatcher Lock */ KiReleaseDispatcherLock(OldIrql);
/* Return previous State */ return PreviousState;}
/* * @implemented */LONGNTAPIKeInsertQueue(IN PKQUEUE Queue, IN PLIST_ENTRY Entry){ LONG PreviousState; KIRQL OldIrql; ASSERT_QUEUE(Queue); ASSERT_IRQL_LESS_OR_EQUAL(DISPATCH_LEVEL);
/* Lock the Dispatcher Database */ OldIrql = KiAcquireDispatcherLock();
/* Insert the Queue */ PreviousState = KiInsertQueue(Queue, Entry, FALSE);
/* Release the Dispatcher Lock */ KiReleaseDispatcherLock(OldIrql);
/* Return previous State */ return PreviousState;}
/* * @implemented * * Returns number of entries in the queue */LONGNTAPIKeReadStateQueue(IN PKQUEUE Queue){ /* Returns the Signal State */ ASSERT_QUEUE(Queue); return Queue->Header.SignalState;}
/* * @implemented */PLIST_ENTRYNTAPIKeRemoveQueue(IN PKQUEUE Queue, IN KPROCESSOR_MODE WaitMode, IN PLARGE_INTEGER Timeout OPTIONAL){ PLIST_ENTRY QueueEntry; LONG_PTR Status; PKTHREAD Thread = KeGetCurrentThread(); PKQUEUE PreviousQueue; PKWAIT_BLOCK WaitBlock = &Thread->WaitBlock[0]; PKWAIT_BLOCK TimerBlock = &Thread->WaitBlock[TIMER_WAIT_BLOCK]; PKTIMER Timer = &Thread->Timer; BOOLEAN Swappable; PLARGE_INTEGER OriginalDueTime = Timeout; LARGE_INTEGER DueTime = {{0}}, NewDueTime, InterruptTime; ULONG Hand = 0; ASSERT_QUEUE(Queue); ASSERT_IRQL_LESS_OR_EQUAL(DISPATCH_LEVEL);
/* Check if the Lock is already held */ if (Thread->WaitNext) { /* It is, so next time don't do expect this */ Thread->WaitNext = FALSE; KxQueueThreadWait(); } else { /* Raise IRQL to synch, prepare the wait, then lock the database */ Thread->WaitIrql = KeRaiseIrqlToSynchLevel(); KxQueueThreadWait(); KiAcquireDispatcherLockAtSynchLevel(); }
/* * This is needed so that we can set the new queue right here, * before additional processing */ PreviousQueue = Thread->Queue; Thread->Queue = Queue;
/* Check if this is a different queue */ if (Queue != PreviousQueue) { /* Get the current entry */ QueueEntry = &Thread->QueueListEntry; if (PreviousQueue) { /* Remove from this list */ RemoveEntryList(QueueEntry);
/* Wake the queue */ KiActivateWaiterQueue(PreviousQueue); }
/* Insert in this new Queue */ InsertTailList(&Queue->ThreadListHead, QueueEntry); } else { /* Same queue, decrement waiting threads */ Queue->CurrentCount--; }
/* Loop until the queue is processed */ while (TRUE) { /* Check if the counts are valid and if there is still a queued entry */ QueueEntry = Queue->EntryListHead.Flink; if ((Queue->CurrentCount < Queue->MaximumCount) && (QueueEntry != &Queue->EntryListHead)) { /* Decrease the number of entries */ Queue->Header.SignalState--;
/* Increase numbef of running threads */ Queue->CurrentCount++;
/* Check if the entry is valid. If not, bugcheck */ if (!(QueueEntry->Flink) || !(QueueEntry->Blink)) { /* Invalid item */ KeBugCheckEx(INVALID_WORK_QUEUE_ITEM, (ULONG_PTR)QueueEntry, (ULONG_PTR)Queue, (ULONG_PTR)NULL, (ULONG_PTR)((PWORK_QUEUE_ITEM)QueueEntry)-> WorkerRoutine); }
/* Remove the Entry */ RemoveEntryList(QueueEntry); QueueEntry->Flink = NULL;
/* Nothing to wait on */ break; } else { /* Check if a kernel APC is pending and we're below APC_LEVEL */ if ((Thread->ApcState.KernelApcPending) && !(Thread->SpecialApcDisable) && (Thread->WaitIrql < APC_LEVEL)) { /* Increment the count and unlock the dispatcher */ Queue->CurrentCount++; KiReleaseDispatcherLockFromSynchLevel(); KiExitDispatcher(Thread->WaitIrql); } else { /* Fail if there's a User APC Pending */ if ((WaitMode != KernelMode) && (Thread->ApcState.UserApcPending)) { /* Return the status and increase the pending threads */ QueueEntry = (PLIST_ENTRY)STATUS_USER_APC; Queue->CurrentCount++; break; }
/* Enable the Timeout Timer if there was any specified */ if (Timeout) { /* Check if the timer expired */ InterruptTime.QuadPart = KeQueryInterruptTime(); if ((ULONG64)InterruptTime.QuadPart >= Timer->DueTime.QuadPart) { /* It did, so we don't need to wait */ QueueEntry = (PLIST_ENTRY)STATUS_TIMEOUT; Queue->CurrentCount++; break; }
/* It didn't, so activate it */ Timer->Header.Inserted = TRUE; }
/* Insert the wait block in the list */ InsertTailList(&Queue->Header.WaitListHead, &WaitBlock->WaitListEntry);
/* Setup the wait information */ Thread->State = Waiting;
/* Add the thread to the wait list */ KiAddThreadToWaitList(Thread, Swappable);
/* Activate thread swap */ ASSERT(Thread->WaitIrql <= DISPATCH_LEVEL); KiSetThreadSwapBusy(Thread);
/* Check if we have a timer */ if (Timeout) { /* Insert it */ KxInsertTimer(Timer, Hand); } else { /* Otherwise, unlock the dispatcher */ KiReleaseDispatcherLockFromSynchLevel(); }
/* Do the actual swap */ Status = KiSwapThread(Thread, KeGetCurrentPrcb());
/* Reset the wait reason */ Thread->WaitReason = 0;
/* Check if we were executing an APC */ if (Status != STATUS_KERNEL_APC) return (PLIST_ENTRY)Status;
/* Check if we had a timeout */ if (Timeout) { /* Recalculate due times */ Timeout = KiRecalculateDueTime(OriginalDueTime, &DueTime, &NewDueTime); } }
/* Start another wait */ Thread->WaitIrql = KeRaiseIrqlToSynchLevel(); KxQueueThreadWait(); KiAcquireDispatcherLockAtSynchLevel(); Queue->CurrentCount--; } }
/* Unlock Database and return */ KiReleaseDispatcherLockFromSynchLevel(); KiExitDispatcher(Thread->WaitIrql); return QueueEntry;}
/* * @implemented */PLIST_ENTRYNTAPIKeRundownQueue(IN PKQUEUE Queue){ PLIST_ENTRY FirstEntry, NextEntry; PKTHREAD Thread; KIRQL OldIrql; ASSERT_QUEUE(Queue); ASSERT_IRQL_LESS_OR_EQUAL(DISPATCH_LEVEL); ASSERT(IsListEmpty(&Queue->Header.WaitListHead));
/* Get the Dispatcher Lock */ OldIrql = KiAcquireDispatcherLock();
/* Check if the list is empty */ FirstEntry = Queue->EntryListHead.Flink; if (FirstEntry == &Queue->EntryListHead) { /* We won't return anything */ FirstEntry = NULL; } else { /* Remove this entry */ RemoveEntryList(&Queue->EntryListHead); }
/* Loop the list */ while (!IsListEmpty(&Queue->ThreadListHead)) { /* Get the next entry */ NextEntry = Queue->ThreadListHead.Flink;
/* Get the associated thread */ Thread = CONTAINING_RECORD(NextEntry, KTHREAD, QueueListEntry);
/* Clear its queue */ Thread->Queue = NULL;
/* Remove this entry */ RemoveEntryList(NextEntry); }
/* Release the dispatcher lock */ KiReleaseDispatcherLockFromSynchLevel();
/* Exit the dispatcher and return the first entry (if any) */ KiExitDispatcher(OldIrql); return FirstEntry;}
/* EOF */