Discussion :

[bartinou]

Bonjour,

Comme le titre le suppose, je travaille actuellement sur un batch qui traite plusieurs millions d'objets en parallèle. J'aimerais que chacun d'entre eux puisse incrémenter différents compteurs (nb d'objets traités, nb d'erreurs de type 1, nb d'erreurs de type 2 etc.), et par conséquent de manière synchronisée.

Je ne suis que novice en matière d'environnement multithread

Je me suis donc orienté vers une classe qui utilise des Tables de hachage.

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
 
/**
 * Maintient des compteurs et fournit des méthodes pour
 * les incrémenter et les décrire.
 *
 */
public final class Compteur {
 
	/** Logger */
	private static final Logger LOGGER = Logger.getLogger(Compteur.class);
 
	/** L'instance singleton de Compteur */
	private static final Compteur SINGLETON = new Compteur();
 
	/** La table des compteurs */
	private final HashMap<String, Integer> mCompteurs;
 
	/** La table des descriptions de compteurs */
	private final HashMap<String, String> mDescriptions;
 
	/**
         * Constructeur privé
         */
	private Compteur(){
		super();
		mCompteurs = new HashMap<String, Integer>();
		mDescriptions = new HashMap<String, String>();
	}
 
	public static synchronized void increment(final String pNomCompteur){
		Integer lCount = SINGLETON.mCompteurs.get(pNomCompteur);
		if (lCount == null){
			SINGLETON.mCompteurs.put(pNomCompteur, 1);
		} else {
			SINGLETON.mCompteurs.put(pNomCompteur, lCount++);
		}
	}
 
}

Deux tables de hachage : l'une pour les compteurs, l'autre pour leur description, chacune identifiant les compteurs par une clé unique. Elles sont locales à une instance singleton de ma classe Compteur, initialisées par le ClassLoader.

J'aimerais porter votre attention sur la méthode d'incrémentation. Celle-ci s'effectue en deux temps : d'abord je récupère la valeur du compteur identifié par sa clé, puis je l'incrémente et remplace sa valeur dans la table des compteurs.

Je me suis orienté vers 2 implémentations, chacune soulevant des questions que j'aimerais vous soumettre :

• celle que j'ai publiée ci dessus, créer la table des compteurs à l'aide de new HashMap<String, Integer>(), donc sans synchronisation intrinsèque. La méthode increment() est donc synchronized. Les questions ici sont donc donc :
  
  1. est-ce une bonne approche ?
  2. étant donné que je parcours des millions d'objets, l'utilisation coûteuse de sychronized ne risque-t-elle pas de dégrader fortement les performances du batch ?
  
  
• une deuxième implémentation serait d'utiliser une ConcurrentHashMap pour la table des compteurs, ce qui synchroniserait la méthode d'écriture (SINGLETON.mCompteurs.replace(pNomCompteur, lCount++)). La méthode d'incrément ne serait donc plus synchronized. Malheureusement, l'accès à la valeur précédente du compteur ne serait pas synchronisée et donc je risquerais de me trouver avec plusieurs threads qui modifient la valeur du compteur sans prendre en compte sa nouvelle valeur. Je pencherais pour l'utilisation de volatile pour la table des compteurs, mais je ne suis pas sûr d'avoir compris son sens. Les questions ici sont donc donc :
  
  1. l'utilisation de volatile est-elle justifiée ici ?
  2. cette implémentation est-elle moins coûteuse que la précédente ?

Je vous remercie d'avance pour vos analyse, et si vous avez de meilleures solutions à me proposer n'hésitez pas à m'en faire part.

[tchize_]

tu va avoir de sacré pertes de perfs si à chaque opération tu fait un synchronisation et du travail sur les hashmap. Le mieux serait que chaque thread aie son propre set de compteur que tu incrémente. Ainsi pas de synchronisation.

Et quand tout est fini, tu fais la somme.

Un truc de ce style

Code :

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

1
2
3
4
5
6
7
8
9
10
11
public class Traitement extends Runnable {
    private int nbTraite = 0;
    private int nbErreur1 = 0;
    private int nbErreur2 = 0;
    public int getNbTraite(){
        return nbTraite;
    }
    // ailleurs dans "faire un traitement"
    nbTraite++;
    // ....
}

Et si tu veux afficher le total en temps réel:

Code :

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

1
2
3
4
5
6
7
8
9
public class Controlleur {
   private Traitement[] traitements;
   public int getNbTraiteTotal(){
      int total= 0;
      for (Traitement:traitements) 
          total+=traitement.getNbTraite();
      return total;
   }
}

Pas de synchro, pas de map.
Bien sur, ici, on peux se débarasser de la synchro pour deux raisons:
1) pas de risque de corruption car un seul et unique thread gère les modification
2) les lectures ne peuvent pas crasher à cause d'une modification en cours. Au pire on lira l'a valeur précédente si on viens de faire la modif, ce qui n'est pas un soucis quand il s'agit juste d'afficher une barre d'avancement. Et les valeurs seront correctes au final quand tous les threads auront fini le travail.

[bartinou]

Merci de ta réponse.

J'aurais dû préciser que je travaille dans un cadre particulier, où le parallélisme est géré par une librairie externe à mon projet.

En gros, le lancement du traitement à effectuer pour une liste d'objets est comme cela :

Code :

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

forAll(iteratorSurTousMesSuperObjets).perform(ClasseTraitement.class).start()

Ces classes ClasseTraitement sont spécifiques, leurs variables locales doivent être final, et on ne leur demande qu'une chose : avoir une méthode run(MonSuperObjet).

Le degré de parallélisme, la taille des lots etc. est configuré avant le lancement du batch (fichier de propriétés).

Du coup, je n'ai pas accès à la liste des threads, et par conséquent à aucune des méthodes publiques que j'aurais pu ajouter à ces ClasseTraitement.

Je ne vois donc pas comment je pourrais y maintenir des compteurs à jour... C'est pour cela que je me suis dirigé vers une classe Compteur.

[tchize_]

Je garderais la logique de plus haut, mais je la jouerais comme ça:


début du run: création ou récupération du dataset lié au thread courant, via ce genre de classe:

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
public class DataSet{
    public int nbErreurs;
    // etc comme plus haut
    private static List<DataSet> existant = Collections.synchronizedList(
          new ArrayList<DataSet>());
    private static ThreadLocal<DataSet> current = new ThreadLocal<DataSet>(){
         @Override protected DataSet initialValue() {
             // nouveau dataset pour le thread courant
             // on l'enregistre
             DataSet dataset = new DataSet();
             existant.add(dataset);
             return dataset;
         }
    }
    public static DataSet getCurrent(){
        return current.get();
    }
 
    public static int getTotalErreurs(){
        int total;
        synchronized(existant){
            for (DataSet data:existant)
              total += data.nbErreurs();
        }
    }
 
}

et dans ton run(MonSuperObjet):

Code :

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

1
2
3
4
5
6
7
 
DataSet dataset = DataSet.getCurrent();
dataset.nbTraite++;
//....
if (erreur)
  dataset.nbErreur++;
//.....

ThreadLocal a la particularité de ne nécessiter aucun synchronisation et de stocker une valeur dans le thread courant. Ainsi, la seul synchro nécessaire à ton batch et une fois par thread, lorsque threadlocal créera l'objet initial, ce qui nécessitera une synchro sur la liste gardant tous les dataset. Si t'as 100 threads pour un batch de 10.000 opérations, ca reste rentable avec 100 fois moins de synchros
Enfin, ton thread d'affichage lui fera plein de synchro, mais on a pas besoin de perfs par là en général, ce qui compte c'est la vitesse du batch