Java Threads reading and writing from a shared stack array

Question

The result of this program should be the same with 1 or 2 or 3 threads. However, the result with thread 1 is the real one. I think I am meshing up with shared and private variables, what am I doing wrong? Threads have to read from the stack an interval and then calculate the quadrature model.If the error is small enough (i.e. within the specified accuracy) then we have the solution. If the error is still too large, the interval is divided into two with half the required error given to each half of the interval. The quadrature is applied again and so on until the error is small enough. The main problem arises through premature termination of the threads. The stack may be empty, but another thread might be about to place new tasks on it. The solution to this is to keep a count of the "active" threads, that is, the ones which are currently processing an interval. Then the code should terminate only when the stack is empty and there are no active threads...

Please, any help would be very appreciate it?

Cheers

 import java.lang.Integer;


class quadtest  {

/* Adaptive Quadrature Code. Finds the value of an integral of a 
   function on a closed interval to a specified accuracy.
*/

  public static void main (String args[]) {

    int nthreads = Integer.parseInt(args[0]);

    double left, right, eps;
    double start_time, time;

    Quad quad =null;
    //Counter counter = new Counter();

    left  = 0.0;
    right = 1.0;
    eps   = 1.0E-11;

    System.out.println("Adaptive Quadrature Program \n");
    System.out.println("eps="+eps+"    n=10000");
    start_time = System.currentTimeMillis();

    //Start threads
    Thread thread_object [] = new Thread[nthreads];

    for(int i=0;i<nthreads;i++){    
        quad = new Quad(left,right,eps,i,nthreads);
        thread_object[i]=new Thread(quad);
    }
    for(int i=0;i<nthreads;i++){    
        thread_object[i].start();
    }
    //Join the threads
    for(int i=0;i<nthreads;i++){
        try{
        thread_object[i].join();
        }catch(InterruptedException x){}
    }

    time = (double) (System.currentTimeMillis()-start_time) / 1000.;
    System.out.println("Result  = "  + quad.getResult() );
    System.out.println("Execution time = "  + time + " seconds ");


    }
}

    import java.lang.Runnable;
import java.util.concurrent.atomic.AtomicInteger;

class Quad implements Runnable{
    //Shared Variables 
    static volatile double [][] stack;
    static volatile boolean first=false;
    static volatile double FinalResult;
    static AtomicInteger threadCounter;
    static AtomicInteger writing;
    static AtomicInteger stackpointer;
    static int nthreads;
    //Constants
    static final int stacksize = 1000;
    static final int il = 0;
    static final int ir = 1;
    static final int ie = 2;
    static final int dims = 3;
    //Private Variables
    private  int tid;
    double left,right,eps;
    private double result;
    private double l,r,ep;


    public Quad(double left, double right, double eps,int tid,int nthreads) {

        this.left = left;
        this.right = right;
        this.eps = eps; 
        this.tid=tid;
        Quad.nthreads = nthreads;
        result = 0.0;
        //Only one thread will do it
        if(first==false){
            first=true;
            stack        =  new double [dims][stacksize];
            threadCounter=  new AtomicInteger(0);
            writing      =  new AtomicInteger(0);
            stackpointer =  new AtomicInteger(1);

            stack[il][stackpointer.get()] = left;
            stack[ir][stackpointer.get()] = right;
            stack[ie][stackpointer.get()] = eps;
            FinalResult=0.0;
         }
    }
    public void run(){
        stackops();
        add();
    }

   public void stackops() {

       double abserror,m, est1, est2;

       while ((stackpointer.get() >= 1)|| threadCounter.get()>0) {


           // Pop next interval off stack.
           synchronized (this){
               threadCounter.incrementAndGet();
               while (writing.get()==1){}
               pop();
           }
           // Compute estimates.
           m    = 0.5 * (l + r);
           est1 = 0.5 * (r - l) * (func(l) + func(r)) ;
           est2 = 0.5 * ((m - l) * (func(l) + func(m)) + (r - m) * 
                 (func(m) + func(r)));
           abserror = Math.abs(est2-est1) / 3.0;


           // Check for desired accuracy: push both halves onto the
           // stack if not accurate enough.        
           if (abserror <= ep) {
               result += est2;
                //System.out.println("ERROR->ID "+tid+"-abserror="+abserror+"-ep="+ep );
                //System.out.flush();
           } else {        
               if (stackpointer.get()+ 2 > stacksize) {
               System.out.println("Stack too small, try stacksize = " 
                          + 2*stacksize);
               }
               //Push into the stack
               synchronized (this){
                    push(m);
                }
             }//end else
             threadCounter.decrementAndGet();
           }//end while
   }//end method

    private synchronized void add(){
        FinalResult +=result;
    }

    private void pop(){
        if(stackpointer.get()>0){
            l   = stack[il][stackpointer.get()];
            r   = stack[ir][stackpointer.get()];
            ep  = stack[ie][stackpointer.get()];
            stackpointer.decrementAndGet();
        }
    }
    private void push (double m){
        writing.set(1);
            if(stackpointer.get()>=-1){
                stackpointer.incrementAndGet();
                stack[il][stackpointer.get()] = l;
                stack[ir][stackpointer.get()] = m;
                stack[ie][stackpointer.get()] = ep * 0.5;

                stackpointer.incrementAndGet();
                stack[il][stackpointer.get()] = m;
                stack[ir][stackpointer.get()] = r;
                stack[ie][stackpointer.get()] = ep * 0.5;
            }
       writing.set(0);
    } 

    public  double getResult(){
        return FinalResult;
    }

    private double func(double x) {

    double q;
    int n;

    n = 10000;
    q = 1000.0;

    for(int i=0;i<n;i++) {
        q -= x;
    }
    if (q == 1.0e10) System.out.println("q = " + q);

    return x * x;

    }
}

It should be almost done, but it takes longer time with more threads, first problem and also the result is not correct with more than 1 thread... — Manolete
– Manolete, Commented Mar 12, 2011 at 12:11

Karmastan · Accepted Answer · 2011-03-12 20:13:16Z

2

Your code does not actually have any mutual exclusion.

When using the synchronized keyword, you must actually synchronize on an object that all threads share. However, the this in your synchronized(this){} statements refers to the unshared Quad objects.
Your writes to FinalResult are not synchronized for the same reason. Also, volatile is unnecessary.
It appears you are trying to use writing as a custom spin-loop to keep multiple threads from popping at the same time. You don't need this--your synchronized blocks should have taken care of this--and you got it wrong. Imagine that one threads starts executing pop() and before it can do the first write, it gets rescheduled. Plus, you have a write in push too that isn't guarded. What if pop() and push() get called concurrently by two separate threads?

Other notes:

If items 2-3 above worked properly, there would be no need for stackPointer to be atomic.

You can initialize static data where they are defined for objects too. I.e.:

class Quad implements Runnable {
    static AtomicInteger threadCounter = new AtomicInteger(0);
    ...
}

answered Mar 12, 2011 at 20:13

Karmastan

5,74620 silver badges24 bronze badges

Sign up to request clarification or add additional context in comments.

2 Comments

Manolete Over a year ago

Thanks a lot Karmastan. What can I use instead of this? Where about in the constructor it should be created? I am very confused..I could add lock1 = new Object(); into the constructor and then use it instead of this. I was wrong with writing as I understand now that it's done by the synchronized clause whit mutual exclusion. However, it seems that the code never finishes so, I think I am not actually sharing the stack and every thread has its own stack.

Karmastan Over a year ago

@Manuel: create a static private Object lock = new Object(); and use that for your synchronized blocks (including add())

Arjan Tijms · Accepted Answer · 2011-03-12 12:13:42Z

0

Without having read all your code but with only glancing at it, it seems that this is exactly the kind of thing that the join/fork framework that will debut in JDK 7 is designed for to solve. See http://blog.quibb.org/2010/03/jsr-166-the-java-forkjoin-framework/

Instead of messing with threads and synchronization yourself, I would suggest looking at this. JDK 7 builds can already be downloaded (and seem to be rather stable). Alternatively there's a standalone version of join/form that can be downloaded separately (see http://g.oswego.edu/dl/concurrency-interest/).

answered Mar 12, 2011 at 12:13

Arjan Tijms

38.2k12 gold badges112 silver badges144 bronze badges

4 Comments

Manolete Over a year ago

I think threads in Java are a bit messy, unfortunately I am doing a HPC comparison with different languages and it has to be done with threads in order to be able to be comparable...

Manolete Over a year ago

It should be easy and straightforward, the only problem is I am not used to Java and my experience is very short...

Arjan Tijms Over a year ago

fork/join uses threads as well. You can initialize the pool with the amount of threads you want to be used, or don't specify this number and let the lib choose for you (which will simply be a number equal to the amount of cores you have).

Arjan Tijms Over a year ago

See http://download.oracle.com/javase/7/docs/api/java/util/concurrent/ForkJoinPool.html#ForkJoinPool(int)

Manolete · Accepted Answer · 2011-03-13 20:20:21Z

class Quad implements Runnable{ //Shared Variables static volatile double [][] stack; static volatile boolean first=true; static Object lock1; static Object lock2; static double FinalResult; static AtomicInteger threadCounter; static AtomicInteger stackpointer; static int nthreads; //Constants static final int stacksize = 1000; static final int il = 0; static final int ir = 1; static final int ie = 2; static final int dims = 3; //Private Variables int tid; double left,right,eps;

private double result;
private double l,r,ep;
private boolean calculate;

public Quad(double left, double right, double eps,int tid,int nthreads) {

this.left  = left;
this.right = right;
this.eps   = eps; 
this.tid   = tid;
Quad.nthreads = nthreads;
result = 0.0;

synchronized(this){
    //Only the first thread will do it
    if(first==true){
    first=false;
    lock1 = new Object();
    lock2 = new Object();
    stack        =  new double [dims][stacksize];
    threadCounter=  new AtomicInteger(0);
    stackpointer =  new AtomicInteger(1);
    stack[il][stackpointer.get()] = left;
    stack[ir][stackpointer.get()] = right;
    stack[ie][stackpointer.get()] = eps;
    FinalResult=0.0;
    System.out.println("I am tid= "+tid );
    }
}
}
public void run(){
    stackops();
    add();
}

public void stackops() {

double abserror,m, est1, est2;
est2=est1=m=abserror=0;

while ((stackpointer.get() >= 1)|| threadCounter.get()>0) {


    // Pop next interval off stack.
    synchronized (lock1){
    pop();
    }
    // Compute estimates.
    if (calculate == true){

    m    = 0.5 * (l + r);
    est1 = 0.5 * (r - l) * (func(l) + func(r)) ;
    est2 = 0.5 * ((m - l) * (func(l) + func(m)) + (r - m) * 
              (func(m) + func(r)));
    abserror = Math.abs(est2-est1) / 3.0;  


    if (abserror <= ep) {
        result += est2;
    } else {           
        //Push into the stack
        synchronized (lock1){
        push(m);
        }  
    }//end else
    threadCounter.decrementAndGet();
    }

   }//end while
System.out.println("I am " + tid+" result = "+result);
}//end method

private void add(){
synchronized(lock1){
    FinalResult +=result;
}
}

private void pop(){
if(stackpointer.get()>0){
    threadCounter.incrementAndGet();
    calculate =true;
    l   = stack[il][stackpointer.get()];
    r   = stack[ir][stackpointer.get()];
    ep  = stack[ie][stackpointer.get()];
    stackpointer.decrementAndGet();
}else{
    calculate =false;
}
}
private void push (double m){

if(stackpointer.get()>=-1){
    stackpointer.incrementAndGet();
    stack[il][stackpointer.get()] = l;
    stack[ir][stackpointer.get()] = m;
    stack[ie][stackpointer.get()] = ep * 0.5;

    stackpointer.incrementAndGet();
    stack[il][stackpointer.get()] = m;
    stack[ir][stackpointer.get()] = r;
    stack[ie][stackpointer.get()] = ep * 0.5;
}
} 

public  double getResult(){
    return FinalResult;
}

private double func(double x) {

double q;
int n;

n = 10000;
q = 1000.0;

for(int i=0;i<n;i++) {
    q -= x;
}
if (q == 1.0e10) System.out.println("q = " + q);

return x * x;

}

}

This is the new code. It works, however, it doesn't if I change the error number. It does not make sense to me... Something is fishy...

Collectives™ on Stack Overflow

Java Threads reading and writing from a shared stack array

3 Answers 3

2 Comments

4 Comments

1 Comment

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

3 Answers 3

2 Comments

4 Comments

1 Comment

Your Answer

Sign up or log in

Post as a guest

Related