/*++
Copyright (c) 1989 Microsoft Corporation
Module Name:
worker.c
Abstract:
This module implements the LAN Manager server FSP worker thread
function. It also implements routines for managing (i.e., starting
and stopping) worker threads, and balancing load.
Author:
Chuck Lenzmeier (chuckl) 01-Oct-1989
David Treadwell (davidtr)
Environment:
Kernel mode
Revision History:
--*/
#include "precomp.h"
#include "worker.tmh"
#pragma hdrstop
#define BugCheckFileId SRV_FILE_WORKER
//
// Local declarations
//
NTSTATUS
CreateQueueThread (
IN PWORK_QUEUE Queue
);
VOID
InitializeWorkerThread (
IN PWORK_QUEUE WorkQueue,
IN KPRIORITY ThreadPriority
);
VOID
WorkerThread (
IN PWORK_QUEUE WorkQueue
);
#ifdef ALLOC_PRAGMA
#pragma alloc_text( PAGE, SrvCreateWorkerThreads )
#pragma alloc_text( PAGE, CreateQueueThread )
#pragma alloc_text( PAGE, InitializeWorkerThread )
#pragma alloc_text( PAGE, WorkerThread )
#endif
#if 0
NOT PAGEABLE -- SrvQueueWorkToBlockingThread
NOT PAGEABLE -- SrvQueueWorkToFsp
NOT PAGEABLE -- SrvQueueWorkToFspAtSendCompletion
NOT PAGEABLE -- SrvBalanceLoad
#endif
NTSTATUS
SrvCreateWorkerThreads (
VOID
)
/*++
Routine Description:
This function creates the worker threads for the LAN Manager server
FSP.
Arguments:
None.
Return Value:
NTSTATUS - Status of thread creation
--*/
{
NTSTATUS status;
PWORK_QUEUE queue;
PAGED_CODE( );
//
// Create the nonblocking worker threads.
//
for( queue = SrvWorkQueues; queue < eSrvWorkQueues; queue++ ) {
status = CreateQueueThread( queue );
if( !NT_SUCCESS( status ) ) {
return status;
}
}
//
// Create the blocking worker threads
//
return CreateQueueThread( &SrvBlockingWorkQueue );
} // SrvCreateWorkerThreads
NTSTATUS
CreateQueueThread (
IN PWORK_QUEUE Queue
)
/*++
Routine Description:
This function creates a worker thread to service a queue.
NOTE: The scavenger occasionally kills off threads on a queue. If logic
here is modified, you may need to look there too.
Arguments:
Queue - the queue to service
Return Value:
NTSTATUS - Status of thread creation
--*/
{
HANDLE threadHandle;
LARGE_INTEGER interval;
NTSTATUS status;
PAGED_CODE();
//
// Another thread is coming into being. Keep the counts up to date
//
InterlockedIncrement( &Queue->Threads );
InterlockedIncrement( &Queue->AvailableThreads );
status = PsCreateSystemThread(
&threadHandle,
PROCESS_ALL_ACCESS,
NULL,
NtCurrentProcess(),
NULL,
WorkerThread,
Queue
);
if ( !NT_SUCCESS(status) ) {
INTERNAL_ERROR(
ERROR_LEVEL_EXPECTED,
"CreateQueueThread: PsCreateSystemThread for "
"queue %X returned %X",
Queue,
status
);
InterlockedDecrement( &Queue->Threads );
InterlockedDecrement( &Queue->AvailableThreads );
SrvLogServiceFailure( SRV_SVC_PS_CREATE_SYSTEM_THREAD, status );
return status;
}
//
// Close the handle so the thread can die when needed
//
SrvNtClose( threadHandle, FALSE );
//
// If we just created the first queue thread, wait for it
// to store its thread pointer in IrpThread. This pointer is
// stored in all IRPs issued for this queue by the server.
//
while ( Queue->IrpThread == NULL ) {
interval.QuadPart = -1*10*1000*10; // .01 second
KeDelayExecutionThread( KernelMode, FALSE, &interval );
}
return STATUS_SUCCESS;
} // CreateQueueThread
VOID
InitializeWorkerThread (
IN PWORK_QUEUE WorkQueue,
IN KPRIORITY ThreadPriority
)
{
NTSTATUS status;
KPRIORITY basePriority;
PAGED_CODE( );
#if SRVDBG_LOCK
{
//
// Create a special system thread TEB. The size of this TEB is just
// large enough to accommodate the first three user-reserved
// longwords. These three locations are used for lock debugging. If
// the allocation fails, then no lock debugging will be performed
// for this thread.
//
//
PETHREAD Thread = PsGetCurrentThread( );
ULONG TebSize = FIELD_OFFSET( TEB, UserReserved[0] ) + SRV_TEB_USER_SIZE;
Thread->Tcb.Teb = ExAllocatePoolWithTag( NonPagedPool, TebSize, BlockTypeMisc );
if ( Thread->Tcb.Teb != NULL ) {
RtlZeroMemory( Thread->Tcb.Teb, TebSize );
}
}
#endif // SRVDBG_LOCK
//
// Set this thread's priority.
//
basePriority = ThreadPriority;
status = NtSetInformationThread (
NtCurrentThread( ),
ThreadBasePriority,
&basePriority,
sizeof(basePriority)
);
if ( !NT_SUCCESS(status) ) {
INTERNAL_ERROR(
ERROR_LEVEL_UNEXPECTED,
"InitializeWorkerThread: NtSetInformationThread failed: %X\n",
status,
NULL
);
SrvLogServiceFailure( SRV_SVC_NT_SET_INFO_THREAD, status );
}
#if MULTIPROCESSOR
//
// If this is a nonblocking worker thread, set its ideal processor affinity. Setting
// ideal affinity informs ntos that the thread would rather run on its ideal
// processor if reasonable, but if ntos can't schedule it on that processor then it is
// ok to schedule it on a different processor.
//
if( SrvNumberOfProcessors > 1 && WorkQueue >= SrvWorkQueues && WorkQueue < eSrvWorkQueues ) {
KeSetIdealProcessorThread( KeGetCurrentThread(), (CCHAR)(WorkQueue - SrvWorkQueues) );
}
#endif
//
// Disable hard error popups for this thread.
//
IoSetThreadHardErrorMode( FALSE );
return;
} // InitializeWorkerThread
VOID
WorkerThread (
IN PWORK_QUEUE WorkQueue
)
{
PLIST_ENTRY listEntry;
PWORK_CONTEXT workContext;
ULONG timeDifference;
ULONG updateSmbCount = 0;
ULONG updateTime = 0;
ULONG iAmBlockingThread = (WorkQueue == &SrvBlockingWorkQueue);
PLARGE_INTEGER Timeout = NULL;
PAGED_CODE();
//
// If this is the first worker thread, save the thread pointer.
//
if( WorkQueue->IrpThread == NULL ) {
WorkQueue->IrpThread = PsGetCurrentThread( );
}
InitializeWorkerThread( WorkQueue, SrvThreadPriority );
//
// If we are the IrpThread, we don't want to die
//
if( WorkQueue->IrpThread != PsGetCurrentThread( ) ) {
Timeout = &WorkQueue->IdleTimeOut;
}
//
// Loop infinitely dequeueing and processing work items.
//
while ( TRUE ) {
listEntry = KeRemoveQueue(
&WorkQueue->Queue,
WorkQueue->WaitMode,
Timeout
);
if( (ULONG_PTR)listEntry == STATUS_TIMEOUT ) {
//
// We have a non critical thread that hasn't gotten any work for
// awhile. Time to die.
//
InterlockedDecrement( &WorkQueue->AvailableThreads );
InterlockedDecrement( &WorkQueue->Threads );
SrvTerminateWorkerThread( NULL );
}
if( InterlockedDecrement( &WorkQueue->AvailableThreads ) == 0 &&
!SrvFspTransitioning &&
WorkQueue->Threads < WorkQueue->MaxThreads ) {
//
// We are running low on threads for this queue. Spin up
// another one before handling this request
//
CreateQueueThread( WorkQueue );
}
//
// Get the address of the work item.
//
workContext = CONTAINING_RECORD(
listEntry,
WORK_CONTEXT,
ListEntry
);
ASSERT( KeGetCurrentIrql() == 0 );
//
// There is work available. It may be a work contect block or
// an RFCB. (Blocking threads won't get RFCBs.)
//
ASSERT( (GET_BLOCK_TYPE(workContext) == BlockTypeWorkContextInitial) ||
(GET_BLOCK_TYPE(workContext) == BlockTypeWorkContextNormal) ||
(GET_BLOCK_TYPE(workContext) == BlockTypeWorkContextRaw) ||
(GET_BLOCK_TYPE(workContext) == BlockTypeWorkContextSpecial) ||
(GET_BLOCK_TYPE(workContext) == BlockTypeRfcb) );
#if DBG
if ( GET_BLOCK_TYPE( workContext ) == BlockTypeRfcb ) {
((PRFCB)workContext)->ListEntry.Flink =
((PRFCB)workContext)->ListEntry.Blink = NULL;
}
#endif
IF_DEBUG(WORKER1) {
KdPrint(( "WorkerThread working on work context %p", workContext ));
}
//
// Make sure we have a resaonable idea of the system time
//
if( ++updateTime == TIME_SMB_INTERVAL ) {
updateTime = 0;
SET_SERVER_TIME( WorkQueue );
}
//
// Update statistics.
//
if ( ++updateSmbCount == STATISTICS_SMB_INTERVAL ) {
updateSmbCount = 0;
GET_SERVER_TIME( WorkQueue, &timeDifference );
timeDifference = timeDifference - workContext->Timestamp;
++(WorkQueue->stats.WorkItemsQueued.Count);
WorkQueue->stats.WorkItemsQueued.Time.QuadPart += timeDifference;
}
{
//
// Put the workContext out relative to bp so we can find it later if we need
// to debug. The block of memory we're writing to is likely already in cache,
// so this should be relatively cheap.
//
PWORK_CONTEXT volatile savedWorkContext;
savedWorkContext = workContext;
}
//
// Make sure the WorkContext knows if it is on the blocking work queue
//
workContext->UsingBlockingThread = iAmBlockingThread;
//
// Call the restart routine for the work item.
//
IF_SMB_DEBUG( TRACE ) {
KdPrint(( "Blocking %d, Count %d -> %p( %p )\n",
iAmBlockingThread,
workContext->ProcessingCount,
workContext->FspRestartRoutine,
workContext
));
}
workContext->FspRestartRoutine( workContext );
//
// Make sure we are still at normal level.
//
ASSERT( KeGetCurrentIrql() == 0 );
//
// We're getting ready to be available (i.e. waiting on the queue)
//
InterlockedIncrement( &WorkQueue->AvailableThreads );
}
} // WorkerThread
VOID SRVFASTCALL
SrvQueueWorkToBlockingThread (
IN OUT PWORK_CONTEXT WorkContext
)
/*++
Routine Description:
This routine queues a work item to a blocking thread. These threads
are used to service requests that may block for a long time, so we
don't want to tie up our normal worker threads.
Arguments:
WorkContext - Supplies a pointer to the work context block
representing the work item
Return Value:
None.
--*/
{
//
// Increment the processing count.
//
WorkContext->ProcessingCount++;
//
// Insert the work item at the tail of the blocking work queue.
//
SrvInsertWorkQueueTail(
&SrvBlockingWorkQueue,
(PQUEUEABLE_BLOCK_HEADER)WorkContext
);
return;
} // SrvQueueWorkToBlockingThread
VOID SRVFASTCALL
SrvQueueWorkToFsp (
IN OUT PWORK_CONTEXT WorkContext
)
/*++
Routine Description:
This is the restart routine for work items that are to be queued to
a nonblocking worker thread in the FSP. This function is also
called from elsewhere in the server to transfer work to the FSP.
This function should not be called at dispatch level -- use
SrvQueueWorkToFspAtDpcLevel instead.
Arguments:
WorkContext - Supplies a pointer to the work context block
representing the work item
Return Value:
None.
--*/
{
//
// Increment the processing count.
//
WorkContext->ProcessingCount++;
//
// Insert the work item at the tail of the nonblocking work queue.
//
if( WorkContext->QueueToHead ) {
SrvInsertWorkQueueHead(
WorkContext->CurrentWorkQueue,
(PQUEUEABLE_BLOCK_HEADER)WorkContext
);
} else {
SrvInsertWorkQueueTail(
WorkContext->CurrentWorkQueue,
(PQUEUEABLE_BLOCK_HEADER)WorkContext
);
}
} // SrvQueueWorkToFsp
NTSTATUS
SrvQueueWorkToFspAtSendCompletion (
IN PDEVICE_OBJECT DeviceObject,
IN PIRP Irp,
IN PWORK_CONTEXT WorkContext
)
/*++
Routine Description:
Send completion handler for work items that are to be queued to
a nonblocking worker thread in the FSP. This function is also
called from elsewhere in the server to transfer work to the FSP.
This function should not be called at dispatch level -- use
SrvQueueWorkToFspAtDpcLevel instead.
Arguments:
DeviceObject - Pointer to target device object for the request.
Irp - Pointer to I/O request packet
WorkContext - Caller-specified context parameter associated with IRP.
This is actually a pointer to a Work Context block.
Return Value:
STATUS_MORE_PROCESSING_REQUIRED.
--*/
{
//
// Check the status of the send completion.
//
CHECK_SEND_COMPLETION_STATUS( Irp->IoStatus.Status );
//
// Reset the IRP cancelled bit.
//
Irp->Cancel = FALSE;
//
// Increment the processing count.
//
WorkContext->ProcessingCount++;
//
// Insert the work item on the nonblocking work queue.
//
if( WorkContext->QueueToHead ) {
SrvInsertWorkQueueHead(
WorkContext->CurrentWorkQueue,
(PQUEUEABLE_BLOCK_HEADER)WorkContext
);
} else {
SrvInsertWorkQueueTail(
WorkContext->CurrentWorkQueue,
(PQUEUEABLE_BLOCK_HEADER)WorkContext
);
}
return STATUS_MORE_PROCESSING_REQUIRED;
} // SrvQueueWorkToFspAtSendCompletion
VOID SRVFASTCALL
SrvTerminateWorkerThread (
IN OUT PWORK_CONTEXT WorkItem OPTIONAL
)
/*++
Routine Description:
This routine is called when a thread is being requested to terminate. There
are two cases when this happens. One is at server shutdown -- in this
case we need to keep requeueing the termination request until all the
threads on the queue have terminated. The other time is if a thread has
not received work for some amount of time.
--*/
{
LONG priority = 16;
//
// Raise our priority to ensure that this thread has a chance to get completely
// done before the main thread causes the driver to unload or something
//
NtSetInformationThread (
NtCurrentThread( ),
ThreadBasePriority,
&priority,
sizeof(priority)
);
if( ARGUMENT_PRESENT( WorkItem ) &&
InterlockedDecrement( &WorkItem->CurrentWorkQueue->Threads ) != 0 ) {
//
// We are being asked to terminate all of the worker threads on this queue.
// So, if we're not the last thread, we should requeue the workitem so
// the other threads will terminate
//
//
// There are still other threads servicing this queue, so requeue
// the workitem
//
SrvInsertWorkQueueTail( WorkItem->CurrentWorkQueue,
(PQUEUEABLE_BLOCK_HEADER)WorkItem );
}
PsTerminateSystemThread( STATUS_SUCCESS ); // no return;
}
#if MULTIPROCESSOR
VOID
SrvBalanceLoad(
IN PCONNECTION connection
)
/*++
Routine Description:
Ensure that the processor handling 'connection' is the best one
for the job. This routine is called periodically per connection from
DPC level. It can not be paged.
Arguments:
connection - the connection to inspect
Return Value:
none.
--*/
{
ULONG MyQueueLength, OtherQueueLength;
ULONG i;
PWORK_QUEUE tmpqueue;
PWORK_QUEUE queue = connection->CurrentWorkQueue;
ASSERT( queue >= SrvWorkQueues );
ASSERT( queue < eSrvWorkQueues );
//
// Reset the countdown. After the client performs BalanceCount
// more operations, we'll call this routine again.
//
connection->BalanceCount = SrvBalanceCount;
//
// Figure out the load on the current work queue. The load is
// the sum of the average work queue depth and the current work
// queue depth. This gives us some history mixed in with the
// load *right now*
//
MyQueueLength = queue->AvgQueueDepthSum >> LOG2_QUEUE_SAMPLES;
MyQueueLength += KeReadStateQueue( &queue->Queue );
//
// If we are not on our preferred queue, look to see if we want to
// go back to it. The preferred queue is the queue for the processor
// handling this client's network card DPCs. We prefer to run on that
// processor to avoid sloshing data between CPUs in an MP system.
//
tmpqueue = connection->PreferredWorkQueue;
ASSERT( tmpqueue >= SrvWorkQueues );
ASSERT( tmpqueue < eSrvWorkQueues );
if( tmpqueue != queue ) {
//
// We are not queueing to our preferred queue. See if we
// should go back to our preferred queue
//
ULONG PreferredQueueLength;
PreferredQueueLength = tmpqueue->AvgQueueDepthSum >> LOG2_QUEUE_SAMPLES;
PreferredQueueLength += KeReadStateQueue( &tmpqueue->Queue );
if( PreferredQueueLength <= MyQueueLength + SrvPreferredAffinity ) {
//
// We want to switch back to our preferred processor!
//
IF_DEBUG( REBALANCE ) {
KdPrint(( "%p C%d(%p) > P%p(%d)\n",
connection,
MyQueueLength,
(PVOID)(connection->CurrentWorkQueue - SrvWorkQueues),
(PVOID)(tmpqueue - SrvWorkQueues),
PreferredQueueLength ));
}
InterlockedDecrement( &queue->CurrentClients );
InterlockedExchangePointer( &connection->CurrentWorkQueue, tmpqueue );
InterlockedIncrement( &tmpqueue->CurrentClients );
SrvReBalanced++;
return;
}
}
//
// We didn't hop to the preferred processor, so let's see if
// another processor looks more lightly loaded than we are.
//
//
// SrvNextBalanceProcessor is the next processor we should consider
// moving to. It is a global to ensure everybody doesn't pick the
// the same processor as the next candidate.
//
tmpqueue = &SrvWorkQueues[ SrvNextBalanceProcessor ];
//
// Advance SrvNextBalanceProcessor to the next processor in the system
//
i = SrvNextBalanceProcessor + 1;
if( i >= SrvNumberOfProcessors )
i = 0;
SrvNextBalanceProcessor = i;
//
// Look at the other processors, and pick the next one which is doing
// enough less work than we are to make the jump worthwhile
//
for( i = SrvNumberOfProcessors; i > 1; --i ) {
ASSERT( tmpqueue >= SrvWorkQueues );
ASSERT( tmpqueue < eSrvWorkQueues );
OtherQueueLength = tmpqueue->AvgQueueDepthSum >> LOG2_QUEUE_SAMPLES;
OtherQueueLength += KeReadStateQueue( &tmpqueue->Queue );
if( OtherQueueLength + SrvOtherQueueAffinity < MyQueueLength ) {
//
// This processor looks promising. Switch to it
//
IF_DEBUG( REBALANCE ) {
KdPrint(( "%p %c%p(%d) > %c%p(%d)\n",
connection,
queue == connection->PreferredWorkQueue ? 'P' : 'C',
(PVOID)(queue - SrvWorkQueues),
MyQueueLength,
tmpqueue == connection->PreferredWorkQueue ? 'P' : 'C',
(PVOID)(tmpqueue - SrvWorkQueues),
OtherQueueLength ));
}
InterlockedDecrement( &queue->CurrentClients );
InterlockedExchangePointer( &connection->CurrentWorkQueue, tmpqueue );
InterlockedIncrement( &tmpqueue->CurrentClients );
SrvReBalanced++;
return;
}
if( ++tmpqueue == eSrvWorkQueues )
tmpqueue = SrvWorkQueues;
}
//
// No rebalancing necessary
//
return;
}
#endif