代码之家  ›  专栏  ›  技术社区  ›  Mister Mystère

随机无休止的线程和各种错误,同时并行逐行读/写

  •  -1
  • Mister Mystère  · 技术社区  · 16 年前

    遇到的问题:

    • 在Unix上,有许多错误:SIGABRT在创建线程时,“内存在分配块之前关闭”(>SIGABRT)创建后,随机在1到15行之间。

    我讨厌给出问题和代码,让其他人回答(我有时是站在你这边的话题),但相信我,我想不出任何其他方法来纠正它(处理线程,调试是一场噩梦),所以我提前道歉。在这里:

    #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
    

    #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
    

    事先谢谢。

    当做,

    2 回复  |  直到 16 年前
        1
  •  1
  •   Anthony Williams    16 年前

    完成逻辑有错误。

    main() 循环正在从队列中读取最后一个条目,并阻塞等待下一个条目 m_finished 标志已设置。

    如果你在打电话之前等了很久 m_finished.set(true) sleep(5) Sleep(5000)

    (这并不能解决seg故障或内存分配错误,这些可能是其他原因)

    有问题的执行是这样的:

    1. 读取器线程从文件中读取最后一项并推送队列。
    2. 主线程从队列中弹出最后一项。
    3. 主螺纹环; m\u完成 未设置,因此它调用 wait_and_pop
    4. 读取器线程意识到它位于文件和集合的末尾 m\u完成

    sleep调用通过在reader线程的第1步和第5步之间设置一个很大的延迟来强制执行这个事件顺序,因此主线程有足够的机会执行第2步到第4步。这是一种针对竞争条件的有用调试技术。

        2
  •  -1
  •   SoapBox    16 年前

    在快速阅读了我发现的唯一明显的问题之后,这可能解释了你的一些(但也许不是全部)问题,那就是你没有正确地发出信号说明你的身体状况 Concurrent_queue::push

    任何时候你打电话来 unlock() 在一个作用域互斥体上,应该向您指示有什么地方出错了。使用作用域互斥锁的一个要点是,当对象进入/离开作用域时,锁定和解锁是隐式的。如果您发现自己需要解锁,那么可能需要重新构造代码。

    void push(Data const& data) {
        boost::mutex::scoped_lock lock(m_mutex);
        m_queue.push(data);
        m_cond.notify_one();
    }
    

    当函数返回时,在发出条件信号后,它将解锁互斥锁。