Java: Threads, how to make them all do something

I am trying to implement nodes talking to each other in Java. I am doing this by creating a new thread for every node that wants to talk to the server.

When the given number of nodes, i.e. that many threads have been created, have connected to the server I want each thread to execute their next bit of code after adding to the "sharedCounter".

I think I need to use 'locks' on the shared variable, and something like signalAll() or notifyAll() to get all the threads going, but I can't seem to make clear sense of exactly how this works or to implement it.

Any help explaining these Java concepts would be greatly appreciated :D

Below is roughly the structure of my code:

import java.net.*;
import java.io.*;

public class Node {

    public static void main(String[] args) {
    ...
    // Chooses server or client launchers depend on parameters.
    ...
    }
}

class sharedResource {
    private int sharedCounter;

    public sharedResource(int i) {
        sharedCounter = i;
    }

    public synchroniz开发者_开发知识库ed void incSharedCounter() {
        sharedCounter--;
        if (sharedCounter == 0)
            // Get all threads to do something
    }
}

class Server {
    ...
        for (int i = 0; i < numberOfThreads; i++) {
            new serverThread(serverSocket.accept()).start();
        }
    ...
        sharedResource threadCount = new sharedResource(numberOfThreads);
    ...
}

class serverThread extends Thread {
...
//some code
Server.threadCount.incSharedCounter();
// Some more code to run when sharedCounte == 0
...
}

class Client {
...
}

　　　　　// Get all threads to do something

Threads (or rather Runnables, which you should implement rather than extending Thread) have a run method that contains the code they are expected to execute.

Once you call Thread#start (which in turn calls Runnable#run), the thread will start doing exactly that.

Since you seem to be new to multi-threading in Java, I recommend that you read an introduction to the Concurrency Utility package, that has been introduced in Java5 to make it easier to implement concurrent operations.

Specifically what you seem to be looking for is a way to "pause" the operation until a condition is met (in your case a counter having reached zero). For this, you should look at a CountDownLatch.

Indeed, the subject is broad, but I'll try to explain the basics. More details can be read from various blogs and articles. One of which is the Java trail.

It is best to see each thread as being runners (physical persons) that run alongside each other in a race. Each runner may perform any task while running. For example, take a cup of water from a table at a given moment in the race. Physically, they cannot both drink from the same cup at once, but in the virtual world, it is possible (this is where the line is drawn).

For example, take again two runners; each of them has to run back and forth a track, and push a button (shared by the runners) at each end for 1'000'000 times, the button is simply incrementing a counter by one each time. When they completed their run, what would be the value of the counter? In the physical world, it would be 2'000'000 because the runners cannot push the button at the same time, they would wait for the first one to leave first... that is unless they fight over it... Well, this is exactly what two threads would do. Consider this code :

public class ThreadTest extends Thread {
    static public final int TOTAL_INC = 1000000;

    static public int counter = 0;

    @Override
    public void run() {
        for (int i=0; i<TOTAL_INC; i++) {
            counter++;
        }
        System.out.println("Thread stopped incrementing counter " + TOTAL_INC + " times");
    }

    public static void main(String[] args) throws InterruptedException {
        Thread t1 = new ThreadTest();
        Thread t2 = new ThreadTest();

        t1.start();
        t2.start();

        t1.join();  // wait for each thread to stop on their own...
        t2.join();  // 

        System.out.println("Final counter is : " + counter + " which should be equal to " + TOTAL_INC * 2);
    }
}

An output could be something like

Thread stopped incrementing counter 1000000 times
Thread stopped incrementing counter 1000000 times
Final counter is : 1143470 which should be equal to 2000000

Once in a while, the two thread would just increment the same value twice; this is called a race condition.

Synchronizing the run method will not work, and you'd have to use some locking mechanism to prevent this from happening. Consider the following changes in the run method :

static private Object lock = new Object();
@Override
public void run() {
    for (int i=0; i<TOTAL_INC; i++) {
        synchronized(lock) {
            counter++;
        }
    }
    System.out.println("Thread stopped incrementing counter " + TOTAL_INC + " times");
}

Now the expected output is

...
Final counter is : 2000000 which should be equal to 2000000

We have synchronized our counter with a shared object. This is like putting a queue line before only one runner can access the button at once.

NOTE : this locking mechanism is called a mutex. If a resource can be accessed by n threads at once, you might consider using a semaphore.

Multithreading is also associated with deadlocking. A deadlock is when two threads mutually waits for the other to free some synchronized resource to continue. For example :

• Thread 1 starts
• Thread 2 starts
• Thread 1 acquire synchronized object1
• Thread 2 acquire synchronized object2
• Thread 2 needs to acquire object2 for continuing (locked by Thread 1)
• Thread 1 needs to acquire object1 for continuing (locked by Thread 2)
• Program hangs in deadlock

While there are many ways to prevent this from happening (it depends on what your threads are doing, and how they are implemented...) You should read about that particularly.

NOTE : the methods wait, notify and notifyAll can only be called when an object is synchronized. For example :

static public final int TOTAL_INC = 10;
static private int counter = 0;
static private Object lock = new Object();

static class Thread1 extends Thread {
    @Override
    public void run() {
        synchronized (lock) {
            for (int i=0; i<TOTAL_INC; i++) {
                try {
                    lock.wait();
                    counter++;
                    lock.notify();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

static class Thread2 extends Thread {
    @Override
    public void run() {
        synchronized (lock) {
            for (int i=0; i<TOTAL_INC; i++) {
                try {
                    lock.notify();
                    counter--;
                    lock.wait();
                } catch (InterruptedException e) {
                    /* ignored */
                }
            }
        }
    }
}

Notice that both threads are running their for...loop blocks within the synchronized block. (The result of counter == 0 when both threads end.) This can be achieved because they "let each other" access the synchronized resource via the resource's wait and notify methods. Without using those two methods, both threads would simply run sequentially and not concurrently (or more precisely, alternately).

I hope this shed some light about threads (in Java).

** UPDATE **

Here is a little proof of concept of everything discussed above, using the CountDownLatch class suggested by Thilo earlier :

static class Server {
    static public final int NODE_COUNT = 5;

    private List<RunnableNode> nodes;
    private CountDownLatch startSignal; 
    private Object lock = new Object();


    public Server() {
        nodes = Collections.synchronizedList(new ArrayList<RunnableNode>());
        startSignal = new CountDownLatch(Server.NODE_COUNT);
    }

    public Object getLock() {
        return lock;
    }

    public synchronized void connect(RunnableNode node) {
        if (startSignal.getCount() > 0) {
            startSignal.countDown();
            nodes.add(node);
            System.out.println("Received connection from node " + node.getId() + " (" + startSignal.getCount() + " remaining...)");
        } else {
            System.out.println("Client overflow! Refusing connection from node " + node.getId());
            throw new IllegalStateException("Too many nodes connected");
        }
    }

    public void shutdown() {
        for (RunnableNode node : nodes) {
            node.shutdown();
        }
    }

    public void awaitAllConnections() {
        try {
            startSignal.await();
            synchronized (lock) {
                lock.notifyAll();  // awake all nodes
            }
        } catch (InterruptedException e) {
            /* ignore */
            shutdown();  // properly close any connected node now
        }
    }
}


static class RunnableNode implements Runnable {
    private Server server;
    private int id;
    private boolean working;

    public RunnableNode(int id, Server server) {
        this.id = id;
        this.server = server;
        this.working = true;
    }
    public int getId() {
        return id;
    }
    public void run() {
        try {
            Thread.sleep((long) (Math.random() * 5) * 1000); // just wait randomly from 0 to 5 seconds....

            synchronized (server.getLock()) {
                server.connect(this);
                server.getLock().wait();
            }

            if (!Thread.currentThread().isAlive()) {
                throw new InterruptedException();
            } else {
                System.out.println("Node " + id + " started successfully!");

                while (working) {
                    Thread.yield();
                }

            }
        } catch (InterruptedException e1) {
            System.out.print("Ooop! ...");
        } catch (IllegalStateException e2) {
            System.out.print("Awwww! Too late! ...");
        }

        System.out.println("Node " + id + " is shutting down");
    }
    public void shutdown() {
        working = false; // shutdown node here...
    }
}


static public void main(String...args) throws InterruptedException {
    Server server = new Server();

    for (int i=0; i<Server.NODE_COUNT + 4; i++) {  // create 4 more nodes than needed...
        new Thread(new RunnableNode(i, server)).start();
    }
    server.awaitAllConnections();

    System.out.println("All connection received! Server started!");

    Thread.sleep(6000); 
    server.shutdown();
}

This is a broad topic. You might try reading through the official guides for concurrency (i.e. threading, more or less) in Java. This isn't something with cut-and-dried solutions; you have to design something.