Readers/Writiers Pass by reference

Question

The below code is part of a Passing the Baton program in Java:

Main P1 (writer) P2 (writer) P3 (reader) P4 (reader)

Main();

package ReadersPreference;

import java.util.concurrent.Semaphore;

/**
  * * @author me
 */
public class Main {

public static void main(String[] args) {


    AnyData x = new AnyData(5.7);//gives writers something to write, 
                                   //readers something to read
    Semaphore e = new Semaphore(1);//control entry
    Semaphore r = new Semaphore(0);//used to delay readers
    Semaphore w = new Semaphore(0);//used to delay writers

    int nr = 0;//readers active
    int nw = 0;//writers active

    int dr = 0;//readers waiting
    int dw = 0;//writers waiting

    P1 r1 = new P1(e, r, w, x, nr, nw, dr, dw); // #reader thread 1
    P2 r2 = new P2(e, r, w, x, nr, nw, dr, dw); // #reader thread 2
    P5 r3 = new P5(e, r, w, x, nr, nw, dr, dw); // #reader thread 3
    P6 r4 = new P6(e, r, w, x, nr, nw, dr, dw); // #reader thread 4

    P3 w1 = new P3(e, r, w, x, nr, nw, dr, dw); // #writer thread 1
    P4 w2 = new P4(e, r, w, x, nr, nw, dr, dw); // #writer thread 2

System.out.println("threads commanded to start");

r1.start();     // calls run() method in Thread
r2.start();
r3.start();     
r4.start();   
w1.start();
w2.start();
}//end of main
}//end of class

Reader Process

package ReadersPreference;

import java.util.concurrent.Semaphore;

public class P1 extends Thread {

private Semaphore e;
private Semaphore r;
private Semaphore w;
private AnyData pillarbox;
private int nw;
private int nr;
private int dr;
private int dw;

public P1(Semaphore e, Semaphore r, Semaphore w, AnyData pbox,
        int nw, int nr, int dr, int dw) {

    this.nw = nw;
    this.nr = nr;
    this.dr = dr;
    this.dw = dw;

    this.e = e;
    this.r = r;
    this.w = w;
    pillarbox = pbox;
}// end of constructor

public void run() {

PERFORM OPERATIONS          

}// end of run method
}// end of class

Now, according to my output, it appears to work. However, my lecturer states two major flaws. One being that the counters (nr,nw,dr,dw) are passed by value rather than by reference. Meaning that each thread checks its own copy of the data rather than working with shared variables, which prevents the program operating as it should.

I've been reading up alot on passing by value and reference, and at first felt my head swelling, I think I understand most of it now, but I'm still not seeing a solution. I can understand why the Semaphores work as they're already a reference (rather their value being passed is a reference). I don't see where exactly the problem with the counters is or how to go about solving it.

When he says thread, does he refer to the process class (constructors/algorithm) or the instantiation in the main class or both?

The closest to a solution I've come up with, based on reading on sharing primitives as objects, is the following:

public void setCounters(int nr){nr = newNR;}   

Again though I'm vague on how to implement it

The second supposedly major flaw was that I created several process class files and ran each as a thread, instead of writing two (reader/writer) and using as many objects as needed. I'm fuzzy one what this means, but also, considering my output print statements are unique to each process, to uniquely ID each in the output for verification, why would this method be considered a flaw when it was done for a purpose, does it affect the solution to readers/writers?

Answer 1

One thing before I start, good variable names are always better than comments; stick to this rule and you cannot go wrong. Imagine I come across your e variable somewhere in the code, I now have to scroll to the top of the class and read the comment to see what it is for then go back to where I was. This makes the code almost unreadable...

Your first problem is that you are using int which is a primitive type, this is passed by value. @Marcin's solution is not threadsafe; if you do something like int++ then this can do all sorts of weird things (such as not incremented, returning the wrong value etc) when called by multiple threads. Always use threadsafe objects in multi-threaded operations.

As @Marcin suggests you can wrap your data in a class to reduce the amount of code:

public class SharedData<T> {

    private final T data;
    private final Semaphore entryControl = new Semaphore(1);
    private final Semaphore readerDelay = new Semaphore(0);
    private final Semaphore writerDelay = new Semaphore(0);
    private final AtomicInteger activeReaders = new AtomicInteger(0);
    private final AtomicInteger activeWriters = new AtomicInteger(0);
    private final AtomicInteger waitingReaders = new AtomicInteger(0);
    private final AtomicInteger waitingWriters = new AtomicInteger(0);

    public SharedData(final T data) {
        this.data = data;
    }
    //getters
}

I have made this a generic class but you can remove the generics and make the 'data' an 'Object' if you wish - generics is better as it provides type safety, read about that here.

The class uses the AtomicInteger object, this is a threadsafe integer that allows for atomic operations such as getAndSet(int newValue) - this removes the possibility of thread safety issues when accessing a single value but it looks like you may want to access two, this is still not threadsafe so you may want to add some methods to the data class along the lines of:

public synchronized void makeReaderActive() {
    //perform checks etc
    waitingReaders.decrementAndGet();
    activeReaders.incrementAndGet();
}

As otherwise there is the possibility that a reader could decrement the waitingReaders then something could read the activeReaders before that is incremented.

That, I think, deals with your first query; now onto your second. Your teacher says that you created several process class files rather than creating instances. This is because you copy pasted the same file calling it different things (P1, P2, etc), this is not how Java works. Consider your code where you create Semaphores:

Semaphore e = new Semaphore(1);//control entry
Semaphore r = new Semaphore(0);//used to delay readers

You have one class file (Semaphore.class) that you have made two instances of. You did not have to copy the JDK's Semaphore class into another file and create that. You did not have to do this:

Semaphore1 e = new Semaphore1(1);//control entry
Semaphore2 r = new Semaphore2(0);//used to delay readers

So; in your example I assume you have some Reader process and some Writer process, you then need two classes:

public class MyReader implements Callable<Void> {

    private final String name;
    private final SharedData sharedData;

    public MyReader(final String name, final SharedData sharedData) {
        this.name = name;
        this.sharedData = sharedData;
    }

    @Override
    public Void call() {
        //do stuff
        return null;
    }
}

public class MyWriter implements Callable<Void> {

    private final String name;
    private final SharedData sharedData;

    public MyWriter(final String name, final SharedData sharedData) {
        this.name = name;
        this.sharedData = sharedData;
    }

    @Override
    public Void call() {
        //do stuff
        return null;
    }
}

One class represents what all your writers do and the other represents what all your readers do. I have made them Callables rather than threads; this brings me onto my next point.

You should not be using Thread objects, these are very low level and hard to manage properly. You should be using the new ExecutorService. So your main method now looks like:

public static void main(String[] args) {
    final SharedData<Double> sharedData = new SharedData<Double>(5.7);
    final List<Callable<Void>> myCallables = new LinkedList<Callable<Void>>();

    for (int i = 0; i < 4; ++i) {
        myCallables.add(new MyReader("reader" + i, sharedData));
    }
    for (int i = 0; i < 2; ++i) {
        myCallables.add(new MyWriter("writer" + i, sharedData));
    }

    final ExecutorService executorService = Executors.newFixedThreadPool(myCallables.size());
    final List<Future<Void>> futures;
    try {
        futures = executorService.invokeAll(myCallables);
    } catch (InterruptedException ex) {
        throw new RuntimeException(ex);
    }
    for (final Future<Void> future : futures) {
        try {
            future.get();
        } catch (InterruptedException ex) {
            throw new RuntimeException(ex);
        } catch (ExecutionException ex) {
            throw new RuntimeException(ex);
        }
    }
}

Let me walk you through this code so that you understand what is going on here. On the first line we create one of our new SharedData classes - in this case it is a SharedData<Double>; this means that the data it contains is of type Double; this can potentially be anything.

Next we create a List of Callable, this is where our work classes will go. We then put the work classes into our List in a loop; notice that we create instances of the same class for each MyReader and MyWriter - we do not need a class file for each one.

We then create a new ExecutorService with a thread pool the same size as the number of work classes we created - notice we can have fewer threads. In that case each thread would allocated a work class then when they were done they would be given a new one, until everything was done. In our case there are enough threads so each one simple get allocated one of the work classes.

We now call invokeAll passing in the List of work classes, this is where we ask the ExecutorService to call all the Callables. This method with block until everything is done, it may throw an InterrupedException as with any method that waits - in this case we throw the exception up and exit.

Finally, and this is where the ExecutorService shines, we loop over the returned list of Future classes and call get - this will throw an ExecutionException if there were any problems running the work associated with that future. In this case we throw up an exception.

Notice the Callables are of type Void (i.e. they are declared as Callable<Void>), this then filters through to the Futures as they are of type Future<Void>. If you wanted to return some data from each process then you could change the type to Callable<MyData> and then the get method of the Future would return you this data.