java.util.concurrent.atomic.DoubleAdder#sum()源码实例Demo

/**
 * adds by multiple threads produce correct sum
 */
public void testAddAndSumMT() throws Throwable {
    final int incs = 1000000;
    final int nthreads = 4;
    final ExecutorService pool = Executors.newCachedThreadPool();
    DoubleAdder a = new DoubleAdder();
    CyclicBarrier barrier = new CyclicBarrier(nthreads + 1);
    for (int i = 0; i < nthreads; ++i)
        pool.execute(new AdderTask(a, barrier, incs));
    barrier.await();
    barrier.await();
    double total = (long)nthreads * incs;
    double sum = a.sum();
    assertEquals(sum, total);
    pool.shutdown();
}
 

@Override
public double f(Vec w, boolean parallel)
{
    DoubleAdder sum = new DoubleAdder();
    DoubleAdder weightSum = new DoubleAdder();
    
    ParallelUtils.run(parallel, D.size(), (start, end)->
    {
        for(int i = start; i < end; i++)
        {
            DataPoint dp = D.getDataPoint(i);
            Vec x = dp.getNumericalValues();
            double y = getTargetY(D, i);
            sum.add(loss.getLoss(w.dot(x), y)*D.getWeight(i));
            weightSum.add(D.getWeight(i));
        }
    });
    
    if(lambda0 > 0)
        return sum.sum()/weightSum.sum() + lambda0*w.dot(w);
    else
        return sum.sum()/weightSum.sum();
}
 

/**
 * adds by multiple threads produce correct sum
 */
public void testAddAndSumMT() throws Throwable {
    final int incs = 1000000;
    final int nthreads = 4;
    final ExecutorService pool = Executors.newCachedThreadPool();
    DoubleAdder a = new DoubleAdder();
    CyclicBarrier barrier = new CyclicBarrier(nthreads + 1);
    for (int i = 0; i < nthreads; ++i)
        pool.execute(new AdderTask(a, barrier, incs));
    barrier.await();
    barrier.await();
    double total = (long)nthreads * incs;
    double sum = a.sum();
    assertEquals(sum, total);
    pool.shutdown();
}
 

public void run() {
    phaser.arriveAndAwaitAdvance();
    phaser.arriveAndAwaitAdvance();
    DoubleAdder a = adder;
    for (int i = 0; i < incs; ++i)
        a.add(1.0);
    result = a.sum();
    phaser.arrive();
}
 

public void run() {
    phaser.arriveAndAwaitAdvance();
    phaser.arriveAndAwaitAdvance();
    DoubleAdder a = adder;
    for (int i = 0; i < incs; ++i)
        a.add(1.0);
    result = a.sum();
    phaser.arrive();
}
 

public void run() {
    phaser.arriveAndAwaitAdvance();
    phaser.arriveAndAwaitAdvance();
    DoubleAdder a = adder;
    for (int i = 0; i < incs; ++i)
        a.add(1.0);
    result = a.sum();
    phaser.arrive();
}
 

public void run() {
    phaser.arriveAndAwaitAdvance();
    phaser.arriveAndAwaitAdvance();
    DoubleAdder a = adder;
    for (int i = 0; i < incs; ++i)
        a.add(1.0);
    result = a.sum();
    phaser.arrive();
}
 

public void run() {
    phaser.arriveAndAwaitAdvance();
    phaser.arriveAndAwaitAdvance();
    DoubleAdder a = adder;
    for (int i = 0; i < incs; ++i)
        a.add(1.0);
    result = a.sum();
    phaser.arrive();
}
 

public void run() {
    phaser.arriveAndAwaitAdvance();
    phaser.arriveAndAwaitAdvance();
    DoubleAdder a = adder;
    for (int i = 0; i < incs; ++i)
        a.add(1.0);
    result = a.sum();
    phaser.arrive();
}
 

public void run() {
    phaser.arriveAndAwaitAdvance();
    phaser.arriveAndAwaitAdvance();
    DoubleAdder a = adder;
    for (int i = 0; i < incs; ++i)
        a.add(1.0);
    result = a.sum();
    phaser.arrive();
}
 

public void run() {
    try {
        barrier.await();
        DoubleAdder a = adder;
        for (int i = 0; i < incs; ++i)
            a.add(1.0);
        result = a.sum();
        barrier.await();
    } catch (Throwable t) { throw new Error(t); }
}
 

public void run() {
    phaser.arriveAndAwaitAdvance();
    phaser.arriveAndAwaitAdvance();
    DoubleAdder a = adder;
    for (int i = 0; i < incs; ++i)
        a.add(1.0);
    result = a.sum();
    phaser.arrive();
}
 

public void run() {
    phaser.arriveAndAwaitAdvance();
    phaser.arriveAndAwaitAdvance();
    DoubleAdder a = adder;
    for (int i = 0; i < incs; ++i)
        a.add(1.0);
    result = a.sum();
    phaser.arrive();
}
 

public void run() {
    phaser.arriveAndAwaitAdvance();
    phaser.arriveAndAwaitAdvance();
    DoubleAdder a = adder;
    for (int i = 0; i < incs; ++i)
        a.add(1.0);
    result = a.sum();
    phaser.arrive();
}
 

public void run() {
    phaser.arriveAndAwaitAdvance();
    phaser.arriveAndAwaitAdvance();
    DoubleAdder a = adder;
    for (int i = 0; i < incs; ++i)
        a.add(1.0);
    result = a.sum();
    phaser.arrive();
}
 

public void run() {
    phaser.arriveAndAwaitAdvance();
    phaser.arriveAndAwaitAdvance();
    DoubleAdder a = adder;
    for (int i = 0; i < incs; ++i)
        a.add(1.0);
    result = a.sum();
    phaser.arrive();
}
 

@Override
public double f(Vec w, boolean parallel)
{
    DoubleAdder sum = new DoubleAdder();
    Vec pred = new DenseVector(D.getClassSize());//store the predictions in
    //bias terms are at the end, treat them seperate and special
    final int subWSize = (w.length() - (useBiasTerm ? bs.length : 0) )/D.getClassSize();
    DoubleAdder weightSum = new DoubleAdder();
    ParallelUtils.run(parallel, D.size(), (start, end)->
    {
        Vec pred_local = pred.clone();
        for (int i = start; i < end; i++)
        {
            DataPoint dp = D.getDataPoint(i);
            Vec x = dp.getNumericalValues();
            for(int k = 0; k < pred_local.length(); k++)
                pred_local.set(k, new SubVector(k*subWSize, subWSize, w).dot(x));
            if(useBiasTerm)
                pred_local.mutableAdd(new SubVector(w.length()-bs.length, bs.length, w));
            loss.process(pred_local, pred_local);
            int y = D.getDataPointCategory(i);
            sum.add(loss.getLoss(pred_local, y)*D.getWeight(i));
            weightSum.add(D.getWeight(i));
        }
    });
    if(lambda0 > 0 )
        return sum.sum()/weightSum.sum() + lambda0*w.dot(w);
    return sum.sum();
}
 

public void run() {
    phaser.arriveAndAwaitAdvance();
    phaser.arriveAndAwaitAdvance();
    DoubleAdder a = adder;
    for (int i = 0; i < incs; ++i)
        a.add(1.0);
    result = a.sum();
    phaser.arrive();
}
 

public void run() {
    phaser.arriveAndAwaitAdvance();
    phaser.arriveAndAwaitAdvance();
    DoubleAdder a = adder;
    for (int i = 0; i < incs; ++i)
        a.add(1.0);
    result = a.sum();
    phaser.arrive();
}
 

public void run() {
    try {
        barrier.await();
        DoubleAdder a = adder;
        for (int i = 0; i < incs; ++i)
            a.add(1.0);
        result = a.sum();
        barrier.await();
    } catch (Throwable t) { throw new Error(t); }
}