std :: thread - 從文件行中逐行讀取

Question

我想從輸出文件中逐行讀取。每個線程讀取一行然後處理數據。同時下一個線程必須讀取下一行。

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

當我運行此代碼時，我得到： 第一個線程讀取所有行。我怎樣才能讓每個線程讀取一行？

編輯

正如評論所有我需要做的是提到大的測試文件。 謝謝你們！

Answer 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 ::法院東西是你的測試循環的忙碌一部分以後要換取真正的代碼。這給了其他線程的時間踢。此外，讓你的測試文件大。線程初始化花費一些時間，第一個線程吃掉所有數據並不少見。

Answer 2

爲了防止每個線程讀取一行（這在您的描述中很明顯），請刪除while循環，然後您需要確保線程數與文件中的行數相同。

要擺脫上述限制，您可以使用boost線程池。

Answer 3

如果你想讓你的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; 
}