如何以“优先级”进行多线程？

Question

我有多个线程在后台处理多个文件，而程序处于空闲状态。
为了提高磁盘吞吐量，我使用临界区来确保不会有两个线程同时使用同一个磁盘。

（伪）代码如下所示：

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

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

一旦用户请求处理文件，我需要停止所有线程 -- 除了正在处理请求文件的线程。处理完文件后，我想再次恢复所有线程。

鉴于SuspendThread 是个坏主意，我该如何停止除处理相关输入的线程之外的所有线程？

我需要什么样的线程对象/功能——互斥体、信号量、事件或其他？我将如何使用它们？ （我希望与 Windows XP 兼容。）

Answer 1

我建议您以完全不同的方式进行。如果您真的希望每个磁盘只有一个线程（我不相信这是一个好主意），那么您应该为每个磁盘创建一个线程，并在排队处理文件时分发文件。

为了实现对特定文件的优先级请求，我会让一个线程在其正常处理过程中的多个点检查“优先级槽”（当然在其主队列等待循环中）。

Answer 2

这里的困难不是优先级本身，而是你希望一个线程退出它持有的锁，让另一个线程拿走它。 “优先级”与应安排运行一组可运行线程中的哪一个有关——您希望使一个线程不可运行（因为它正在等待另一个线程持有的锁）。

所以，你想实现（如你所说）：

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

由于您（明智地）使用范围锁，我想反转逻辑：

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);
    }
}

阅读条件变量——所有最近的操作系统都有它们，具有类似的基本操作。它们也在 Boost 和 C++11 中。

如果您无法编写函数process_part_of_the_file，那么您就不能以这种方式构造它。相反，您需要一个可以释放和重新获得磁盘锁的作用域锁。最简单的方法是使其成为互斥锁，然后您可以使用相同的互斥锁等待 condvar。您仍然可以以几乎相同的方式使用 mutex/condvar 对和 thread_with_priority 对象。

您可以根据需要系统对优先级更改的响应程度来选择“文件的一部分”的大小。如果您需要它非常响应，那么该方案实际上并不起作用——这是合作多任务处理。

我对这个答案并不完全满意，如果有很多其他线程已经在同一个磁盘锁上等待，那么具有优先级的线程可能会被饿死很长时间。为了避免这种情况，我会考虑更多。可能不应该有每个磁盘的锁，而是应该在条件变量及其关联的互斥锁下处理整个事情。不过，我希望这能让你开始。

Answer 3

您可以要求线程优雅地停止。只需检查线程内循环中的一些变量，然后根据其值继续或终止工作。

关于它的一些想法：

• 此值的设置和检查应在临界区内完成。
• 因为临界区会减慢线程的速度，所以检查应该经常进行，以便在需要时快速停止线程，并且很少进行检查，这样线程就不会因获取和释放临界区而停止。

Answer 4

每个工作线程处理文件后，检查与该线程关联的条件变量。条件变量可以简单地实现为布尔+临界区。或使用 InterlockedExchange* 功能。老实说，我通常只是在线程之间使用不受保护的布尔值来表示“需要退出”——如果工作线程可能正在休眠，有时会使用事件句柄。

为每个线程设置条件变量后，主线程通过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

您还可以使用线程池并通过使用I/O Completion port 来调节它们的工作。

池中的线程通常会休眠以等待 I/O 完成端口事件/活动。 当您有请求时，I/O 完成端口会释放线程并开始执行工作。

Answer 6

好的，这个怎么样：

每个磁盘有两个线程，用于高优先级和低优先级请求，每个线程都有自己的输入队列。

最初提交的高优先级磁盘任务随后将与任何正在运行的低优先级任务并行发出其磁盘请求。它可以重置低优先级线程在可能时等待的 ManualResetEvent（WaitForSingleObject），因此如果高优先级线程正在执行磁盘操作，它将被阻塞。高优先级线程应该在完成任务后设置事件。

这应该将磁盘抖动限制在提交高优先级任务和低优先级线程可以在 MRE 上等待之间的时间间隔（如果有）。提高为高优先级队列提供服务的线程的 CPU 优先级可能有助于提高此时间间隔内高优先级工作的性能。

编辑：“队列”是指线程安全的、阻塞的、生产者-消费者队列，（明确一点：）。

更多编辑 - 如果发布线程需要通知作业完成，则发布到队列的任务可以包含一个“OnCompletion”事件，以使用任务对象作为参数进行调用。例如，事件处理程序可以发出一个 AutoResetEvent 信号，表明发起线程正在等待，从而提供同步通知。