私のプログラムでは、プロデューサー スレッドがテキスト ファイル (約 8000 行のテキスト) からテキスト行を読み取り、その行を同時実行キューに読み込みます。
3 つのコンシューマー スレッドがキューから行を読み取り、それぞれが別のファイルに書き込みます。
プログラムを実行すると、プロデューサー スレッドとコンシューマー スレッドの 1 つだけが完了します。他の 2 つのスレッドはハングしているようです。
すべてのコンシューマー スレッドにファイルの終わりに到達したことを確実に伝えるにはどうすればよいでしょうか。
私のプラットフォームは Windows 7 64 ビットです
VC11.
64 ビットと 32 ビットとしてコンパイルされたコードは同じ動作をしました。
これがコードです。(自己完結型でコンパイル可能です)
#include <queue>
#include<iostream>
#include<fstream>
#include <atomic>
#include <thread>
#include <condition_variable>
#include <mutex>
#include<string>
#include<memory>
template<typename Data>
class concurrent_queue
{
private:
std::queue<Data> the_queue;
mutable std::mutex the_mutex;
std::condition_variable the_condition_variable;
public:
void push(Data const& data){
{
std::lock_guard<std::mutex> lock(the_mutex);
the_queue.push(data);
}
the_condition_variable.notify_one();
}
bool empty() const{
std::unique_lock<std::mutex> lock(the_mutex);
return the_queue.empty();
}
const size_t size() const{
std::lock_guard<std::mutex> lock(the_mutex);
return the_queue.size();
}
bool try_pop(Data& popped_value){
std::unique_lock<std::mutex> lock(the_mutex);
if(the_queue.empty()){
return false;
}
popped_value=the_queue.front();
the_queue.pop();
return true;
}
void wait_and_pop(Data& popped_value){
std::unique_lock<std::mutex> lock(the_mutex);
while(the_queue.empty()){
the_condition_variable.wait(lock);
}
popped_value=the_queue.front();
the_queue.pop();
}
};
std::atomic<bool> done(true);
typedef std::vector<std::string> segment;
concurrent_queue<segment> data;
const int one_block = 15;
void producer()
{
done.store(false);
std::ifstream inFile("c:/sample.txt");
if(!inFile.is_open()){
std::cout << "Can't read from file\n";
return;
}
std::string line;
segment seg;
int cnt = 0;
while(std::getline(inFile,line)){
seg.push_back(line);
++cnt;
if( cnt == one_block ){
data.push( seg );
seg.clear();
cnt = 0;
}
}
inFile.close();
done.store(true);
std::cout << "all done\n";
}
void consumer( std::string fname)
{
std::ofstream outFile(fname.c_str());
if(!outFile.is_open()){
std::cout << "Can't write to file\n";
return;
}
do{
while(!data.empty()){
segment seg;
data.wait_and_pop( seg );
for(size_t i = 0; i < seg.size(); ++i)
{
outFile << seg[i] << std::endl;
}
outFile.flush();
}
} while(!done.load());
outFile.close();
std::cout << fname << " done.\n";
}
int main()
{
std::thread th0(producer);
std::thread th1(consumer, "Worker1.txt");
std::thread th2(consumer, "Worker2.txt");
std::thread th3(consumer, "Worker3.txt");
th0.join();
th1.join();
th2.join();
th3.join();
return 0;
}