Mutithread sur meme methode

Salut a tous,
j'aurai besoin d'aide en Thread, je souhaite paralelliser mon programme pour que ça devienne plus rapide. Donc voila j'ai trois classe :
Classe 1 : classe main

Code:

---

1 2 3 4 5 6 7 8 9 10

W2VEstimator w2v_sim = new W2VEstimator(wordVectors); HashMap<Integer, Double> topic_sim_avg = new HashMap<Integer, Double>(); double scores = 1; for(Entry<Integer, HashSet<String>> entry : map_words.entrySet()) { Integer k = entry.getKey(); Double sim = w2v_sim.prob(word_m, entry.getValue()); topic_sim_avg.put(k, sim); scores += sim; }

---

La classe 2 : W2VEstimator qui a la méthode suivante

Code:

---

1 2 3 4 5 6 7 8 9 10 11

public double prob(String w_i, HashSet<String> words_t_i){ Similarity sim = new Similarity(wordVectors); double score_prob = 1; Iterator<String> it = words_t_i.iterator(); while (it.hasNext()) { score_prob += sim.computeSim(w_i, it.next()); } score_prob = score_prob/words_t_i.size();   return score_prob; }

---

en enfin la classe 3 : Similarity qui fait a la methode

Code:

---

1 2 3 4 5 6

public double computeSim(String w_1, String w_2){ double score_sim = 0.0; score_sim = ...... return score_sim; } }

---

Donc en gros , mon algo est le suivant :

Algorithm :
j'ai un ensemble K tel que chaque element de K est un ensemble qui contient des termes
Je veux calculer une similarité (une methode) entre un term t1 et les termes de chaque ensemble K
for each k in setK
for each elt in set k
res += compute(t1, elt)


les methodes que j'ai fait decrivent un peu ce process
et je souhaite le faire de façon parallellent donc un thread pour chaque setK et un thread pour chaque comput(t1,elt) et a la fin je somme les resultats
pour avoir pour K1, res, K2, res ..... ainsi de suite


J'ai essayé plein de truc mais ça me marche pas

merci beaucoup pour votre aide

donc en gros tu as n*n appels à la méthode compute, où n est la taille de ton set K, et tu veux rendre parallèles ces appels pour utiliser tout la puissance du CPU. Alors première remarque, c'est à la base une mauvaise bonne idée. A moins que n soit de l'ordre de 4 maximum ou que tu dispose d'une ferme de calcul. Un pc standard, c'est aujourd'hui à tout casser 8 coeurs, avec de l'hyper threading, ça te fait 16 unités de calculs parallèles. Si n=5, tu va balancer 25 threads dans 16 unités de calculs, tu va avoir déjà 9 threads qui vont essayer de foutre les autres dehors. Si on monte plus loin, avec 20 éléments, tu va créer 400 threads. Ton OS va tout doucement commencer à se demander ce que tu veux là, surtout que ce sont 400 threads qui bossent, il dorment pas. Avec 100 éléments, tu arrive à 10.000 threads, tu peux commencer à débrancher la prise, l'OS ne répond plus.


Cela étant dit, il existe des manières simple en java de paralléliser ton code. Par exemple, au lieu de faire

Code:

---

1 2 3 4 5 6 7

for(Entry<Integer, HashSet<String>> entry : map_words.entrySet()) { Integer k = entry.getKey(); Double sim = w2v_sim.prob(word_m, entry.getValue()); topic_sim_avg.put(k, sim); scores += sim; }

---

tu peux faire un truc du genre

Code:

---

1 2 3 4 5 6 7

scores = map_words.entrySet().parallelStream().mapToInt( (entry) -> { Double sim = w2v_sim.prob(word_m, entry.getValue()); topic_sim_avg.put(entry.getKey(), sim); return sim; } ).sum();

---

Bon, le parallelstream a ses inconvénient, tu peux aussi garder ta boucle et pour chaque tour de boucle, créer un Worker que tu balance dans un ThreadPoolExecutor que tu calibre en fonction de la machine.

Merci tchize_ pour ta reponse,
bien au fait, je vais lancer le programme sur un serveur de calcul qui un processur Bi Octo Core, et 256go de ram, est ce mieux et possible de paralleliser dans ce cas ?

merci

Bn encore une fois ça dépend de combien de thread tu peux faire en parallèle sur cette machine sans dégrader les performances et de ton calcul. Paralléliser pour optimiser en général on fait quand même ça après avoir fait les autres optimisation dans le code.

Salut,


On peut utiliser Runtime.getRuntime().availableProcessors() afin de déterminer le nombre de coeurs.

A noter que par défaut les Streams utilisent le pool commun, contenant n-1 coeurs, ce qui fait qu'avec le thread courant cela utilise autant de thread que de coeurs.
Si l'on souhaite utiliser moins de thread il faut créer explicitement un ForkJoinPool et y exécuter le stream :

Code:

---

1 2 3 4 5

ForkJoinPool forkJoinPool = new ForkJoinPool(3); // 3 threads forkJoinPool.submit(() -> { map_words.entrySet().parallelStream() ... });

---

A noter également que le parallélisme peut entrainer un surcoût qui peut être pénalisant lorsqu'on traite un petit nombre d'élément.
Il peut être utile de ne l'activer qu'à partir d'un certain nombre d'élément.




Sinon attention avec l'exemple de tchize_ , car topic_sim_avg est initialement créée en tant qu'HashMap.
Problème : HashMap n'est pas thread-safe et le fait d'y ajouter des éléments depuis différent thread amène à un comportement indéterminé (comme la perte de certains éléments) !

Bref il serait préférable d'utiliser une Map synchronisé (mais cela peut être couteux), ou d'utiliser collect(toMap()) qui gère cela proprement (en créant une Map par thread, puis en les regroupant à la fin) :

Code:

---

1 2 3 4

Map<Integer, Double> topic_sim_avg = map_words.entrySet().parallelStream() .collect(Collectors.toMap( Entry::getKey, (entry) -> w2v_sim.prob(word_m, entry.getValue()) ) );

---

Sauf que tu perds le scores total...

Maintenant si tu veux vraiment récupérer ce scores cela devient un peu plus compliqué car il faut écrire soit-même le collector et utiliser un type spécifique pour stocker les deux résultats :

Code:

---

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23

class Result { double scores = 0.0; HashMap<Integer, Double> map = new HashMap<>(); }   Result result = map_words.entrySet().parallelStream() .collect(Collector.of( // Le supplier : crée l'instance qui stockera le résultat (1 par thread) Result::new, // L'accumulator : fonction permettant d'ajouter un élément dans le résultat (r,entry) -> { double dim = w2v_sim.prob(word_m, entry.getValue()); r.scores += dim; r.map.put(entry.getKey(), dim); }, // Le combiner : appelé à la fin du travail de deux threads, afin de fusionner le résultat (r1, r2) -> { r1.scores += r2.scores; r1.map.putAll(r2.map); return r1; } // combiner ) );

---

a++