How to go about multithreading with "priority"?

Question

I have multiple threads processing multiple files in the background, while the program is idle.
To improve disk throughput, I use critical sections to ensure that no two threads ever use the same disk simultaneously.

The (pseudo-)code looks something like this:

void RunThread(HANDLE fileHandle)
{
    // Acquire CRITICAL_SECTION for disk
    CritSecLock diskLock(GetDiskLock(fileHandle));

    for (...)
    {
        // Do some processing on file
    }
}

Once the user requests a file to be processed, I need to stop all threads -- except the one which is processing the requested file. Once the file is processed, then I'd like to resume all the threads again.

Given the fact that SuspendThread is a bad idea, how do I go about stopping all threads except the one that is processing the relevant input?

What kind of threading objects/features would I need -- mutexes, semaphores, events, or something else? And how would I use them? (I'm hoping for compatibility with Windows XP.)

Answer 1

The difficulty here isn't priority as such, it's the fact that you want a thread to back out of a lock that it's holding, to let another thread take it. "Priority" relates to which of a set of runnable threads should be scheduled to run -- you want to make a thread runnable that isn't (because it's waiting on a lock held by another thread).

So, you want to implement (as you put it):

if (ThisThreadNeedsToSuspend()) { ReleaseDiskLock(); WaitForResume(); ReacquireDiskLock(); }

Since you're (wisely) using a scoped lock I would want to invert the logic:

while (file_is_not_finished) {
    WaitUntilThisThreadCanContinue();
    CritSecLock diskLock(blah);
    process_part_of_the_file();
}
ReleasePriority();

...

void WaitUntilThisThreadCanContinue() {
    MutexLock lock(thread_priority_mutex);
    while (thread_with_priority != NOTHREAD and thread_with_priority != thisthread) {
        condition_variable_wait(thread_priority_condvar);
    }
}

void GiveAThreadThePriority(threadid) {
    MutexLock lock(thread_priority_mutex);
    thread_with_priority = threadid;
    condition_variable_broadcast(thread_priority_condvar);
}

void ReleasePriority() {
    MutexLock lock(thread_priority_mutex);
    if (thread_with_priority == thisthread) {
        thread_with_priority = NOTHREAD;
        condition_variable_broadcast(thread_priority_condvar);
    }
}

Read up on condition variables -- all recent OSes have them, with similar basic operations. They're also in Boost and in C++11.

If it's not possible for you to write a function process_part_of_the_file then you can't structure it this way. Instead you need a scoped lock that can release and regain the disklock. The easiest way to do that is to make it a mutex, then you can wait on a condvar using that same mutex. You can still use the mutex/condvar pair and the thread_with_priority object in much the same way.

You choose the size of "part of the file" according to how responsive you need the system to be to a change in priority. If you need it to be extremely responsive then the scheme doesn't really work -- this is co-operative multitasking.

I'm not entirely happy with this answer, the thread with priority can be starved for a long time if there are a lot of other threads that are already waiting on the same disk lock. I'd put in more thought to avoid that. Possibly there should not be a per-disk lock, rather the whole thing should be handled under the condition variable and its associated mutex. I hope this gets you started, though.

Answer 2

OK, how about this:

Two threads per disk, for high and low priority requests, each with its own input queue.

A high-priority disk task, when initially submitted, will then issue its disk requests in parallel with any low-priority task that is running. It can reset a ManualResetEvent that the low-priority thread waits on when it can, (WaitForSingleObject) and so will get blocked if the high-prioriy thread is perfoming disk ops. The high-priority thread should set the event after finishing a task.

This should limit the disk-thrashing to the interval, (if any), between the submission of the high-priority task and whenver the low-priority thread can wait on the MRE. Raising the CPU priority of the thread servicing the high-priority queue may assist in improving performance of the high-priority work in this interval.

Edit: by 'queue', I mean a thread-safe, blocking, producer-consumer queue, (just to be clear:).

More edit - if the issuing threads needs notification of job completion, the tasks issued to the queues could contain an 'OnCompletion' event to call with the task object as a parameter. The event handler could, for example, signal an AutoResetEvent that the originating thread is waiting on, so providing synchronous notification.

Answer 3

You can also use the pool of threads and regulate their work by using the I/O Completion port.

Normally threads from the pool would sleep awaiting for the I/O Completion port event/activity. When you have a request the I/O Completion port releases the thread and it starts to do a job.

Answer 4

After each worker thread processes a file, check a condition variable associated with that thread. The condition variable could implemented simply as a bool + critical section. Or with InterlockedExchange* functions. And to be honest, I usually just use an unprotected bool between threads to signal "need to exit" - sometimes with an event handle if the worker thread could be sleeping.

After setting the condition variable for each thread, Main thread waits for each thread to exit via WaitForSingleObject.

DWORD __stdcall WorkerThread(void* pThreadData)
{
    ThreadData* pData = (ThreadData*) pTheradData;

    while (pData->GetNeedToExit() == false)
    {
        ProcessNextFile();
    }
    return 0;
}

void StopWokerThread(HANDLE hThread, ThreadData* pData)
{
   pData->SetNeedToExit = true;
   WaitForSingleObject(hThread);
   CloseHandle(hThread);
}

struct ThreadData()
{
    CRITICAL_SECITON _cs;
    ThreadData()
    {
        InitializeCriticalSection(&_cs);
    }
    ~ThreadData()
    {
        DeleteCriticalSection(&_cs);
    }

    ThreadData::SetNeedToExit()
    {
        EnterCriticalSection(&_cs);
          _NeedToExit = true;
        LeaveCriticalSeciton(&_cs);
    }

    bool ThreadData::GetNeedToExit()
    {
        bool returnvalue;
        EnterCriticalSection(&_cs);
          returnvalue = _NeedToExit = true;
        LeaveCriticalSeciton(&_cs);
        return returnvalue;
    }

};

Answer 5

You may ask the threads to stop gracefully. Just check some variable in loop inside threads and continue or terminate work depending on its value.

Some thoughts about it:

• The setting and checking of this value should be done inside critical section.
• Because the critical section slows down the thread, the checking should be done often enough to quickly stop the thread when needed and rarely enough, such that thread won't be stalled by acquiring and releasing the critical section.

Answer 6

I recommend you go about it in a completely different fashion. If you really want only one thread for every disk (I'm not convinced this is a good idea) then you should create one thread per disk, and distribute files as you queue them for processing.

To implement priority requests for specific files I would then have a thread check a "priority slot" at several points during its normal processing (and of course in its main queue wait loop).