11. Sémaphores, barrières et latches

11.1. Introduction

L'API Concurrent introduit enfin des verrous explicites, qui viennent enrichir les possibilités de ce que l'on pouvait faire jusqu'en Java 4. En fait, jusqu'à cette version de Java, du moins avec les éléments standard qui nous étaient fournis dans le JDK, on n'avait guère que le mot-clé synchronized pour synchroniser des blocs d'instruction.

11.2. Notion de sémaphore, classe `Semaphore`

Un sémaphore est un objet qui contrôle l'accès à certaines ressources. Un thread qui souhaite accéder à cette ressource doit demander une autorisation à ce sémaphore, qui peut accorder l'accès ou pas. En Java, les sémaphores possèdent un certain nombre d'autorisations, fixé à la construction du sémaphore. Ce nombre marque la limite du nombre de threads qui peuvent accéder simultanément aux ressources gardées par ce sémaphore. Vu de cette façon, un sémaphore ressemble à un Lock, mais ces deux notions n'ont en fait rien à voir.

Quand un thread prend un verrou, il en devient le propriétaire, et c'est à lui de le rendre. Un sémaphore ne retient pas quel thread a obtenu une autorisation, d'ailleurs cette notion d'autorisation n'est associée à aucun objet. Un sémaphore sait juste qu'il a autorisé un certain nombre de threads à utiliser la ressource qu'il garde, sans avoir retenu lesquels. Un sémaphore gère un compteur, sans retenir qui a demandé une des autorisations qu'il gère.
La sémantique du Lock vis à vis de la synchronisation des opérations qu'il garde est la même que celle du bloc synchronized. Lorsqu'un thread sort d'un bloc d'instructions gardé par un Lock, toutes les variables qu'il a modifiées sont publiées auprès de tous les autres threads qui les connaissent. Il n'en est rien pour un sémaphore.

La classe Semaphore expose deux types de méthodes : celles qui permettent de gérer les autorisations, et celles qui permettent d'interroger combien de threads sont éventuellement en attente d'une autorisation.

acquire() et acquire(int permits) : demande une ou plusieurs autorisations. Ces deux méthodes sont bloquantes : elles ne rendent la main que lorsque le nombre d'autorisations demandé est disponible. Elles jettent une exception InterruptedException si le thread est interrompu.
acquireUninterruptibly(), acquireUninterruptibly(int permits) : ces deux méthodes sont analogues aux précédentes, sauf qu'elles ne jettent pas d'exception. Si le thread est interrompu, il continue à attendre une autorisation. Lorsque cette méthode rend la main, le statut interrupted du thread est à true.
tryAcquire() et tryAcquire(int permits) : demande une ou plusieurs autorisations. Ces deux méthodes rendent immédiatement la main, et retournent true ou false, suivant que le bon nombre de permis a été octroyé.
tryAcquire(long timeout, TimeUnit unit) et tryAcquire(int permits, long timeout, TimeUnit unit) : mêmes méthodes que les précédentes. Au lieu de rendre la main immédiatement, elles peuvent attendre le temps passé en paramètre, que des permis se libèrent.
release() et release(int permits) : rend une autorisation, ou le nombre d'autorisations passées en paramètre.
availablePermits() : retourne le nombre d'autorisations disponibles pour ce sémaphore.
drainPermits() : demande tous les autorisations restant disponibles pour ce sémaphore. Le nombre d'autorisations acquis est retourné par cette méthode.
hasQueuedThreads() et getQueueLength() : retourne true si des threads sont en attente d'autorisations de la part de ce sémaphore, et une estimation du nombre de threads en attente.

On peut utiliser un sémaphore pour limiter le nombre d'éléments dans une collection synchronisée.

Exemple 87. Sémaphore pour limiter la taille d'une collection

public  class CollectionTailleMax<T> {

    // création de la collection et d'un sémaphore
    private Collection<T> collection ;
    private Semaphore semaphore ;
   
    public CollectionTailleMax(int tailleMax) {
      
       // initialisation d'un collection synchronisée
       this.collection = Collections.synchronizedCollection(new ArrayList<T>(tailleMax)) ;
       this.semaphore =  new Semaphore(tailleMax) ;
   }
   
    public  boolean add(T t)  throws InterruptedException {
      
       // ajout d'un objet, demande d'une autorisation auprès du sémaphore
       // cette méthode est bloquante, s'il n'y a pas d'autorisation disponible
       // on attend...
      semaphore.acquire() ;
      collection.add(t) ;
      
       return true ;
   }
   
    public  boolean remove(Object o)  throws InterruptedException {
      
       // on commence par vérifier que l'objet passé est
       // bien retiré de la collection
       boolean removed = collection.remove(o) ;
       if (removed) {
          // dans ce cas on rend une autorisation au sémaphore
         semaphore.release() ;
      }
      
       return removed ;
   }
}

11.3. Notion de latch, classe `CountDownLatch`

Le concept de latch est très simple. Il s'agit d'une porte, qui est construite dans l'état fermé, et que l'on peut ouvrir. Une fois ouverte, elle laisse passer tous les threads sans contrôle, mais tant qu'elle est fermée, tous sont en attente. Un latch est construit sur un nombre, qui joue le rôle de compteur. Ce compteur est décrémenté sur demande (méthode countDown()). Quand il arrive à 0, le latch s'ouvre. On ne peut pas réinitialiser, c'est-à-dire fermer, un latch qui a été ouvert. Un tel concept peut être utile au démarrage d'une application. Lors de ce démarrage, un certain nombre de ressources, ou d'accès à des ressources, est initialisé, dans des threads différents. Puis les threads applicatifs se lancent, et doivent attendre la fin des threads d'initialisation. Typiquement, un latch peut enregistrer la fin des threads d'initialisation, et s'ouvrir quand l'initialisation globale est terminée. Les threads applicatifs, qui sont en attente de l'ouverture de ce latch peuvent alors commencer leur travail normalement, certains que toutes les initialisations se sont déroulées correctement. La classe CountDownLatch est une classe très simple, qui n'expose que quatre méthodes (sans compter la surcharge de toString()).

countDown() : permet de décrémenter l'entier sur lequel ce latch est construit. Quand ce compteur interne arrive à zéro le latch s'ouvre.
await() et await(long timeout, TimeUnit unit) : permet d'attendre l'ouverture de ce latch , avec éventuellement un temps maximal.
getCount() : retourne la valeur du compteur interne de ce latch .

11.4. Notion de barrière, classe `CyclicBarrier`

Une barrière est un concept proche du latch . Avec un latch , un appel explicite est fait pour décrémenter le compteur interne, et lorsqu'il atteint la valeur zéro, tous les threads en attente sont lancés. Le nombre de ces threads n'est pas limité, et ce ne sont pas ces threads (du moins a priori ) qui appellent la méthode countDown(). Une barrière est également construite sur un compteur, mais ce compteur ne se décrémente pas. Des threads arrivent à cette barrière et appellent sa méthode await(). Cela indique à la barrière qu'ils sont attente dessus. Tant que la barrière ne s'ouvre pas, cette méthode await() ne rend pas la main. Durant ce temps, la barrière compte le nombre de threads en attente, et s'ouvre quand ce nombre à atteint la valeur de son compteur. À ce moment les méthodes await() rendent la main et les threads en attente peuvent continuer leurs traitements. On peut de plus refermer cette barrière, par appel à sa méthode reset(), ce qui n'est pas possible avec un latch . La classe CyclicBarrier est également assez simple, et n'expose que six méthodes.

await() et await(long timeout, TimeUnit unit) : permettent à un thread de se mettre en attente sur cette barrière. La barrière compte ces threads en attente, et lorsque le bon nombre est atteint, elle ouvre la barrière. Ces méthodes rendent alors la main, et les threads peuvent continuer leur travail.
getParties() : retourne le nombre de thread qui doivent être en attente pour que cette barrière s'ouvre.
getNumberWaiting() : retourne le nombre de threads en attente sur cette barrière.
reset() : réinitialise cette barrière.
isBroken() : retourne true si cette barrière est dans un état corrompu.

Une barrière peut se trouver dans un état corrompu, examinons ce point de plus près. Quand un thread fait un appel à la méthode await() d'une barrière, il est mis en sommeil, et la machine Java ne lui passera plus la main, du moins tant que la barrière ne s'est pas ouverte. Si le statut interrupted de ce thread est true lorsqu'il appelle cette méthode, ou, alors qu'il a appelé cette méthode, passe à true avant qu'elle n'ait rendu la main, alors une exception InterruptedException est jetée. Si un appel à reset() est fait sur cette barrière, tous les threads en attente jettent une exception de type BrokenBarrierException. Si un des threads en attente est interrompu (appel à sa méthode interrupt()), alors tous les autres threads en attente jettent une exception de type BrokenBarrierException. Si un des threads a attendu trop longtemps par rapport au temps maximal d'attente passé en paramètre, alors il jette une exception de type TimeoutException. Dans tous les cas où un des threads qui étaient en attente jette une exception, la barrière passe en état corrompu, et sa méthode isBroken() retourne true. Tous les autres threads jettent alors une exception de type BrokenBarrierException.

Java API avancées

Retour au blog Java le soir

Cours & Tutoriaux