如何多线程读取C++ 11中的文件？

Question

我有一个大文件，我必须通过块阅读它。每次当我读块时，我都要做一些耗时的操作，所以我认为多线程读取可能会有所帮助，每个线程都会逐个读取一块并执行操作。这是我在C++编写代码11

#include<iostream> 
#include<fstream> 
#include <condition_variable> 
#include <mutex> 
#include <thread> 
using namespace std; 
const int CHAR_PER_FILE = 1e8; 
const int NUM_THREAD = 2; 
int order = -1; 
bool is_reading = false; 
mutex mtx; 
condition_variable file_not_reading; 
void partition(ifstream& is) 
{ 
    while (is.peek() != EOF) 
    { 
     unique_lock<mutex> lock(mtx); 
     while (is_reading) 
      file_not_reading.wait(lock); 

     is_reading = true; 
     char *c = new char[CHAR_PER_FILE]; 

     is.read(c, CHAR_PER_FILE); 
     order++; 

     is_reading = false; 

     file_not_reading.notify_all(); 
     lock.unlock(); 

     char oc[3]; 
     sprintf(oc, "%d", order); 
     this_thread::sleep_for(chrono::milliseconds(2000));//some operations that take long time 
     ofstream os(oc, ios::binary); 
     os.write(c, CHAR_PER_FILE); 
     delete[] c; 
     os.close(); 
    } 
} 

int main() 
{ 
    ifstream is("bigfile.txt",ios::binary); 
    thread threads[NUM_THREAD]; 
    for (int i = 0; i < NUM_THREAD; i++) 
     threads[i] = thread(partition, ref(is)); 

    for (int i = 0; i < NUM_THREAD; i++) 
     threads[i].join(); 

    is.close(); 
    system("pause"); 
    return 0; 
} 

但我的代码没有工作，它不仅创造，而不是`bigfilesize/CHAR_PER_FILE 4个文件，和线程似乎卡住了，我怎样才能使它发挥作用？

是否有任何c + + 11多线程读取文件的实现或例子？

谢谢。

Answer 1

我的建议是：

• 使用一个线程从文件中读取数据块。每次读取块时，都会将其发布到请求队列中。这是不值得阅读多线程，因为会有内部锁/阻塞阅读一个共同的资源。
• 使用线程池。他们每个人都从队列中读取，检索一个块，执行昂贵的操作并返回等待一个新的请求。
• 队列必须受到互斥锁保护。
• 请勿使用比您拥有的处理单元（CPU/Cores/HyperThreads）数量多的线程。
• 以上主要警告是它不能保证处理顺序。您可能需要将结果发布到可重新排序的中央位置（再次集中位置 - >必须受到互斥锁保护）。

Answer 2

该文件是否必须以“顺序”顺​​序读取，即组块是否必须按特殊顺序“操作”？否则，你可以例如创建4个线程并让每个线程读取文件的1/4（可以通过使用tellg并在例如向量或变量中保存位置来完成）。这样你就不必使用锁。

也许你可以告诉我们你读的数据是如何评估的。

Answer 3

也许......

void partition(ifstream& is) 
{ 
    unique_lock<mutex> lock(mtx); 
    std::vector<char> c(CHAR_PER_FILE); 
    is.read(c.data(), CHAR_PER_FILE); 
    lock.unlock(); 

    if (is.fail() && !is.eof()) return; 
    size_t num_bytes_read = is.gcount(); 

    std::ostringstream oc; 
    oc << order; 
    this_thread::sleep_for(chrono::milliseconds(2000)); //take long time 
    if (std::ofstream os(oc, ios::binary)) 
     os.write(c.data(), CHAR_PER_FILE); 
} 

注：

• 互斥已串行化的操作 - 无需使用条件变量。

• 我添加了一个小错误输入和字节读取处理 - os.write()后，你应该检查过，添加一个else失败ofstream创作等

Answer 4

你可以使用基于任务的并行std :: async：

class result; // result of expensive operation 
result expensive_operation(std::vector<char> const& data) 
{ 
result r = // long computation 
return r; 
} 

std::vector<char>::size_type BLOCK_SIZE = 4096; 

std::vector<std::future<result>> partition(ifstream& in) 
{ 
    std::vector<std::future<result>> tasks; 

    while (!in.eof() && !in.fail()) 
    { 
     std::vector<char> c(BLOCK_SIZE); 
     is.read(c.data(), BLOCK_SIZE); 
     c.resize(in.gcount()); 
     tasks.push_back(std::async([](std::vector<char> data) 
     { 
      return expensive_operation(data); 
     }, 
     std::move(c))); 
    } 
    return tasks; 
} 

int main() 
{ 
    ifstream is("bigfile.txt",ios::binary); 
    auto results = partition(is); 

    // iterate over results and do something with it 
}