écrire avec un thread et lire avec un autre

**lilocratch** · 10/05/2018, 15h58

Bonjour,

J'essai de faire un programme qui lit et écrit en même temps. Le premier thread à terme sera différent mais ce sera qui il écrira continuellement sur le .txt.
Le deuxième thread doit traiter une par une les lignes du document quand elles sont complètes.
J'ai donc fait le programme ci-dessous avec level qui devrait indiquer le nombre de lignes et donc si il reste des lignes à traiter. Mais il n'affiche rien.

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
#include <iostream>
#include <thread>
#include <windows.h>
#include <fstream>
#include <string>
#include <sstream>
#include <mutex>
 
using namespace std;
 
int main()
{   mutex lock;
    string line, checker;
    ostringstream data;
    int h=0,m=0,level=0;
    float s=0;
 
    fstream myfile;
    if (myfile.is_open()) myfile.close();
    myfile.open("data.txt",ios::out | ios::in | ios::trunc);
    thread t1 ([&h,&s,&m,&myfile,&data,&checker,&level](){
 
              while(1)
                {s+=0.1;
                if (s==60)
                    {s=0;m+=1;}
                if (m==60)
                    {m=0;h+=1;}
                if (myfile.is_open()){
                data<<h<<" "<<m<<" "<<s<<endl;
                checker=data.str();
                if(checker.find("\n")) level+=1;
 
                                     }
 
                Sleep(100);
 
 
 
              }
              });
    thread t2 ([&h,&m,&s,&myfile,&line,&level,&lock](){
 
               while(1){
 
 
                 if (myfile.is_open()&& level>0){
                    lock.lock();
                    myfile.seekg(0,myfile.beg);
                    getline(myfile,line);
                    lock.unlock();
                    cout<<line;
                    level-=1;
                 }
 
 
              }
              });
 
 
    t1.join();
    t2.join();
    myfile.close();
    return 0;
}

Merci pour votre aide.

**dalfab** · 10/05/2018, 23h31

Bonjour,

L'avantage des threads c'est cela permet à d'avoir des flux d'exécution qui partagent un même espace mémoire.
Mais attention, il ne faut pas oublier l'énorme inconvénient des threads : on peut utiliser les mêmes données entre plusieurs threads! Et c'est souvent catastrophique.
Il faut donc tout faire pour qu'il n'y ait pas d'accès simultané aux mêmes données.
Le mieux est de n'avoir rien en commun et d'utiliser des mécanismes de communication.
Si on s'y est mal pris, en dernier ressort on peut utiliser des mutex ou des mécanismes de synchro.

Ici les objets utilisés par les deux threads : myfile et level.
Inversement un mutex qui ne serait utilisé que par un thread ne sert à rien (il faut garantir que les 2 ne font pas en même temps qqchose), donc lock n'est certainement pas utilisé correctement ici.
myfile est donc utilisé par 2 threads, ce qui n'a aucun sens. Cet objet est rempli de variables gérant la communication avec un fichier. On ne peut pas avoir 2 threads qui partagent cet objet. Attention, je n'ai jamais dit que deux threads ne pouvaient pas accéder à un unique fichier en même temps.

**Bktero** · 11/05/2018, 08h25

Bonjour,

Code :

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

1
2
3
4
5
6
7
8
9
int main()
{   mutex lock;
    string line, checker;
    ostringstream data;
    int h=0,m=0,level=0;
    float s=0;
 
    // reste du code
}

Woh ! On est plus en C89 ! Il faut déclarer et initialiser ses variables au plus proche de l'endroit où on commence à s'en servir ! Ca évite des erreurs et ça augmente surtout la lisibilité.

Code :

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

1
2
fstream myfile;
    if (myfile.is_open()) myfile.close();

Je ne pense pas que le fichier puisse être ouvert vu que tu n'appelles pas open()...

Et sinon de manière plus globale, pourquoi passer par un fichier ? Ce qu'il te faut, c'est une queue protégée par un mutex que tu partages entre tes deux threads. Le thread producteur mets les éléments au fur et à mesure qu'ils sont prêts ; le thread consommateur les lit au fur et à mesure qu'il finit de traiter les précédents.

**lilocratch** · 14/05/2018, 10h38

Merci pour vos réponses.

Donc si j'ai bien compris, je dois ouvrir le fichier avec un objet différent pour chacun de thread et mieux utiliser le mutex pour gérer la variable level .
Pour rapport à l'initialisation des variables: je la met just avant le thread 1 du coup?

Envoyé par Bktero

Et sinon de manière plus globale, pourquoi passer par un fichier ?

J'ai besoin pendre les données qui arrivent par un port rs232 jusqu'au retour à la ligne. Le fichier me semblait une méthode adaptée puisque je pouvais écrire toutes les données qui arrivent et "couper" là où j'ai besoin.
Je ne connais pas très bien les queues. D'après ce que tu dit elles ont l'air de fonctionné sur le principe FIFO je vais me renseigner et peut-être essayer cette méthode.

Merci encore,

**Bktero** · 14/05/2018, 11h40

En gros, ton thread T1 récupère des caractères depuis la liaison RS232, les mets dans un fichier, puis ton thread T2 relit le fichier pour le traiter ?

**lilocratch** · 14/05/2018, 13h34

Oui,c'est ça.

**Bktero** · 14/05/2018, 16h55

Pourquoi ne pas faire :
- T1 mets les caractères lus dans un stockage temporaire (au hasard, un std::ostringstream)
- Quand T1 rencontre '\n', il récupère une std::string de son stockage temporaire, pousse cette std::string dans une std::queue
- T1 remet à 0 son stockage et recommence sa quête d'un '\n'
- T2 attend qu'un élément soit disponible dans la std::queue
- T2 traite cet élément (une std::string)

Il faudra sûrement un std::mutex pour protéger les accès à la std::queue.

**lilocratch** · 14/05/2018, 17h22

Merci,
Je vais essayer ça quand je me serais familiarisé avec les queues et les ostringstreams.

**Bousk** · 14/05/2018, 18h21

Passer par un fichier est de loin la pire solution, et ton cas ne fait pas exception.
Surtout quand tout ce qui t'intéresse c'est des lignes entières.
Tu reçois les données, les pousses dans une structure intermédiaire. (ostringstream pourquoi pas, mais un bête vector est au moins aussi bien)
Puis tu parses cette structure pour vérifier les éventuels fins de ligne.
Pour chaque fin de ligne trouvée, tu split et pousses les données dans la structure partagée entre les threads.
Le temps de lock est minimal et tu tues pas tes perfs à faire de la lecture/écriture sur un fichier.

**lilocratch** · 16/05/2018, 15h50

Merci à tous,
Avec ce que j'ai retenu de vos commentaires j'ai fait le code ci-joint. Mais je n'ai pas très bien compris comment fonctionnait les queue et j'en aurai surement besoin si la fréquence des entrées augmente.
De plus il y a quelques bugs dans mon programme dont je ne parviens pas à identifier la source. (un très visible quand on lance le programme et l'autre est que le programme retourne parfois au début).
J’aurais donc encore besoin de quelques conseils.

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
#include <iostream>
#include <thread>
#include <windows.h>
#include <fstream>
#include <string>
#include <sstream>
#include <mutex>
#include <stdlib.h>
 
using namespace std;
 
int main()
{   mutex lock;
    string entry="",output,entry2="",output2;
    stringstream stream;
    stringstream stream2;
    int i,n,r;
    int h=0,m=0;
    float s=0;
    thread t1 ([&h,&s,&m,&entry,&entry2,&stream,&n,&stream2,&i,&output,&r,&lock](){
 
 
 
              while(1)
                {s+=0.1;
                if (s==60)
                    {s=0;m+=1;}
                if (m==60)
                    {m=0;h+=1;}
                stream<<h<<" "<<m<<" "<<s<<endl;
                entry=stream.str();
 
                r=5;
                entry2=entry.substr(r,n-1);
                entry=entry.substr(0,r-1);
                entry=entry2+entry;    //entry est juste une simulation d'une entrée que je pourais obtenir.
                n=entry.length();
                for(i=0;i<n;i++)
                    {
 
 
                     if(entry [i]=='\n')
                       {
 
                       lock.lock();
                       output=stream2.str();
                       lock.unlock();
                       stream2.str("");
                       }
                    else{stream2<<entry[i]; }
 
 
 
 
                    }
 
 
                Sleep(100);
 
 
 
              }
              });
    thread t2 ([&output,&output2,&lock](){
 
               while(1){
                lock.lock();
                output2=output,
                output="";
                lock.unlock();
                if(output2!="")
                   {
                    try{
                       cout<<"h= "<<output2.substr(0,output2.find(" "));
                       output2=output2.substr(output2.find(" ")+1);
                       cout<<" m= "<<output2.substr(0,output2.find(" "));
                       output2=output2.substr(output2.find(" ")+1);
                       cout<<" s= "<<output2.substr(0,output2.find(" "))<<endl;
                       }
                    catch(std::out_of_range& er){}
                    catch(std::invalid_argument& e){}
 
                       }
              }
              });
 
 
    t1.join();
    t2.join();
 
    return 0;
}

**lilocratch** · 18/05/2018, 11h29

J'ai essayé de corriger mon programme, mais il n'agit toujours pas comme ce que j'avais prévu. Quand j'affiche stream2.str() c'est correcte au début mais ou bout d'un moment il manque des charactéres. Et quand je lance mon programme tel qu'il est là , il n'y a aucune sortie alors que je m'attendais au moins à ce que output2 soit identique à stream2.str().
Si quelqu'un pouvais me donner un coup de main je le remercie d'avance.

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
#include <iostream>
#include <thread>
#include <windows.h>
#include <fstream>
#include <string>
#include <sstream>
#include <mutex>
#include <stdlib.h>
#include <queue>
 
using namespace std;
 
int main()
{   mutex lock;
    string entry="",entry2="",output2;
    stringstream stream;
    stringstream stream2;
    int i,n,r,position;
    int h=0,m=0;
    float s=0;
    queue <string> output;
    thread t1 ([&h,&s,&m,&position,&entry,&entry2,&stream,&n,&stream2,&i,&output,&r,&lock](){
 
 
 
              while(1)
                {s+=0.1;
                if (s==60)
                    {s=0;m+=1;}
                if (m==60)
                    {m=0;h+=1;}
                stream<<h<<" "<<m<<" "<<s<<endl;;
                entry=stream.str();
                stream.str("");
                //entry+="\n";
 
                n=entry.length();
 
 
 
                r=5+(int)(10*s)%4;
                entry=entry2+entry;
                entry2=entry.substr(r,n);
                entry=entry.substr(0,r);//entry est juste une simulation d'une entrée que je pourais obtenir.
                n=entry.length();
 
                for(i=0;i<n;i++)
                    {
 
 
                    stream2<<entry[i];
                    if(entry[i]=='\n')
                       {
                        lock.lock();
                        output.push(stream2.str());
                        lock.unlock();
                        stream2.str("");
 
                       }
 
 
                   }
 
 
 
                Sleep(100);
 
 
 
              }
              });
    thread t2 ([&output,&output2,&lock](){
 
               while(1){
                lock.lock();
                if(!output.empty()) output2=output.front(),
                lock.unlock();
                cout<<output2;
                if(output2!="")
                   {
                    try{
                       cout<<"h= "<<output2.substr(0,output2.find(" "));
                       output2=output2.substr(output2.find(" ")+1);
                       cout<<" m= "<<output2.substr(0,output2.find(" "));
                       output2=output2.substr(output2.find(" ")+1);
                       cout<<" s= "<<output2.substr(0,output2.find(" "))<<endl;
                       }
                    catch(std::out_of_range& er){}
                    catch(std::invalid_argument& e){}
 
                       }
              }
              });
 
 
    t1.join();
    t2.join();
 
    return 0;

**Bktero** · 18/05/2018, 17h33

Ton code est pas mal bordélique... Tu gagnerais à clarifier un peu tout ça. Ça passe par l'indentation, des lambdas qui n'ont pas 36 paramètres(*), des variables mieux nommées que entry1 / entry2...

(*) Soit tu utilises des captures par défaut (genre [&]), soit tu remarques que la plupart des variables peuvent être déclarés à l'intérieur de tes threads. Je t'ai dit plutôt qu'on n'était plus en C89, je le maintiens

Un code basique qui fait ce que tu veux, c'est ça :

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
#include <iostream>
#include <mutex>
#include <queue>
#include <sstream>
#include <thread>
 
int main(int, char*[]) {
 
	std::queue<std::string> queue;
	std::mutex mutex;
 
	// Producer
	std::thread producer([&]() {		
 
		auto produce = [&](const char* message) {
                        std::stringstream stream;
			stream << message;
			auto element = stream.str();
 
			mutex.lock();
			std::cout << "Pushing " << message << std::endl;
			queue.push(element);
			mutex.unlock();
		};
 
		produce("hello, world");
		produce("bye bye!");
		produce(":)");
	});
 
	// Consumer
	std::thread consumer([&]() {
		while (true) {
			if(not queue.empty()) {
				auto message = queue.front();
				std::cout << "Received = " << message << std::endl;
				queue.pop();}
		}
	});
 
	// Main
	producer.join();
	consumer.join();
}

Tu vois que les seules variables à créer dans le main sont la queue et le mutex. Toutes les autres devraient être à l'intérieur de tes threads.

En écrivant ce code, je me suis rendu compte que tu ne faisais pas de pop() après avoir fait front(). Si je ne m'abuse, tu n'enlèves donc jamais les éléments de ta queue (voir http://en.cppreference.com/w/cpp/container/queue/front).

CORRECTION : il manque une protection via mutex quand on accède à la queue dans 'consumer'. Le code devrait être ainsi :

Code :

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

1
2
3
4
5
6
7
8
9
10
11
12
	std::thread consumer([&]() {
		while (true) {
			if(not queue.empty()) {
				mutex.lock();
				auto message = queue.front();
				queue.pop();
				mutex.unlock();
 
				std::cout << "Received = " << message << std::endl;
			}
		}
	});

**lilocratch** · 21/05/2018, 10h50

Merci beaucoup je vais essayer avec ça. Mais ne faut il pas mettre un mutex dans le thread consumer?

**dalfab** · 21/05/2018, 14h54

Oui, il faut bien sûr protéger toutes les utilisations de queue.

**lilocratch** · 21/05/2018, 17h03

J'ai pas encore très bien compris comment fonctionnait les queue. Est-ce que les char* reste entier dans la queue ou sont ils sépares en chars. Si j'entre "un groupe\n un autre groupe" je vais avec push récupérer le même objet ou récupérer seulement "un groupe"?

**Bousk** · 21/05/2018, 17h15

Envoyé par lilocratch

Est-ce que les char* reste entier dans la queue ou sont ils sépares en chars.

Envoyé par lilocratch

Si j'entre "un groupe\n un autre groupe" je vais avec push récupérer le même objet ou récupérer seulement "un groupe"?

Pourquoi et par quelle magie aurais-tu quelque chose de différent ?! Tu ajoutes "nimporte quoi" dans la collection, tout ce qu'elle fait c'est garder ça en elle pour que tu puisses y accéder quand bon te semble..

**lilocratch** · 21/05/2018, 21h57

OK merci,
je pensais que la queue prenait les elements du tableau et les poussait jusqu'a \n.

**Bktero** · 22/05/2018, 10h04

Il manque effectivement un mutex lock/unlock lors qu'on souhaite récupérer des éléments de la queue, mea culpa !

J'aurais dû faire ça comme lambda en fait :

Code :

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

auto produce = [&](const std::string& message) { ... };

En C++, on évite de jouer avec les const char* si on n'a pas de bonnes raisons.

La queue ne sait rien d'un quelconque '\n'. C'est à toi de récupérer ton texte, de créer des std::strings et d'appeler produce() avec.

Pour répondre à la question de ton message supprimé (oui, les modérateurs ont des pouvoirs de vision

), tu n'as pas à t'occuper du type de 'produce'. En fait, tu ne pourrais même pas l'écrire, seul le compilateur y arrive. C'est un des cas magiques de l'utilisation de 'auto'. Si tu veux vraiment tout savoir, on peut réussir à faire cracher le type complet au compilateur et ça donne ceci : 'main(int, char**)::<lambda()>::<lambda(const string&)>'. Comme tu vois, 'auto', c'est très bien

**lilocratch** · 22/05/2018, 11h19

Merci à tous. J'ai réussit à faire mon programme et (pour l'instant) il marche correctement.

**koala01** · 23/05/2018, 22h02

Salut,

@Bktero : Par contre, pourquoi protéger les accès à la file

Parce que, en principe, c'est justement tout l'intérêt d'une file : on "push" à "back" et on "pop" à "top". Il n'y a donc théoriquement aucun risque de collision entre l'ajout et le retrait

écrire avec un thread et lire avec un autre

Threads & Processus C++

Discussions similaires

Partager

Partager