TestParallelism

package jsr166y;

import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import org.junit.Test;

import junit.framework.Assert;

/**
 * @author Quartet FS
 */
@SuppressWarnings("serial")
public class TestParallelism {

    @Test
    public void testWorkersAllocation() throws Exception {
        final ForkJoinPool pool = new ForkJoinPool();
        final AtomicBoolean success = new AtomicBoolean(false);
        final AtomicInteger running = new AtomicInteger(0);

        // Run our test in the pool.
        // We fork some work to do and make sure
        // that we reach the expected parallelism level.
        final int numIterations = 1 << 15;
        pool.invoke(new RecursiveAction() {
            @Override
            protected void compute() {
                for (int i = 0; i < numIterations; ++i) {
                    new RecursiveAction() {
                        @Override
                        protected void compute() {
                            // No work if we have already achieved
                            // great success
                            if (success.get())
                                return;

                            // Register our worker and do some work.
                            final int numWorkers = running.incrementAndGet();
                            try {
                                // Stop if we got enough workers
                                if (numWorkers >= pool.getParallelism()) {
                                    success.set(true);
                                    return;
                                }

                                // Burn some CPU.
                                doSomething();
                            } finally {
                                running.decrementAndGet();
                            }
                        }
                    }.fork();
                }

                // Quiesce the pool before returning
                pool.helpQuiescePool(((ForkJoinWorkerThread)Thread.currentThread()).workQueue);
            }
        });

        // Make sure we were successful
        Assert.assertTrue("Did not have enough workers allocated", success.get());
    }

    /**
     * Do something that will burn some CPU.
     */
    protected void doSomething() {
        final long start = System.nanoTime();
        do {} while ((System.nanoTime() - start) < 100L * 1000L * 1000L);
    }


}