c++ - 使用单独的线程从文件预加载数据

Question

我有一个处理大量(相对较小)文件的小型应用程序。它按顺序运行:从文件加载数据，对其执行操作，然后移动到下一个文件。我注意到在运行时，CPU 使用率不是 100%，我猜这是由于硬盘上的 I/O 操作所花费的时间。

因此，我们的想法是在处理当前数据的同时，使用单独的线程将下一个数据加载到内存中(所讨论的数据只是一个存储在 vector 中的 int 序列)。这似乎是一个非常普遍的问题，但我很难找到一个简单、普通的 C++ 示例来做到这一点!现在 C++0x 即将推出，使用新线程工具的简单演示代码，没有外部库，会非常好。

此外，虽然我知道这取决于很多因素，但是否可以对这种方法的好处(或挫折)进行有根据的猜测，例如要加载的数据文件的大小？我猜对于大文件，磁盘 I/O 操作无论如何都很少，因为数据已经被缓冲(使用 fstream(?))

奥利维尔

Answer 1

关于如何使用某些 C++0x 线程和同步工具的玩具程序。不知道这有什么表现(我推荐马特的回答)，为了举例，我的重点是清晰度和正确性。

根据您的要求，这些文件是单独读取的。它们不会转换为 int 的序列然而，我觉得这更多地与处理相关，而不是严格的 I/O。所以文件被转储到一个普通的 std::string .

#include <fstream>
#include <sstream>
#include <string>
#include <vector>
#include <deque>
#include <future>
#include <mutex>
#include <condition_variable>

int
main()
{
    // this is shared
    std::mutex mutex;
    std::condition_variable condition;
    bool more_to_process = true;
    std::deque<std::string> to_process;

    /* Reading the files is done asynchronously */
    std::vector<std::string> filenames = /* initialize */
    auto process = std::async(std::launch::async, [&](std::vector<std::string> filenames)
    {
        typedef std::lock_guard<std::mutex> lock_type;
        for(auto&& filename: filenames) {
            std::ifstream file(filename);
            if(file) {
                std::ostringstream stream;
                stream << file.rdbuf();
                if(stream) {
                    lock_type lock(mutex);
                    to_process.push_back(stream.str());
                    condition.notify_one();
                }
            }
        }
        lock_type lock(mutex);
        more_to_process = false;
        condition.notify_one();
    }, std::move(filenames));

    /* processing is synchronous */
    for(;;) {
        std::string file;
        {
            std::unique_lock<std::mutex> lock(mutex);
            condition.wait(lock, [&]
            { return !more_to_process || !to_process.empty(); });

            if(!more_to_process && to_process.empty())
                break;
            else if(to_process.empty())
                continue;

            file = std::move(to_process.front());
            to_process.pop_front();
        }

        // use file here
    }

    process.get();
}

一些注意事项:

• 互斥体、条件变量、停止标志和std::string container 在逻辑上都是相关的。您也可以将它们替换为线程安全的容器/ channel
• 我使用 std::async而不是 std::thread因为它有更好的异常安全特性
• 没有错误处理可言；如果由于某种原因无法读取文件，则会自动跳过该文件。你有几种选择:发出没有更多需要处理的信号并尽快抛出处理；或使用 boost::variant<std::string, std::exception_ptr>将错误传递给事物的处理端(此处错误作为异常传递，但您可以使用 error_code 或您喜欢的任何东西)。无论如何都不是详尽的 list 。