duarten/Combiner.java

## Combiner.java
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

public final class Combiner {

    public interface Action {
        public void apply();
    }

    private static final class Node {
        public Action request;
        public AtomicInteger wait = new AtomicInteger();
        public boolean complete;
        public AtomicReference<Node> next = new AtomicReference<Node>();
    }

    private ThreadLocal<Node> _myNode = new ThreadLocal<Node>() {
        @Override
        protected Node initialValue() {
          return new Node();
        }
    };

    private final int _limit;
    private AtomicReference<Node> _tail;

    public Combiner(int limit) {
        _limit = limit;
        _tail = new AtomicReference<Node>(new Node());
    }

    public void combine(Action action) {
        Node nextNode = _myNode.get();
        nextNode.complete = false;
        nextNode.wait.set(1);

        Node curNode = _tail.getAndSet(nextNode);
        _myNode.set(curNode);

        //
        // There's now a window where nextNode/_tail can't be reached.
        // So, any communication has to be done via the previous node
        // in the list, curNode.
        //

        curNode.request = action;
        curNode.next.lazySet(nextNode);

        // Wait until our request has been fulfilled or we are the combiner.

        final int maxSpins = 256;
        int spin = 0;
        while (curNode.wait.get() == 1) {
            if (++spin == maxSpins) {
              Thread.yield();
              spin = 0;
            }
        }

        if (curNode.complete)
            return;

        // We are now the combiner. We copy n's Next field into nn, as n will
        // become untouchable after n.wait.lazySet(0), due to reuse.
        Node n = curNode;
        Node nn;
        for (int c = 0; c < _limit && (nn = n.next.get()) != null; ++c, n = nn) {
            n.request.apply();

            n.next.set(null);
            n.request = null;
            n.complete = true;
            n.wait.lazySet(0);
        }

        // Make someone else the combiner.
        n.wait.set(0);
    }

    static final int threads = 100;
    static final int ops = 1 << 20;

    private static void testWithoutCombining() {
        final AtomicInteger[] counters = new AtomicInteger[5];
        for (int i = 0; i < counters.length; ++i)
            counters[i] = new AtomicInteger();

        final java.util.concurrent.CountDownLatch l = new java.util.concurrent.CountDownLatch(threads);
        for (int i = 0; i < threads; ++i) {
        new Thread() {
            @Override
            public void run() {
              for (int j = 0; j < ops; ++j) {
                for (AtomicInteger c : counters) {
                  c.incrementAndGet();
                }
              }
            l.countDown();
            }
          }.start();
        }

        do {
          try {
            l.await();
            break;
          } catch (InterruptedException e) { }
      } while (true);

      for (int i = 0; i < counters.length; ++i)
        System.out.println(counters[i].get());
    }

    private static void testWithCombining() {
    final int[] counters = new int[5];
    final Combiner c = new Combiner(100);
    final java.util.concurrent.CountDownLatch l = new java.util.concurrent.CountDownLatch(threads);
    for (int i = 0; i < threads; ++i) {
        new Thread() {
            @Override
            public void run() {
                for (int j = 0; j < ops; ++j) {
                    c.combine(new Combiner.Action() {
                        @Override
                        public void apply() {
                            for (int c = 0; c < counters.length; ++c) {
                                counters[c] += 1;
                            }
                        }});
                }
                l.countDown();
            }
        }.start();
    }

    do {
        try {
            l.await();
            break;
        } catch (InterruptedException e) { }
    } while (true);

    for (int i = 0; i < counters.length; ++i)
        System.out.println(counters[i]);
    }

    public static void main(String[] args) {
        long startTime = System.nanoTime();
        testWithCombining();
        //testWithoutCombining();
        long estimatedTime = System.nanoTime() - startTime;
        System.out.println("time: " + estimatedTime / 1000000);
    }
}
	import java.util.concurrent.atomic.AtomicInteger;
	import java.util.concurrent.atomic.AtomicReference;

	public final class Combiner {

	public interface Action {
	public void apply();
	}

	private static final class Node {
	public Action request;
	public AtomicInteger wait = new AtomicInteger();
	public boolean complete;
	public AtomicReference<Node> next = new AtomicReference<Node>();
	}

	private ThreadLocal<Node> _myNode = new ThreadLocal<Node>() {
	@Override
	protected Node initialValue() {
	return new Node();
	}
	};

	private final int _limit;
	private AtomicReference<Node> _tail;

	public Combiner(int limit) {
	_limit = limit;
	_tail = new AtomicReference<Node>(new Node());
	}

	public void combine(Action action) {
	Node nextNode = _myNode.get();
	nextNode.complete = false;
	nextNode.wait.set(1);

	Node curNode = _tail.getAndSet(nextNode);
	_myNode.set(curNode);

	//
	// There's now a window where nextNode/_tail can't be reached.
	// So, any communication has to be done via the previous node
	// in the list, curNode.
	//

	curNode.request = action;
	curNode.next.lazySet(nextNode);

	// Wait until our request has been fulfilled or we are the combiner.

	final int maxSpins = 256;
	int spin = 0;
	while (curNode.wait.get() == 1) {
	if (++spin == maxSpins) {
	Thread.yield();
	spin = 0;
	}
	}

	if (curNode.complete)
	return;

	// We are now the combiner. We copy n's Next field into nn, as n will
	// become untouchable after n.wait.lazySet(0), due to reuse.
	Node n = curNode;
	Node nn;
	for (int c = 0; c < _limit && (nn = n.next.get()) != null; ++c, n = nn) {
	n.request.apply();

	n.next.set(null);
	n.request = null;
	n.complete = true;
	n.wait.lazySet(0);
	}

	// Make someone else the combiner.
	n.wait.set(0);
	}

	static final int threads = 100;
	static final int ops = 1 << 20;

	private static void testWithoutCombining() {
	final AtomicInteger[] counters = new AtomicInteger[5];
	for (int i = 0; i < counters.length; ++i)
	counters[i] = new AtomicInteger();

	final java.util.concurrent.CountDownLatch l = new java.util.concurrent.CountDownLatch(threads);
	for (int i = 0; i < threads; ++i) {
	new Thread() {
	@Override
	public void run() {
	for (int j = 0; j < ops; ++j) {
	for (AtomicInteger c : counters) {
	c.incrementAndGet();
	}
	}
	l.countDown();
	}
	}.start();
	}

	do {
	try {
	l.await();
	break;
	} catch (InterruptedException e) { }
	} while (true);

	for (int i = 0; i < counters.length; ++i)
	System.out.println(counters[i].get());
	}

	private static void testWithCombining() {
	final int[] counters = new int[5];
	final Combiner c = new Combiner(100);
	final java.util.concurrent.CountDownLatch l = new java.util.concurrent.CountDownLatch(threads);
	for (int i = 0; i < threads; ++i) {
	new Thread() {
	@Override
	public void run() {
	for (int j = 0; j < ops; ++j) {
	c.combine(new Combiner.Action() {
	@Override
	public void apply() {
	for (int c = 0; c < counters.length; ++c) {
	counters[c] += 1;
	}
	}});
	}
	l.countDown();
	}
	}.start();
	}

	do {
	try {
	l.await();
	break;
	} catch (InterruptedException e) { }
	} while (true);

	for (int i = 0; i < counters.length; ++i)
	System.out.println(counters[i]);
	}

	public static void main(String[] args) {
	long startTime = System.nanoTime();
	testWithCombining();
	//testWithoutCombining();
	long estimatedTime = System.nanoTime() - startTime;
	System.out.println("time: " + estimatedTime / 1000000);
	}
	}