Discussion :

[kidpaddle2]

Bonjour,

J'ai aléatoirement des erreurs du type :

1. Memory cloberred before allocated block ; Aborted, des fois après Rd3 ready
2. Segmentation fault Edit: Apparemment, le Sigabrt vient exclusivement à la ligne writer.run();
3. Munmap_chunk() : invalid pointer (faisant suite à 1.) après Rd2 ready / Rd1 ready, mais je ne sais pas si c'est une coïncidence.

Avec le code suivant :
Main.cpp

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
#include "io.hpp"
 
#include <iostream>
 
#include "debug.hpp"
 
int main(int argc, char *argv[]) {
    CSV::Reader rd1, rd2, rd3;
    CSV::Writer writer;
 
    bool success_open = rd1.open("first_file.csv");
    success_open &= rd2.open("second_file.csv");
    success_open &= rd3.open("third_file.csv");
 
    if(success_open && writer.open("output.txt")) {
        CSV::Row rd1_row, rd2_row, rd3_row;
 
        rd1.run(); //Reads the CSV file and fills the read queue
        rd2.run();
        rd3.run();
        writer.run(); //Reads the to-be-written queue and writes it to a txt file
 
        bool rd1_ready = false, rd2_ready = false, rd3_ready = false;
 
        //The loop is supposed to end only if the reader is finished and empty
        while(!(rd1.is_finished() && rd1.empty() &&
                rd2.is_finished() && rd2.empty() &&
                rd3.is_finished() && rd3.empty())) {
            //Line by line
            if(!rd1_ready) {
                if(rd1_ready = rd1.try_pop(rd1_row)) {
                    Debug::print("rd1 ready");
                }
            }
            if(!rd2_ready) {
                if(rd2_ready = rd2.try_pop(rd2_row)) {
                    Debug::print("rd2 ready");
                }
            }
            if(!rd3_ready) {
                if(rd3_ready = rd3.try_pop(rd3_row)) {
                    Debug::print("rd3 ready");
                }
            }
 
            if(rd1_ready && rd2_ready && rd3_ready) {
                writer.push(rd1_row);
                Debug::print("Written");
 
                rd1_ready = false;
                rd2_ready = false;
                rd3_ready = false;
            }
        }
        //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 :

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
#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;
 
                //The writer ends only if he's finishing and empty
                while(!(m_finishing.get() && empty())) {
                    if(try_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 :

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
#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;
}

threads.hpp :

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
#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, m_thread = NULL;
        }
    }
 
    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();
    }
 
    bool try_pop(Data& popped) {
        boost::mutex::scoped_lock lock(m_mutex);
        bool ret = false;
 
        if(!m_queue.empty()) {
            popped = m_queue.front();
            m_queue.pop();
            ret = true;
        }
        return ret;
    }
};
 
#endif // THREAD_HPP_INCLUDED

Debug.hpp :

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
#ifndef DEBUG_HPP_INCLUDED
#define DEBUG_HPP_INCLUDED
 
#include <boost/thread/mutex.hpp>
 
class Debug {
    static boost::mutex m_io_mutex;
public:
    static void print(char const *str) {
        boost::mutex::scoped_lock lock(Debug::m_io_mutex);
 
        std::cout << str << std::endl;
    }
};
 
boost::mutex Debug::m_io_mutex;
 
#endif // DEBUG_HPP_INCLUDED

Je n'ai jamais rencontré de memory cloberred auparavant, donc j'ai l'impression que cela porte sur un mauvais multithread, mais où ? Au niveau des mutex ?

Merci d'avance.

Cordialement,

Kidpaddle2

[zul]

Ça me fait penser à des allocations / desallocations / pointeurs fou / (iterateur invalide) / (reference invalide ...) . Je pense que le mieux à faire c'est de sortir un outil comme valgrind qui te dira bien plus précisement où est le problème (sous Linux) (y'en a un bien cher sous windows dont j'ai oublié le nom)).

Méfie toi de tes conditions du genre while(!(rd1.is_finished() && rd1.empty() ... ou while(!(m_finishing.get() && empty())) , comme tu ne locke pas globalement ta structure de donnée, les conditions peuvent être évalués à vrai dans la condition, et être fausse dans le corps de la boucle (parce qu'il s'est passé quelquechose entre ces deux moments distincts)).

[Bisouuux]

Bonjour,

Nous sommes 2 à bosser sur ce projet (kidpaddle2 et moi-même), aussi ais-je fait des tests sous linux (ubuntu 10.04), et le comportement est quand même super étrange.

Je m'explique :
En lançant le programme en console, la tendance est que l'on a 2 fois sur 3 l'histoire du memory clobbered before allocated block, de quoi résulte un sigabrt.

En revanche, en lançant sous valgrind (en mode memcheck), tout fonctionne PARFAITEMENT, le programme se termine, le fichier de sortie est intègre et correspond à ce que l'on veut.

Je me permets de joindre une partie du log valgrind, dont je suppose que seule la fin est intéressante.

Merci d'avance.

@zul : en effet, ces conditions sont problématiques, puisqu'après débuggage, il semblerait qu'avant le fameux message d'erreur qui nous embete tant, le programme soit bloqué dans une boucle infinie, dans la boucle while de la fonction work() du writer...

[zul]

Le problème avec Valgrind, c'est que ça ralentit sensiblement l'execution, et donc que du coup, certains problèmes de synchronisation ne sont plus reproductibles (et donc difficile à débugger)). Je peux difficilement en dire plus. Mais globalement il y'a quand même un problème dans le design : ça ne sert à rien de faire des tests si lorsque vous vous servez du résultat du test, celui-ci a potentiellement changé.