Getting my head around Java threading, monitors etc....

Question

I am working on a project where I need to circularqueues and threading to simulate a train picking up and dropping off passengers. I am using Object.notify and Object.wait to control the threads. I am new to Threading applications and I am getting this error that I don't understand:

Exception in thread "Thread-0" java.lang.IllegalMonitorStateException
    at java.lang.Object.notify(Native Method)
    at Station.run(Station.java:37)
    at java.lang.Thread.run(Unknown Source)

Here is my main

public static void main(String[] args) 
    {
        Track theTrack = new Track();
        Station[] stations = {new Station(TRAIN_STATION_TYPE.ASHMONT)};
        Train[] trains = {new Train(TRAIN_TYPE.A_TRAIN, theTrack)};  

        Thread[] stationThreads = new Thread[stations.length]; 
        Thread[] trainThreads = new Thread[trains.length]; 

        theTrack.setStations(stations);
        theTrack.setTrains(trains); 
        for(int i = 0; i<stations.length; i++)
        {
            stationThreads[i] = new Thread(stations[i]);
            stationThreads[i].start(); 
        }
        for(int i = 0; i<trains.length; i++)
        {
            trainThreads[i] = new Thread(trains[i]);
            trainThreads[i].start(); 
        }
    }

Here is my Train class that will move along the track to each station dropping off and picking up passengers:

public class Train implements Runnable
{
    TRAIN_TYPE trainType; 
    int location; 
    private boolean goingForward, trainIsRunning; 
    private Track track; 
    private CircularQueue <Passenger> passengers;
    private Station stationBoarded; 
    Train()
    {
        this(null);
    }
    Train(TRAIN_TYPE trainType)
    {
        this(trainType, null); 
    }
    Train(TRAIN_TYPE trainType, Track track)
    {
        this.trainType = trainType;
        location = trainType.location(); 
        this.track = track; 
        trainIsRunning = true;
        goingForward = true; 
    }
    @Override
    public void run() 
    {
        while(trainIsRunning)
        {
            moveTrain();// train is moving up or down the track until it hits the location of a station 
            setStationBoarded(track.arriveAtStation(location)); // board station
            stationBoarded.queueTrain(this);// queue this train
            justWait(); // wait to be notified
            unloadPassengers();// unload passengers
            stationBoarded.notify();//notify station boarded to allow passengers to board the train. 
            justWait(); // wait to be notified to leave
            depart(); 
        }
    }
    public boolean boardTrain(Passenger p)
    {
        if(!passengers.reachedCapacity())
        {
            passengers.enqueue(p);
            return true; 
        }
            return false; 
    }
    public void moveTrain()
    {
        while(track.arriveAtStation(location) == null)
        {
            TIME_CONSTANT.TRAIN_MOVE_TIME.sleepAWhile();
            if(goingForward)
                location++;
            else
                location--;
            if(!track.isOnTrack(location))
                goingForward = !goingForward;
        }
    }
    public void unloadPassengers()
    {
        for(int i = 0; i<passengers.getLength(); i++)
        {
            if(passengers.peekAtIndex(i).getDestination() == stationBoarded.getTrainStation())
                stationBoarded.queuePassenger(passengers.remove(passengers.peekAtIndex(i)));
        }
    }
    public synchronized void justWait()
    {
        try {
            wait();
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } 
    }
    public void depart()
    {
        this.stationBoarded = null; 
    }
    public synchronized Passenger leaveTrain()
    {
        return passengers.dequeue(); 
    }
    public boolean isTrainIsRunning() {
        return trainIsRunning;
    }
    public void setTrainIsRunning(boolean trainIsRunning) {
        this.trainIsRunning = trainIsRunning;
    }
    public int getLocation() {
        return location;
    }
    public void setLocation(int location) {
        this.location = location;
    }
    public int getCapacity() 
    {
        return this.trainType.capacity();
    }
    public Station getStationBoarded() 
    {
        return stationBoarded;
    }
    public void setStationBoarded(Station stationBoarded) {
        this.stationBoarded = stationBoarded;
    }
    public boolean trainIsEmpty()
    {
        return this.passengers.isEmpty();
    }
    public boolean trainHasReachedCapacity()
    {
        return passengers.reachedCapacity();
    }
    public Track getTrack() {
        return track;
    }
    public void setTrack(Track track) {
        this.track = track;
    }
}

Here is my station class where it generates passengers to go on the train while it waits for a train to board the station:

import java.util.Random;

public class Station implements Runnable  
{
    CircularQueue <Passenger> passengers;
    CircularQueue <Train> trains; 
    TRAIN_STATION_TYPE trainStation; 
    Train trainInStation;
    int location, numOfPassengers;
    Passenger tempPassenger; 
    Random ran = new Random(); 
    Station()
    {
        this (null);
    }
    Station(TRAIN_STATION_TYPE tranStation)
    {
        this(tranStation, null);
    }
    Station(TRAIN_STATION_TYPE trainStation, Train train)
    {
        this.trainStation = trainStation;
        this.trainInStation = train; 
        this.trains = new CircularQueue(); 
        this.passengers = new CircularQueue();
    }
    public void run() 
    {
        while(trains.isEmpty())
        {
            genPassengers();
        }
        while(!trains.isEmpty())
        {
            trainInStation = trains.dequeue();
            trainInStation.notify(); // notify the train to let the passengers off
            justWait();  // wait for train to unload passengers 
            unloadPassengers(); 
            trainInStation.notify();//notify the train to depart
            departTrain(); 
        }
    }
    public void genPassengers()
    {
        TIME_CONSTANT.PASSENGER_GEN_TIME.sleepAWhile();
        passengers.enqueue(new Passenger()); 
    }
    public void departTrain()
    {
        trainInStation = null; 
    }
    public void arrive(Train train)
    {
        this.trainInStation = train;
    }
    public Train depart()
    {
        Train tempTrain = this.trainInStation;
        this.trainInStation = null;
        return tempTrain; 
    }
    public int getLocation() {
        return location;
    }
    public void setLocation(int location) {
        this.location = location;
    }
    public boolean isBoarded()
    {
        return (trainInStation != null);
    }
    public Train getTrain() {
        return trainInStation;
    }
    public void setTrain(Train train) {
        this.trainInStation = train;
    }
    public synchronized void queueTrain(Train train) {
        trains.enqueue(train);
    }
    public synchronized Train dequeue() {
        return trains.dequeue(); 
    }
    public synchronized void queuePassenger(Passenger passenger){
        passengers.enqueue(passenger);
    }
    public synchronized Passenger dequeuePassenger()
    {
        return passengers.dequeue(); 
    }
    public TRAIN_STATION_TYPE getTrainStation() {
        return trainStation;
    }
    public void setTrainStation(TRAIN_STATION_TYPE trainStation) {
        this.trainStation = trainStation;
    }
    public void justWait()
    {
        try {
            this.wait(); //wait for passengers to get off
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
    public void unloadPassengers()
    {
//      for(int i = 0; i<passengers.getLength() && !trainInStation.trainHasReachedCapacity(); i++)
//      {
//          trainInStation.boardTrain(passengers.dequeue());
//      }
        while(!passengers.isEmpty() && trainInStation.getCapacity()>numOfPassengers)
        {
            Passenger temp = dequeuePassenger(); 
            System.out.println(temp.toString() + " got on train " + this.trainInStation.trainType.getName()); 
            trainInStation.boardTrain(temp);
        }
    }
}

As you can see the control of the execution of the program flows between the train and the station objects using .wait(); and .notify();. I have a feeling that there is a much more elegant way to handle threading... Why does .notify() have a on monitor on it? Is there a better way to handle threading in my case? Sorry if these are stupid questions.

Answer 1

Exception in thread "Thread-0" java.lang.IllegalMonitorStateException
   at java.lang.Object.notify(Native Method)
   at Station.run(Station.java:37)
   at java.lang.Thread.run(Unknown Source)

This is trying to tell you that you are calling notify on an object without being inside a synchronized block for that object. It may be here:

stationBoarded.notify();

You'll need to synchronize it with something like:

synchronized (stationBoarded) {
    stationBoarded.notify();
}

or maybe one of these:

trainInStation.notify(); // notify the train to let the passengers off
...
trainInStation.notify();//notify the train to depart

Basically anyplace you call notify() or notifyAll().

Answer 2

You cannot call notify or wait on an object without first acquiring the lock on an object.

In Station.java, you are calling trainStation.notify() in your run() method, but the block of code is not synchronzied against trainStation.

synchronize the blocks of code that call wait or notify against the object you are calling wait or notify on and your immediate problems should go away.

Answer 3

You need to add a synchronized block around your wait() calls.