Um thread parando muito cedo, independentemente do CyclicBarrier

Question

Estou ciente do fato de que o código a seguir pode parecer vulgar, mas sou novo nessas coisas e tentei de tudo para fazê-lo funcionar.

Problema:Mesmo que eu esteja usando (possivelmente de maneira errada) um CyclicBarrier, um - e parece ser sempre o mesmo - thread para muito cedo e imprime seu vetor, deixando 1 de 11 dessas mensagens de "Conexão de entrada" ausentes.Provavelmente há algo terrivelmente errado com a última iteração do meu loop, mas não consigo encontrar exatamente o que.Agora o programa apenas fica em loop esperando para processar a última conexão.

public class VectorClockClient implements Runnable {
/*
 * Attributes
 */

/*
 * The client number is store to provide fast
 * array access when, for example, a thread's own
 * clock simply needs to be incremented.
 */
private int clientNumber;
private File configFile, inputFile;
int[] vectorClock;

/*
 * Constructor
 * @param
 * - File config
 * - int line
 * - File input
 * - int clients
 */
public VectorClockClient(File config, int line, File input, int clients) {
    /*
     * Make sure that File handles aren't null and that
     * the line number is valid.
     */
    if (config != null && line >= 0 && input != null) {
        configFile = config;
        inputFile = input;
        clientNumber = line;
        /*
         * Set the array size to the number of lines found in the
         * config file and initialize with zero values.
         */
        vectorClock = new int[clients];
        for (int i = 0; i < vectorClock.length; i++) {
            vectorClock[i] = 0;
        }
    }
}

private int parsePort() {
    int returnable = 0;
    try {
        FileInputStream fstream = new FileInputStream(configFile.getName());
        DataInputStream in = new DataInputStream(fstream);
        BufferedReader br = new BufferedReader(new InputStreamReader(in));
        String strLine = "";
        for (int i = 0; i < clientNumber + 1; i++) {
            strLine = br.readLine();
        }
        String[] tokens = strLine.split(" ");
        returnable = Integer.parseInt(tokens[1]);
    }
    catch (Exception e) {
        e.printStackTrace();
    }
    System.out.println("[" + clientNumber + "] returned with " + returnable + ".");
    return returnable;
}

private int parsePort(int client) {
    int returnable = 0;
    try {
        FileInputStream fstream = new FileInputStream(configFile.getName());
        DataInputStream in = new DataInputStream(fstream);
        BufferedReader br = new BufferedReader(new InputStreamReader(in));
        String strLine = "";
        for (int i = 0; i < client; i++) {
            strLine = br.readLine();
        }
        String[] tokens = strLine.split(" ");
        returnable = Integer.parseInt(tokens[1]);
    }
    catch (Exception e) {
        e.printStackTrace();
    }
    return returnable;
}

private int parseAction(String s) {
    int returnable = -1;
    try {
        FileInputStream fstream = new FileInputStream(configFile.getName());
        DataInputStream in = new DataInputStream(fstream);
        BufferedReader br = new BufferedReader(new InputStreamReader(in));
        String[] tokens = s.split(" ");
        if (!(Integer.parseInt(tokens[0]) == this.clientNumber + 1)) {
            return -1;
        }
        else {
            if (tokens[1].equals("L")) {
                vectorClock[clientNumber] += Integer.parseInt(tokens[2]);
            }
            else {
                returnable = Integer.parseInt(tokens[2]);
            }
        }
    }
    catch (Exception e) {
        e.printStackTrace();
    }
    return returnable;
}

/*
 * Do the actual work.
 */
public void run() {
    try {
        InitClients.barrier.await();
    }
    catch (Exception e) {
        System.out.println(e);
    }
    int port = parsePort();
    String hostname = "localhost";
    String strLine;
    ServerSocketChannel ssc;
    SocketChannel sc;
    FileInputStream fstream;
    DataInputStream in;
    BufferedReader br;
    boolean eof = false;
    try {
        ssc = ServerSocketChannel.open();
        ssc.socket().bind(new InetSocketAddress(hostname, port));
        ssc.configureBlocking(false);
        fstream = new FileInputStream("input_vector.txt");
        in = new DataInputStream(fstream);
        br = new BufferedReader(new InputStreamReader(in));

        try {
            InitClients.barrier.await();
        }
        catch (Exception e) {
            System.out.println(e);
        }

        while (true && (eof == false)) {
            sc = ssc.accept();

            if (sc == null) {
                if ((strLine = br.readLine()) != null) {
                    int result = parseAction(strLine);
                    if (result >= 0) {
                        //System.out.println("[" + (clientNumber + 1)
                        //+ "] Send a message to " + result + ".");
                        try {
                            SocketChannel client = SocketChannel.open();
                            client.configureBlocking(true);
                            client.connect(
                                    new InetSocketAddress("localhost",
                                    parsePort(result)));
                            //ByteBuffer buf = ByteBuffer.allocateDirect(32);
                            //buf.put((byte)0xFF);
                            //buf.flip();
                            //vectorClock[clientNumber] += 1;
                            //int numBytesWritten = client.write(buf);
                            String obj = Integer.toString(clientNumber+1);
                            ObjectOutputStream oos = new 
                                    ObjectOutputStream(
                                    client.socket().getOutputStream());
                            oos.writeObject(obj);
                            oos.close();
                        }
                        catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }
                else {
                    eof = true;
                }
            }
            else {
                ObjectInputStream ois = new 
                        ObjectInputStream(sc.socket().getInputStream());
                String clientNumberString = (String)ois.readObject();
                System.out.println("At {Client[" + (clientNumber + 1)
                        + "]}Incoming connection from: "
                        + sc.socket().getRemoteSocketAddress()
                        + " from {Client[" + clientNumberString + "]}");
                sc.close();
            }
            try {
                InitClients.barrier.await();
            }
            catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
    catch (Exception e) {
        e.printStackTrace();
    }
    printVector();
}

private void printVector() {
    System.out.print("{Client[" + (clientNumber + 1) + "]}{");
    for (int i = 0; i < vectorClock.length; i++) {
        System.out.print(vectorClock[i] + "\t");
    }
    System.out.println("}");
}

}
Para esclarecer, aqui estão os formatos dos arquivos utilizados.Config contém nomes de host e portas usadas por clientes que são threads e linhas do arquivo de entrada significam "este cliente envia uma mensagem para aquele cliente" ou "este cliente incrementa seu relógio lógico em algum valor constante".

1 M 2 (M significa enviar uma mensagem)
2m3
3 M 4
2 L 7 (L significa incrementar o relógio)
2 M 1
...
127.0.0.1 9000
127.0.0.1 9001
127.0.0.1 9002
127.0.0.1 9003
...

Answer 1

Eu examinaria a lógica relacionada a quando você espera uma conexão de soquete de entrada.Pela sua pergunta, parece que você espera um certo número de conexões de soquete de entrada (potencialmente uma conexão de entrada após cada mensagem enviada?).Como você está usando E/S sem bloqueio no soquete de entrada, é sempre possível que sua instrução while faça um loop antes que um soquete de entrada possa ser estabelecido.Como resultado, um thread seria capaz de continuar e ler a próxima linha do arquivo sem receber uma conexão.Como o seu estado final é alcançado quando o final do arquivo é alcançado, é possível que você perca uma conexão de soquete de entrada.

Eu acrescentaria algumas impressões simples que são exibidas quando você lê o arquivo, quando envia uma mensagem e quando recebe uma conexão de entrada.Isso deve informar rapidamente se um determinado thread está faltando ou não em uma conexão de entrada esperada.Se descobrir que o problema é devido à E/S sem bloqueio, talvez seja necessário desabilitar a E/S sem bloqueio quando você espera um soquete de entrada ou implementar um controle que monitore quantos soquetes de entrada você espera e continua até que esse objetivo seja alcançado.

Espero que isto ajude.