std::thread - 逐行读取文件答案

【问题标题】：std::thread - read from file line by linestd::thread - 逐行读取文件
【发布时间】：2015-01-03 00:00:34
【问题描述】：

我想从输出文件中逐行并行读取。每个线程读取一行然后处理数据。同时下一个线程必须读取下一行。

std::ifstream infile("test.txt");
std::mutex mtx;

void read(int id_thread){
   while(infile.good()){
     mtx.lock();
     std::string sLine;
     getline(infile, sLine);
     std::cout << "Read by thread: " << id_thread;
     std::cout << sLine << std::endl;
     mtx.unlock();
   }
}

void main(){
  std::vector<std::thread> threads;
  for(int i = 0; i < num; i++){
     threads.push_back(std::thread(parallelFun, i));
  }

  for(auto& thread : threads){
      thread.join();
  }
  return 0;
}

当我运行这段代码时，我得到了这个：第一个线程读取所有行。我怎样才能让每个线程都读一行？

编辑

正如 cmets 中所述，我需要做的只是更大的测试文件。谢谢大家！

【问题讨论】：

有什么问题？无论如何读取都是由互斥锁序列化的。
线程就是这样工作的，它们运行一段时间，然后让另一个线程运行一段时间，以此类推。如果一个线程运行足够长的时间来读取一个短文件中的所有行，那么你必须想办法一次只读取一行（比如发出条件变量的信号？）。
如果您希望读取与处理并行发生，您应该在读取该行后立即解锁互斥锁，而不是在您处理完您读取的数据之后。
有一个可能的问题是，在检查文件是否仍然良好后立即安排线程，仅在文件完成读取后才返回。这不是死锁情况，但它可能会在大文件和许多线程上产生 IO 错误。解决方法是将infile.good() 移动到临界区，并使用其结果更新bool，用于while 循环。
您要求的基本上是线程滥用。说：“[我]想要：第一个线程读取第一行，第二个线程第二行，...，第 n 个线程第 n 行”几乎是在说：“我想要串行执行。”如果要串行执行，请不要使用多个线程。

标签： c++ multithreading file-io line-by-line

【解决方案1】：

我会把循环改成

while(infile.good()){
     mtx.lock();
     std::string sLine;
     getline(infile, sLine);
     mtx.unlock();
     std::cout << "Read by thread: " << id_thread;
     std::cout << sLine << std::endl;
   }

你的 std::cout 东西是你测试循环的繁忙部分，你想稍后交换真实代码。这使其他线程有时间启动。此外，使您的测试文件大。线程初始化需要一些时间而第一个线程吃掉所有数据的情况并不少见。

【讨论】：

在我看来，这是在使用糟糕的代码并使其变得更糟。原件有一个损坏的输入循环，您也同样损坏了它。通过将输出移到 mutex 部分之外，您允许多个线程同时写入，因此输出可能不再一致。
这也不是异常安全的，请使用 std::lock_guard 或其他 RAII 互斥处理程序之一，以确保在下一次循环迭代或函数退出之前始终释放锁
@JerryCoffin 我的假设是这只是一个简约的示例代码，并发线程应该在实际代码中执行一些非常繁重的工作，不一定与同步 I/O 相关。如果问题是“如何在线程之间同步 I/O”，我同意应该提及这一点

【解决方案2】：

如果您希望您的 5 个线程准确地每 5 行读取一次，则必须同步读取，因此每个线程都必须知道前一个线程已完成读取其部分。此要求可能会导致巨大效率低下，因为某些线程可能会等待前一个线程很长时间才能运行。

概念代码，未经测试使用自担风险。

让我们首先创建一个默认类来处理原子锁。我们将其对齐以避免错误共享和相关的缓存乒乓。

constexpr size_t CACHELINESIZE = 64; // could differ on your architecture
template<class dType>
class alignas(CACHELINESIZE) lockstep {
  std::atomic<dType> lock = dType(0);

public:
  // spinlock spins until the previous value is prev and then tries to set lock to value
  // until success, restart the spin if prev changes.
  dType Spinlock(dType prev = dType(0), dType next = dType(1)) {
     dType expected = prev;
     while (!lock.compare_exchange_weak(expected, next)) { // request for locked-exclusiv ~100 cycles?
       expected = prev;  // we wish to continue to wait for expected
       do {
         pause(); // on intel waits roughly one L2 latency time.
       } while(lock.load(std::memory_order_relaxed) != prev);  // only one cache miss per change
     }
     return expected;
  }

  void store(dType value) {
    lock.store(value);
  }
};

lockstep<int> lock { 0 };

constexpr int NoThreads = 5;

std::ifstream infile("test.txt");

void read(int id_thread) {
   locks[id_thread].lock = id_thread;
   bool izNoGood = false;
   int next = id_thread;

   while(!izNoGood){
     // get lock for next iteration
     lock.spinlock(next, next); // wait on our number

     // moved file check into locked region     
     izNoGood = !infile.good();
     if (izNoGood) {
       lock.store(next+1); // release next thread to end run.
       return;
     }

     std::string sLine;
     getline(infile, sLine);

     // release next thread
     lock.store(next+1);

     // do work asynchronous
     // ...

     // debug log, hopefully the whole line gets written in one go (atomic)
     // but can be in "random" order relative to other lines.
     std::cout << "Read by thread: " << id_thread << " line no. " << next
               << " text:" << sLine << std::endl;  // endl flushes cout, implicit sync?
     next += NoThreads;  // our next expected line to process
   }
}

void main() {
  std::vector<std::thread> threads;
  for(int i = 0; i < NoThreads; i++) {
     threads.push_back(std::thread(parallelFun, i));
  }

  for(auto& thread : threads){
      thread.join();
  }
  return 0;
}

【讨论】：

【解决方案3】：

万一您希望每个线程读取一行（这从您的描述中很明显），请删除 while 循环，然后您需要确保线程数与文件中的行数相同。

要摆脱上述限制，您可以使用 boost 线程池。

【讨论】：