Cyclic barrier reuse?

Question

I'm currently trying to implement an application to communicate with a machine, and it should essentially work as follows:

• The program sends a message to the server (in this case, the first 255 bytes of a file).
• The machine responds with a "message successfully received" or "error receiving message" response.
• The program then has to decide whether to send the next message (next 255 bytes) or not (error on the last message, must start over) depending on the machine's response.
• And so on for each message the program needs to send (depends on the size of the file).

So, we thought of having a thread to do the sending, and another to do the receiving, since we have an api to register a class as the one to receive messages from the machine (just by implementing an interface), and the methods to send messages to the machine are not of the blocking type, so a way to wait for the machine's response so the program can decide what to do after the response arrives is necessary.

So, we needed to synchronize these two threads somehow, since it's possible to determine how many messages they would exchange, which made us try a CyclicBarrier. This is the code for testing whether a CyclicBarrier would help us solve this problem (the program doesn't actually use sockets to communicate with the machine, this is just for testing the barrier):

import java.io.*;
import java.net.*;
import java.util.concurrent.*;

public class BlockingTest{
  private CyclicBarrier barrier;

  class Receiver implements Runnable{
    @Override public void run(){
      try{
        ServerSocket ss = new ServerSocket(8080);
        while(!barrier.isBroken()){
          System.out.println("Waiting message...");
          Socket response = ss.accept();
          BufferedReader br = new BufferedReader(new InputStreamReader(
              response.getInputStream()));
          System.out.printf("Received: %s\n", br.readLine());
          barrier.await();
        }
      }catch(InterruptedException | BrokenBarrierException |
        IOException ex){
        System.err.println(ex.getMessage());
      }
    }
  }

  public BlockingTest(){
    this.barrier = new CyclicBarrier(2, new Runnable(){
      @Override public void run(){
        System.out.println("done.");
      }
    });

    new Thread(new Receiver()).start();

    try{
      Socket sender = new Socket("localhost", 8080);
      PrintWriter pw = new PrintWriter(sender.getOutputStream(), true);
      for(int i = 0; i < 3; i++){
        System.out.println("Sending message:");
        pw.println("Message!");
        this.barrier.await();
      }
    }catch(InterruptedException | BrokenBarrierException | IOException ex){
      System.err.println(ex.getMessage());
    }
  }

  public static void main(String[] arg){
    new BlockingTest();
  }
}

This code works as expected if we only send one message (no for block in the BlockingTest() constructor, just send the message), but after adding the for block, it doesn't work as expected. It only works the first time, then it hangs:

Waiting message...
Sending message:
Received: Message!
done.
Waiting message...
Sending message:

The questions are:

• How to make the barrier reusable? is that automatic or does it have to be done by hand?

• Is the program hanging due to something i missed with the sockets (or the barrier code)?

Answer 1

The reason it hangs is that you're opening one connection to the server and keep sending data to it, but in the receiving end, you're discarding the first connection, and start waiting for the next connection to happen.

Either you can create a new connection from the sender every time you send data. The code block

Socket sender = new Socket("localhost", 8080);
PrintWriter pw = new PrintWriter(sender.getOutputStream(), true);

has to be moved inside the for loop. (of course, take care to release all resources)

OR,

fix the receiver to read data from the first connection, instead of waiting for a new connection for the second pack of data.