我想基于boost::thread逐行实现并行读-处理-写，但当前版本行为不定：下面的测试通过填充读取（并发）队列来读取一个CSV文件，该队列被简单地传输到要写入输出文件的写入队列（暂时不处理）。
遇到的问题：
- 在 Windows 和 Unix 上，程序随机永远不会结束（~3/5 次），并生成一个 SIGSEGV（~1/100）
- 在 Unix 上，有很多错误：创建线程时出现 SIGABRT，创建后“内存在分配块之前被破坏”（-> SIGABRT 也是如此），随机在 1 到 ~15 行之间。
我讨厌给出问题和代码并让其他人回答（我有时会站在你这边），但相信我，我想不出任何其他方法来纠正它（处理线程，调试是一场噩梦） ，所以我提前道歉。这里是 ：
主要.cpp：
#include "io.hpp"
#include <iostream>
int main(int argc, char *argv[]) {
CSV::Reader reader;
CSV::Writer writer;
if(reader.open("test_grandeur_nature.csv") && writer.open("output.txt")) {
CSV::Row row;
reader.run(); //Reads the CSV file and fills the read queue
writer.run(); //Reads the to-be-written queue and writes it to a txt file
//The loop is supposed to end only if the reader is finished and empty
while(!(reader.is_finished() && reader.empty())) {
//Transfers line by line from the read to the to-be-written queues
reader.wait_and_pop(row);
writer.push(row);
}
//The reader will likely finish before the writer, so he has to finish his queue before continuing.
writer.finish();
}
else {
std::cout << "File error";
}
return EXIT_SUCCESS;
}
io.hpp：
#ifndef IO_H_INCLUDED
#define IO_H_INCLUDED
#include "threads.hpp"
#include <fstream>
namespace CSV {
class Row {
std::vector<std::string> m_data;
friend class Iterator;
friend void write_row(Row const &row, std::ostream &stream);
void read_next(std::istream& csv);
public:
inline std::string const& operator[](std::size_t index) const {
return m_data[index];
}
inline std::size_t size() const {
return m_data.size();
}
};
/** Reading *************************************************************************/
class Iterator {
public:
Iterator(std::istream& csv) : m_csv(csv.good() ? &csv : NULL) {
++(*this);
}
Iterator() : m_csv(NULL) {}
//Pre-Increment
Iterator& operator++() {
if (m_csv != NULL) {
m_row.read_next(*m_csv);
m_csv = m_csv->good() ? m_csv : NULL;
}
return *this;
}
inline Row const& operator*() const {
return m_row;
}
inline bool operator==(Iterator const& rhs) {
return ((this == &rhs) || ((this->m_csv == NULL) && (rhs.m_csv == NULL)));
}
inline bool operator!=(Iterator const& rhs) {
return !((*this) == rhs);
}
private:
std::istream* m_csv;
Row m_row;
};
class Reader : public Concurrent_queue<Row>, public Thread {
std::ifstream m_csv;
Thread_safe_value<bool> m_finished;
void work() {
if(!!m_csv) {
for(Iterator it(m_csv) ; it != Iterator() ; ++it) {
push(*it);
}
m_finished.set(true);
}
}
public:
Reader() {
m_finished.set(false);
}
inline bool open(std::string path) {
m_csv.open(path.c_str());
return !!m_csv;
}
inline bool is_finished() {
return m_finished.get();
}
};
/** Writing ***************************************************************************/
void write_row(Row const &row, std::ostream &stream);
//Is m_finishing really thread-safe ? By the way, is it mandatory ?
class Writer : public Concurrent_queue<Row>, public Thread {
std::ofstream m_csv;
Thread_safe_value<bool> m_finishing;
void work() {
if(!!m_csv) {
CSV::Row row;
while(!(m_finishing.get() && empty())) {
wait_and_pop(row);
write_row(row, m_csv);
}
}
}
public:
Writer() {
m_finishing.set(false);
}
inline void finish() {
m_finishing.set(true);
catch_up();
}
inline bool open(std::string path) {
m_csv.open(path.c_str());
return !!m_csv;
}
};
}
#endif
io.cpp：
#include "io.hpp"
#include <boost/bind.hpp>
#include <boost/tokenizer.hpp>
void CSV::Row::read_next(std::istream& csv) {
std::string row;
std::getline(csv, row);
boost::tokenizer<boost::escaped_list_separator<char> > tokenizer(row, boost::escaped_list_separator<char>('\\', ';', '\"'));
m_data.assign(tokenizer.begin(), tokenizer.end());
}
void CSV::write_row(Row const &row, std::ostream &stream) {
std::copy(row.m_data.begin(), row.m_data.end(), std::ostream_iterator<std::string>(stream, ";"));
stream << std::endl;
}
线程.hpp：
#ifndef THREADS_HPP_INCLUDED
#define THREADS_HPP_INCLUDED
#include <boost/bind.hpp>
#include <boost/thread.hpp>
class Thread {
protected:
boost::thread *m_thread;
virtual void work() = 0;
void do_work() {
work();
}
public:
Thread() : m_thread(NULL) {}
virtual ~Thread() {
catch_up();
if(m_thread != NULL) {
delete m_thread;
}
}
inline void catch_up() {
if(m_thread != NULL) {
m_thread->join();
}
}
void run() {
m_thread = new boost::thread(boost::bind(&Thread::do_work, boost::ref(*this)));
}
};
/** Thread-safe datas **********************************************************/
#include <queue>
#include <boost/thread/mutex.hpp>
#include <boost/thread/condition.hpp>
template <class T>
class Thread_safe_value : public boost::noncopyable {
T m_value;
boost::mutex m_mutex;
public:
T const &get() {
boost::mutex::scoped_lock lock(m_mutex);
return m_value;
}
void set(T const &value) {
boost::mutex::scoped_lock lock(m_mutex);
m_value = value;
}
};
template<typename Data>
class Concurrent_queue {
std::queue<Data> m_queue;
mutable boost::mutex m_mutex;
boost::condition_variable m_cond;
public:
void push(Data const& data) {
boost::mutex::scoped_lock lock(m_mutex);
m_queue.push(data);
lock.unlock();
m_cond.notify_one();
}
bool empty() const {
boost::mutex::scoped_lock lock(m_mutex);
return m_queue.empty();
}
void wait_and_pop(Data& popped) {
boost::mutex::scoped_lock lock(m_mutex);
while(m_queue.empty()) {
m_cond.wait(lock);
}
popped = m_queue.front();
m_queue.pop();
}
};
#endif // THREAD_HPP_INCLUDED
这个项目很重要，如果您能帮助我，我将不胜感激 =)
预先感谢。
问候，
密斯泰尔先生。