Discussion :

[divxdede]

Bonjour,

Je suis en train d'écrire une librairie permettant d'échanger des objets java entre deux programmes (utilisant nio). Dans l'ensemble la librairire arrive à la fin de son cycle de développement mais il me reste une synchronisation à m'occuper.

Voici le contexte: admettons les deux méthodes suivantes

Code :

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

1
2
3
4
5
6
7
8
 
   /** Arrette le service courant 
     */
   public void stop();
 
   /** Emet un objet sur le canal spécifié
    */
   public void write(Object o,Channel c);

La méthode stop() arrete le service et rends la main une fois complètement arreté.
1. Il est illégal d'appeler une méthode write() pendant la phase d'arrêt
2. La phase d'arret s'assure qu'aucun thread n'est en cours de write avant de s'executer (laisse se complèter les tâches en cours)

A la vue de la sémantique de ces deux méthodes, l'objectif est d'essayer d'avoir le moins de contention dans la méthode write() (car elle est appelées souvents) pour gérer un cas qui arrive peu fréquemment(l'arret du service)

1ere solution:

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
 
    private Object LOCK = new Object();
    private boolean running = true;
    private int  op = 0;
 
    public void stop() {
        synchronized(LOCK) {
            this.running = false; 
            while( op > 0 )  {
                try {
                    LOCK.wait();
                }
                catch(Exception e) { /* do nothing */ }
            }
        }
 
        // ******
        // DO JOB
        // ******
    }
 
    public void write(...) {
 
        synchronized(LOCK) {
            if( ! this.running ) throw new IllegalStateException("service unavailable");
            op++;
        }   
 
        // ******
        // DO JOB
        // ******
 
        synchronized(LOCK) {
            op--;
            if( !this.running && op == 0 ) LOCK.notifyAll();
        }
    }

Cette solution est la plus simple mais la méthode write() acquiert deux fois le même moniteur uniquement pour gérer la synchronisation avec la méthode stop() ce qui me semble assez inefficace.

La deuxieme solution tends a rendre moins efficace la méthode stop() au profit de la méthode write()

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
 
    private static final int THRESHOLD = 1000000;
    private AtomicInteger op = new AtomicInteger(1);
 
    public void stop() {
        synchronized(this) {
 
            op.addAndGet(THRESHOLD);
            while( op.get() > THRESHOLD )  {
                try {
                    this.wait(100);
                }
                catch(Exception e) { /* do nothing */ }
            }
        }
 
        // ******
        // DO JOB
        // ******
    }
 
    public void write(Object o) {
 
        int value = op.incrementAndGet();
        if( value >= THRESHOLD ) throw new IllegalStateException("service unavailable");
 
        // ******
        // DO JOB
        // ******
 
        op.decrementAndGet();
    }

La deuxieme solution permet à la méthode write() de ne pas effectuer de synchronisation et de ne pas créer un goulet d'étranglement entre les differents threads effectuant des write(). La contre-partie est de ne pas disposer de notification dans la méthode stop() et donc de devoir effectuer des wait() a durée limitée pour re-tester l'état.

Ma Question:
En fait il s'agit plutôt d'une discussion ouverte quand à la solution à adopter !!La solution 2 est elle juste et surtout plus efficace que la 1ère ?
Quelle est votre opinon, avez vous une meilleure solution ?

[gifffftane]

J'imagine qu'il existe un ou plusieurs threads dédiés à l'écriture ?... en ce cas, il suffirait de les repérer, puis d'attendre leur fin par join() ou join(long millis).

[divxdede]

J'imagine qu'il existe un ou plusieurs threads dédiés à l'écriture ?... en ce cas, il suffirait de les repérer, puis d'attendre leur fin par join() ou join(long millis).

Pas tout à fait.
La librairire fonctionne en mode asynchrone, donc oui les threads réalisant effectivement l'écriture sont determinés et determinables mais la méthode write() ci-dessus permet uniquement de mettre en queue une demande d'écriture et celle-ci est appelée de n'importe quel thread "business".

En réalité cette méthode ne fait que deux choses:
1. Serializer l'objet java dans le thread appelant (pour fixer l'image de l'objet)
2. Mettre en queue le bitblob à écrire sur le canal précis

Ces deux tâches gérent déja leur propres accés concurrent (un pool de serializer pour la 1. et une queue spéciale pour le 2.

Il est a noter que la méthode stop() attends déja la fin effective des threads d'execution.

[gifffftane]

OK...

Je comprends qu'il y a deux niveaux :

1) dans le niveau business, on sérialize l'objet à transmettre et on met en file d'attente la transmission.
2) dans le niveau transmission, un thread à part, on... transmet en dépilant la file d'attente.

À partir du moment où le niveau 1 est commencé, il faut attendre la fin du niveau 2 pour faire le stop.

Pour le 2, on sait déjà attendre, mais pas pour le niveau 1.

Ce qui m'étonne avec ce niveau 1, est que je n'ai pas l'impression que les étapes soient très consommatrices de temps ?... Sérializer un objet va assez vite, et je n'imagine pas, au milieu de l'opération, aller vérifier un booléen pour savoir s'il faut continuer ?... Quand à mettre un objet sérialiser en file d'attente... Le jeu en vaut-il la chandelle ?

Sur tes deux solutions, je verrais plutot la première, moi la deuxième utilise deux verrous (avec le AtomicInteger), j'aime moins.

Ce qui m'étonne un peu dans tes solutions, est que, à ce que je comprends, il faut couvrir deux étapes ; comment fais-tu pour aller jusqu'à la deuxième ?

[divxdede]

Le niveau de transmission est plus délicat que ca.
En réalité l'écriture est également gérée de manière asynchrone. Ce qui veut dire que lorsque j'ai une écriture dans la file d'attente, la librairie souscrit à une notification OP_WRITE.
C'est a dire qu'un thread dédié (aux evenements asynchrones) est prevenus lorsqu'un canal est disponible afin d'emmettre des données (buffer du socket non plein).
Schematiquement à ce moment là, une tache d'écriture est déléguée a un pool de threads gérés par un ExecutorService.

Dans la méthode stop() il faut finalement s'assurer de plusieurs choses:
1. Ne plus accepter de nouveaux write()
2. Laisser se terminer les write() en cours
3. Laisser se vider la queue d'écriture

Aujourd'hui sur ce point mon implémentation actuelle est fausse puisque dans la méthode stop() j'appelle un ExecutorService.shutdown()
Ce qui me permet effectivement d'arretter mes threads d'executions tout en les laissants terminer les tâches en cours.
Mais cet arret ne s'interresse ni de pret, ni de loin à l'état de ma file d'attente d'écriture.

Pour compliquer un peu l'architecture, la phase de sérialisation utilise un pool de Serializer ce qui permet principalement de ne pas recréer un ObjectOuptutStream (et son ByteArrayOutputStream) a chaque appel d'une méthode write().
Mais la contre partie est qu'il est nécessaire dans la méthode stop() de libérer le pool et ainsi de fermer les ressources associées aux differents serializers (les ObjectOutputStreams & co).

C'est pourquoi la méthode stop() doit attendre que tout les sérializers soient retournés dans leur pool et empecher que de nouveaux soient demandés

Il me semble que la synchronisation dont on discute ne doit pas s'occuper de gerer du vidage de la file d'attente mais uniquement gérer la synchro sur les appels à la méthode write() uniquement.
Il est par contre évident que la méthode stop() devra en sus se synchroniser sur la fin de la file d'attente mais c'est une autre histoire


Pour revenir sur le AtomicInteger, j'ai commencé a regarder l'implementation est il me semble au contraire qu'il ne pose pas de verrou strict.
Il utilise une propriété volatile (dont potentiellement moins performante), et gére l'atomicité par une iteration jusqu'a réussite.
Mais je ne connais pas ce que fait le sun.misc.Unsafe puisqu'il s'agit d'une boite noire trés.....obscure.

La 1ere solution prends deux moniteurs, ce qui me semble tout autant pénalisant.
Je dois dire que je n'ai pas assez de feedback sur les AtomicXXXXX pour être sur que la 2ème solution soit plus performante.

[gifffftane]

Je pense que, pour empêcher de nouveaux write après un stop, un booléen suffit.

Pour le reste... ne serait-il pas possible, au lieu de suivre l'ensemble des processus, d'émettre un événement interne à chaque début de write, et un autre à chaque fin, tous processus, pools, threads, et toutes couches comprises ?

De cette façon, ton stop attendrait simplement que tous les évènements de fin arrivent, au lieu de se préoccuper de vérifier que tous les intermédiaires ont fini ?...

[divxdede]

Je ne saisis pas bien l'architecture d'évenements que tu proposes, cependant tu viens de mettre le doigt sur un point important !!!

Il suffit de bloquer l'accés à la méthode write(), ensuite la méthode stop() n'a pas s'occuper des appels concurrents sur cette méthode mais uniquement attendre le vidage complet de la file d'attente (qui ne peut plus recevoir de nouvelle demande).

De cette façon, un boolean suffit effectivement dans la méthode write().

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
 
    private AtomicBoolean running = new AtomicBoolean(true);
 
    public void stop() {
 
            this.running.set(false);
 
            // ***********************************************************
            // Attendre le vidage complet de la liste d'attente d'écriture
            // ***********************************************************
            synchronized( this) {
                while( ! queue.isEmpty() ) {
                    try {
                        this.wait( 100 );
                    }
                    catch(Exception e) { /* do nothing */ }
                }
            }
 
            // ******
            // DO JOB
            // ******
    }
 
    public void write(Object o) {
 
            if( ! this.running.get() ) throw new IllegalStateException("service unavailable");
 
            // ******
            // DO JOB
            // ******
    }

En tout cas merci beaucoup de participer à mes reflections

Le fait d'attendre par "iteration" le vidage de la queue permet d'éviter tout code d'evenements (genre LOCK.notifyAll() ) du coté de la queue. La relative ineficacité de la méthode stop() est raisonnable et même préferable au profits des autres process.