AtomicReference で遊ぶ

gistfile1.java

/**
 * 
 */
package sandbox.atomic;

import java.util.List;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicMarkableReference;
import java.util.concurrent.atomic.AtomicReference;

import junit.framework.TestCase;

import com.google.common.base.Supplier;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;

/**
 * {@link AtomicReference} で遊ぶ
 * 
 * @author monzou
 */
public class Atomic extends TestCase {

	public void testAtomicReference() throws Exception {

		final AtomicReference<Double> reference = new AtomicReference<Double>();
		final Supplier<Double> supplier = new Supplier<Double>() {

			@Override
			public Double get() {
				double v = Math.random();
				System.out.println(v);
				return v;
			}
		};

		int nThreads = 100;
		final CyclicBarrier start = new CyclicBarrier(nThreads + 1);
		final CyclicBarrier end = new CyclicBarrier(nThreads + 1);

		List<Callable<Double>> requests = Lists.newArrayList();
		for (int i = 0; i < nThreads; i++) {
			requests.add(new Callable<Double>() {

				@Override
				public Double call() throws Exception {
					start.await();
					Double v = setIfNullAndGet(reference, supplier);
					end.await();
					return v;
				}
			});
		}

		ExecutorService executor = Executors.newCachedThreadPool();
		List<Future<Double>> futures = Lists.newArrayList();
		for (Callable<Double> request : requests) {
			futures.add(executor.submit(request));
		}

		// 同時に始めて終わるまで待機
		start.await();
		end.await();

		Set<Double> values = Sets.newHashSet();
		for (Future<Double> future : futures) {
			values.add(future.get());
		}

		assertEquals(1, values.size());
		assertEquals(values.iterator().next(), reference.get());

		executor.shutdown();

	}

	public void testAtomicMarkableReference() throws Exception {

		// false -> true

		final AtomicMarkableReference<Double> reference = new AtomicMarkableReference<Double>(null, false);
		final Supplier<Double> nullValueSupplier = new Supplier<Double>() {

			@Override
			public Double get() {
				return null;
			}
		};
		final Supplier<Double> randomValueSupplier = new Supplier<Double>() {

			@Override
			public Double get() {
				double v = Math.random();
				System.out.println(v);
				return v;
			}
		};

		int nThreads = 100;
		final CyclicBarrier start = new CyclicBarrier(nThreads + 1);
		final CyclicBarrier end = new CyclicBarrier(nThreads + 1);

		List<Callable<Double>> requests = Lists.newArrayList();
		for (int i = 0; i < nThreads / 2; i++) {
			requests.add(new Callable<Double>() {

				@Override
				public Double call() throws Exception {
					start.await();
					Double v = setIfNotAvailableAndGet(reference, true, nullValueSupplier); // available
					end.await();
					return v;
				}
			});
		}
		for (int i = 0; i < nThreads / 2; i++) {
			requests.add(new Callable<Double>() {

				@Override
				public Double call() throws Exception {
					start.await();
					Thread.sleep(1000);
					Double v = setIfNotAvailableAndGet(reference, false, randomValueSupplier); // not available
					end.await();
					return v;
				}
			});
		}

		ExecutorService executor = Executors.newCachedThreadPool();
		List<Future<Double>> futures = Lists.newArrayList();
		for (Callable<Double> request : requests) {
			futures.add(executor.submit(request));
		}

		// 同時に始めて終わるまで待機
		start.await();
		end.await();

		Set<Double> values = Sets.newHashSet();
		for (Future<Double> future : futures) {
			values.add(future.get());
		}

		// false -> true -> false になるハズ
		assertEquals(2, values.size());
		assertNotNull(reference.getReference());
		assertFalse(reference.isMarked());

		executor.shutdown();

	}

	/**
	 * 参照が <code>null</code> でない場合はその参照を返し, <code>null</code> の場合は指定された {@link Supplier} の値を設定して参照値を返す。
	 * 
	 * @param reference 参照
	 * @param supplier 値
	 * @return 参照の示す値
	 */
	private static <V> V setIfNullAndGet(AtomicReference<V> reference, Supplier<? extends V> supplier) {
		V value = reference.get();
		if (value == null) {
			reference.compareAndSet(null, supplier.get());
		}
		return reference.get();
	}

	/**
	 * {@link AtomicMarkableReference} のマークが <code>availableMark</code> のときはその参照を返し, 異なる場合は {@link Supplier} の値を設定してマーク状態を更新し参照値を返す。
	 * 
	 * @param reference 参照
	 * @param availableMark 有効な状態を示すマーク
	 * @param supplier 値
	 * @return 参照の示す値
	 */
	private static <V> V setIfNotAvailableAndGet(AtomicMarkableReference<V> reference, boolean availableMark, Supplier<? extends V> supplier) {
		boolean[] markHolder = new boolean[1];
		V value = reference.get(markHolder);
		if (markHolder[0] != availableMark) {
			reference.compareAndSet(value, supplier.get(), !availableMark, availableMark);
		}
		return reference.getReference();
	}

}