reactos/ntoskrnl/ex/work.c

   1 /*
   2  * COPYRIGHT:          See COPYING in the top level directory
   3  * PROJECT:            ReactOS kernel
   4  * FILE:               mkernel/kernel/work.c
   5  * PURPOSE:            Manage system work queues
   6  * PROGRAMMER:         David Welch (welch@mcmail.com)
   7  * REVISION HISTORY:
   8  *             29/06/98: Created
   9  */
  10
  11 /* INCLUDES ******************************************************************/
  12
  13 #include <ddk/ntddk.h>
  14
  15 #include <internal/ps.h>
  16
  17 #include <internal/debug.h>
  18
  19 /* DEFINES *******************************************************************/
  20
  21 #define NUMBER_OF_WORKER_THREADS   (5)
  22
  23 /* TYPES *********************************************************************/
  24
  25 typedef struct _WORK_QUEUE
  26 {
  27    /*
  28     * PURPOSE: Head of the list of waiting work items
  29     */
  30    LIST_ENTRY Head;
  31
  32    /*
  33     * PURPOSE: Sychronize access to the access
  34     */
  35    KSPIN_LOCK Lock;
  36
  37    /*
  38     * PURPOSE: Worker threads with nothing to do wait on this event
  39     */
  40    KSEMAPHORE Sem;
  41
  42    /*
  43     * PURPOSE: Thread associated with work queue
  44     */
  45    HANDLE Thread[NUMBER_OF_WORKER_THREADS];
  46 } WORK_QUEUE, *PWORK_QUEUE;
  47
  48 /* GLOBALS *******************************************************************/
  49
  50 /*
  51  * PURPOSE: Queue of items waiting to be processed at normal priority
  52  */
  53 WORK_QUEUE EiNormalWorkQueue;
  54
  55 WORK_QUEUE EiCriticalWorkQueue;
  56
  57 WORK_QUEUE EiHyperCriticalWorkQueue;
  58
  59 /* FUNCTIONS ****************************************************************/
  60
  61 static NTSTATUS ExWorkerThreadEntryPoint(PVOID context)
  62 /*
  63  * FUNCTION: Entry point for a worker thread
  64  * ARGUMENTS:
  65  *         context = Parameters
  66  * RETURNS: Status
  67  * NOTE: To kill a worker thread you must queue an item whose callback
  68  * calls PsTerminateSystemThread
  69  */
  70 {
  71    PWORK_QUEUE queue = (PWORK_QUEUE)context;
  72    PWORK_QUEUE_ITEM item;
  73    PLIST_ENTRY current;
  74
  75    for(;;)
  76      {
  77         current = ExInterlockedRemoveHeadList(&queue->Head,
  78                                               &queue->Lock);
  79         if (current!=NULL)
  80           {
  81              item = CONTAINING_RECORD(current,WORK_QUEUE_ITEM,Entry);
  82              item->Routine(item->Context);
  83           }
  84         else
  85           {
  86              KeWaitForSingleObject((PVOID)&queue->Sem,
  87                                    Executive,
  88                                    KernelMode,
  89                                    FALSE,
  90                                    NULL);
  91           }
  92      }
  93 }
  94
  95 static VOID ExInitializeWorkQueue(PWORK_QUEUE WorkQueue,
  96                                   KPRIORITY Priority)
  97 {
  98    ULONG i;
  99    PETHREAD Thread;
 100
 101    InitializeListHead(&WorkQueue->Head);
 102    KeInitializeSpinLock(&WorkQueue->Lock);
 103    KeInitializeSemaphore(&WorkQueue->Sem,
 104                          0,
 105                          256);
 106    for (i=0; i<NUMBER_OF_WORKER_THREADS; i++)
 107      {
 108         PsCreateSystemThread(&WorkQueue->Thread[i],
 109                              THREAD_ALL_ACCESS,
 110                              NULL,
 111                              NULL,
 112                              NULL,
 113                              ExWorkerThreadEntryPoint,
 114                              WorkQueue);
 115         ObReferenceObjectByHandle(WorkQueue->Thread[i],
 116                                   THREAD_ALL_ACCESS,
 117                                   PsThreadType,
 118                                   KernelMode,
 119                                   (PVOID*)&Thread,
 120                                   NULL);
 121         KeSetPriorityThread(&Thread->Tcb,
 122                             Priority);
 123         ObDereferenceObject(Thread);
 124      }
 125 }
 126
 127 VOID ExInitializeWorkerThreads(VOID)
 128 {
 129    ExInitializeWorkQueue(&EiNormalWorkQueue,
 130                          LOW_PRIORITY);
 131    ExInitializeWorkQueue(&EiCriticalWorkQueue,
 132                          LOW_REALTIME_PRIORITY);
 133    ExInitializeWorkQueue(&EiHyperCriticalWorkQueue,
 134                          HIGH_PRIORITY);
 135 }
 136
 137 VOID ExInitializeWorkItem(PWORK_QUEUE_ITEM Item,
 138                           PWORKER_THREAD_ROUTINE Routine,
 139                           PVOID Context)
 140 /*
 141  * FUNCTION: Initializes a work item to be processed by one of the system
 142  * worker threads
 143  * ARGUMENTS:
 144  *         Item = Pointer to the item to be initialized
 145  *         Routine = Routine to be called by the worker thread
 146  *         Context = Parameter to be passed to the callback
 147  */
 148 {
 149    ASSERT_IRQL(DISPATCH_LEVEL);
 150
 151    Item->Routine = Routine;
 152    Item->Context = Context;
 153    Item->Entry.Flink = NULL;
 154    Item->Entry.Blink = NULL;
 155 }
 156
 157 VOID ExQueueWorkItem(PWORK_QUEUE_ITEM WorkItem,
 158                      WORK_QUEUE_TYPE QueueType)
 159 /*
 160  * FUNCTION: Inserts a work item in a queue for one of the system worker
 161  * threads to process
 162  * ARGUMENTS:
 163  *        WorkItem = Item to insert
 164  *        QueueType = Queue to insert it in
 165  */
 166 {
 167    assert(WorkItem!=NULL);
 168    ASSERT_IRQL(DISPATCH_LEVEL);
 169
 170    /*
 171     * Insert the item in the appropiate queue and wake up any thread
 172     * waiting for something to do
 173     */
 174    switch(QueueType)
 175      {
 176       case DelayedWorkQueue:
 177         ExInterlockedInsertTailList(&EiNormalWorkQueue.Head,
 178                                     &WorkItem->Entry,
 179                                     &EiNormalWorkQueue.Lock);
 180         KeReleaseSemaphore(&EiNormalWorkQueue.Sem,
 181                            1,
 182                            IO_NO_INCREMENT,
 183                            FALSE);
 184         break;
 185
 186       case CriticalWorkQueue:
 187         ExInterlockedInsertTailList(&EiCriticalWorkQueue.Head,
 188                                     &WorkItem->Entry,
 189                                     &EiCriticalWorkQueue.Lock);
 190         KeReleaseSemaphore(&EiCriticalWorkQueue.Sem,
 191                            1,
 192                            IO_NO_INCREMENT,
 193                            FALSE);
 194         break;
 195
 196       case HyperCriticalWorkQueue:
 197         ExInterlockedInsertTailList(&EiHyperCriticalWorkQueue.Head,
 198                                     &WorkItem->Entry,
 199                                     &EiHyperCriticalWorkQueue.Lock);
 200         KeReleaseSemaphore(&EiHyperCriticalWorkQueue.Sem,
 201                            1,
 202                            IO_NO_INCREMENT,
 203                            FALSE);
 204         break;
 205
 206      }
 207 }