Práctica 6 — Monitores y consenso

Programas_P6/unam.fc.concurrent.practica6/src/module-info.java

/**
 * 
 */
/**
 * 
 */
module unam.fc.concurrent.practica6 {
}

Programas_P6/unam.fc.concurrent.practica6/src/unam/fc/concurrent/practica6/CASConsensus.java

package unam.fc.concurrent.practica6;
/*
 * Programa que implementa un protocolo de consenso con compareAndSet()y get()
 */
import java.util.concurrent.atomic.AtomicInteger;

public class CASConsensus<T> implements ConsensusProtocol<T>{
    private T[] propose;
    private final int capacity;
    private final int FIRST=-1;
    private AtomicInteger r = new AtomicInteger(FIRST);

    public CASConsensus(int c, T init){
        capacity = c;
        propose =  (T[]) new Object[capacity];
        for (int i = 0; i < capacity; i++) {
            propose[i] = init;
        }
    }

    public T decide(T value, int me){
        propose[me]=value;
        if (r.compareAndSet(FIRST, me)) {
            //System.err.println("WIN " + me);
            return propose[me];
        } else {
            //System.err.println("LOSE " + me);
            return propose[r.get()];
        }
    }

}

Programas_P6/unam.fc.concurrent.practica6/src/unam/fc/concurrent/practica6/ConsensusProtocol.java

package unam.fc.concurrent.practica6;

/*
 * Todo protocolo de consenso tiene un metodo decide(value) que propone un value y devuelve el valor ganador
 */
public interface ConsensusProtocol<T> {
    public T decide(T value, int me);
}

Programas_P6/unam.fc.concurrent.practica6/src/unam/fc/concurrent/practica6/ExecConsenRounds.java

package unam.fc.concurrent.practica6;
/*
 * Programa que utiliza el programa de CountDownLatch para ejecutar varias rondas de consenso
 * El objeto de consenso es una instancia de ConsensusProtocol
 */
import java.util.Arrays;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;



public class ExecConsenRounds {
    public static int[] winners;
    public static void task(CountDownLatch latch, CASConsensus<Integer> cas, int c, int round){
        Thread thread = Thread.currentThread();
        long me = thread.getId();
        int id = (int) (me % c);
        int winner = cas.decide(id, id);//Se ejecuta el protocolo de consenso, se devuelve al ganador
        winners[round] = winner;// se guarda al ganador en el arreglo
        System.out.println("Thread: "+ id + " says WIN: " + winner);
        latch.countDown(); // Una vez que terminas, disminuyes el contador de "latch"
    }
    public static void main(String[] args) {
        int c = 4, rounds = 10; // c es el no. de hilos, rounds es el no. de rondas
        winners = new int[rounds];
        ExecutorService executor = Executors.newFixedThreadPool(c);//El candado solo funciona para dos hilos
        
        
        for (int j = 0; j < rounds; j++) {//Iteras en el no. de rondas
            CountDownLatch latch =  new CountDownLatch(c);
            CASConsensus<Integer> protocolCAS = new CASConsensus<Integer>(c, -1);
            int currRound = j;
            System.out.println("Round: " + j);
            for (int i = 0; i < c; i++) { //creas "c" tareas, una por cada hilo
                executor.execute(() -> task(latch, protocolCAS, c, currRound)); 
            }
            try {
                latch.await(); // El hilo main espera a que los "c" hilos terminen
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
            
        }

        System.out.println("\n Winners: " + Arrays.toString(winners)); 

        executor.shutdown();
        
        //while(!executor.isTerminated()) {};
        
    }
}