processing/src/test/java/org/apache/druid/segment/data/BenchmarkIndexibleWrites.java - druid - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *   http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing,
  * software distributed under the License is distributed on an
  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  * KIND, either express or implied.  See the License for the
  * specific language governing permissions and limitations
  * under the License.
  */

 package org.apache.druid.segment.data;

 import com.carrotsearch.junitbenchmarks.AbstractBenchmark;
 import com.carrotsearch.junitbenchmarks.BenchmarkOptions;
 import com.carrotsearch.junitbenchmarks.Clock;
 import com.google.common.collect.ImmutableList;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.ListeningExecutorService;
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import org.apache.druid.java.util.common.StringUtils;
 import org.junit.Assert;
 import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;

 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
 import java.util.Random;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;


 // AbstractBenchmark makes this ignored unless explicitly run
 @RunWith(Parameterized.class)
 public class BenchmarkIndexibleWrites extends AbstractBenchmark
 {
   @Parameterized.Parameters
   public static Collection<Object[]> getParameters()
   {
     return ImmutableList.of(
         new Object[]{new ConcurrentStandardMap<Integer>()},
         new Object[]{new ConcurrentExpandable<Integer>()}
     );
   }

   public BenchmarkIndexibleWrites(ConcurrentIndexible<Integer> concurrentIndexible)
   {
     this.concurrentIndexible = concurrentIndexible;
   }

   private interface ConcurrentIndexible<V>
   {
     void set(Integer index, V object);

     V get(Integer index);

     void clear();
   }

   private static class ConcurrentStandardMap<V> implements ConcurrentIndexible<V>
   {
     private final ConcurrentHashMap<Integer, V> delegate = new ConcurrentHashMap<>();

     @Override
     public void set(Integer index, V object)
     {
       delegate.put(index, object);
     }

     @Override
     public V get(Integer index)
     {
       return delegate.get(index);
     }

     @Override
     public void clear()
     {
       delegate.clear();
     }
   }

   private static class ConcurrentExpandable<V> implements ConcurrentIndexible<V>
   {
     private static Integer INIT_SIZE = 1 << 10;
     private final AtomicReference<V[]> reference = new AtomicReference<>();
     private final AtomicLong resizeCount = new AtomicLong(0);
     private final Integer initSize;

     public ConcurrentExpandable()
     {
       this(INIT_SIZE);
     }

     @SuppressWarnings("unchecked")
     public ConcurrentExpandable(Integer initSize)
     {
       reference.set((V[]) new Object[initSize]);
       this.initSize = initSize;
     }

     @Override
     public V get(Integer index)
     {
       return reference.get()[index];
     }

     @SuppressWarnings("unchecked")
     @Override
     public void clear()
     {
       reference.set((V[]) new Object[initSize]);
     }

     private static Boolean wasCopying(Long val)
     {
       return (val & 1L) > 0;
     }

     @Override
     public void set(Integer index, V object)
     {
       ensureCapacity(index + 1);
       Long pre, post;
       do {
         pre = resizeCount.get();
         reference.get()[index] = object;
         post = resizeCount.get();
       } while (wasCopying(pre) || wasCopying(post) || (!pre.equals(post)));
     }

     private final Object resizeMutex = new Object();

     private void ensureCapacity(int capacity)
     {
       synchronized (resizeMutex) {
         if (reference.get().length < capacity) {
           // We increment twice per resize. Once before the copy starts and once after the swap.
           //
           // Any task who sees a resizeCount which is *odd* between the start and stop of their critical section
           // has access to a nebulous aggList and should try again
           //
           // Any task who sees a resizeCount which changes between the start and stop of their critical section
           // should also try again
           resizeCount.incrementAndGet();
           reference.set(Arrays.copyOf(reference.get(), reference.get().length << 1));
           resizeCount.incrementAndGet();
         }
       }
     }
   }

   private final ConcurrentIndexible<Integer> concurrentIndexible;
   private final Integer concurrentThreads = 1 << 2;
   private final Integer totalIndexSize = 1 << 20;

   @BenchmarkOptions(warmupRounds = 100, benchmarkRounds = 100, clock = Clock.REAL_TIME, callgc = true)
   @Ignore @Test
   /**
    * CALLEN - 2015-01-15 - OSX - Java 1.7.0_71-b14
    BenchmarkIndexibleWrites.testConcurrentWrites[0]: [measured 100 out of 200 rounds, threads: 1 (sequential)]
    round: 0.24 [+- 0.01], round.block: 0.00 [+- 0.00], round.gc: 0.02 [+- 0.00], GC.calls: 396, GC.time: 1.88, time.total: 50.60, time.warmup: 24.84, time.bench: 25.77
    BenchmarkIndexibleWrites.testConcurrentWrites[1]: [measured 100 out of 200 rounds, threads: 1 (sequential)]
    round: 0.15 [+- 0.01], round.block: 0.00 [+- 0.00], round.gc: 0.02 [+- 0.00], GC.calls: 396, GC.time: 2.11, time.total: 33.14, time.warmup: 16.09, time.bench: 17.05
    */
   public void testConcurrentWrites() throws ExecutionException, InterruptedException
   {
     final ListeningExecutorService executorService = MoreExecutors.listeningDecorator(
         Executors.newFixedThreadPool(
             concurrentThreads,
             new ThreadFactoryBuilder()
                 .setDaemon(false)
                 .setNameFormat("indexible-writes-benchmark-%d")
                 .build()
         )
     );
     final AtomicInteger index = new AtomicInteger(0);
     List<ListenableFuture<?>> futures = new ArrayList<>();

     final Integer loops = totalIndexSize / concurrentThreads;

     for (int i = 0; i < concurrentThreads; ++i) {
       futures.add(
           executorService.submit(
               new Runnable()
               {
                 @Override
                 public void run()
                 {
                   for (int i = 0; i < loops; ++i) {
                     final Integer idx = index.getAndIncrement();
                     concurrentIndexible.set(idx, idx);
                   }
                 }
               }
           )
       );
     }
     Futures.allAsList(futures).get();
     Assert.assertTrue(StringUtils.format("Index too small %d, expected %d across %d loops", index.get(), totalIndexSize, loops), index.get() >= totalIndexSize);
     for (int i = 0; i < index.get(); ++i) {
       Assert.assertEquals(i, concurrentIndexible.get(i).intValue());
     }
     concurrentIndexible.clear();
     futures.clear();
     executorService.shutdown();
   }

   /**
    BenchmarkIndexibleWrites.TestConcurrentReads[0]: [measured 100 out of 200 rounds, threads: 1 (sequential)]
    round: 0.28 [+- 0.02], round.block: 0.00 [+- 0.00], round.gc: 0.02 [+- 0.00], GC.calls: 396, GC.time: 1.84, time.total: 59.98, time.warmup: 30.51, time.bench: 29.48
    BenchmarkIndexibleWrites.TestConcurrentReads[1]: [measured 100 out of 200 rounds, threads: 1 (sequential)]
    round: 0.12 [+- 0.01], round.block: 0.00 [+- 0.00], round.gc: 0.02 [+- 0.00], GC.calls: 396, GC.time: 2.05, time.total: 29.21, time.warmup: 14.65, time.bench: 14.55

    */
   @BenchmarkOptions(warmupRounds = 100, benchmarkRounds = 100, clock = Clock.REAL_TIME, callgc = true)
   @Ignore @Test
   public void testConcurrentReads() throws ExecutionException, InterruptedException
   {
     final ListeningExecutorService executorService = MoreExecutors.listeningDecorator(
         Executors.newFixedThreadPool(
             concurrentThreads,
             new ThreadFactoryBuilder()
                 .setDaemon(false)
                 .setNameFormat("indexible-writes-benchmark-reader-%d")
                 .build()
         )
     );
     final AtomicInteger index = new AtomicInteger(0);
     final AtomicInteger queryableIndex = new AtomicInteger(0);
     List<ListenableFuture<?>> futures = new ArrayList<>();

     final Integer loops = totalIndexSize / concurrentThreads;

     final AtomicBoolean done = new AtomicBoolean(false);

     final CountDownLatch start = new CountDownLatch(1);

     for (int i = 0; i < concurrentThreads; ++i) {
       futures.add(
           executorService.submit(
               new Runnable()
               {
                 @Override
                 public void run()
                 {
                   try {
                     start.await();
                   }
                   catch (InterruptedException e) {
                     throw new RuntimeException(e);
                   }
                   final Random rndGen = ThreadLocalRandom.current();
                   while (!done.get()) {
                     Integer idx = rndGen.nextInt(queryableIndex.get() + 1);
                     Assert.assertEquals(idx, concurrentIndexible.get(idx));
                   }
                 }
               }
           )
       );
     }

     {
       final Integer idx = index.getAndIncrement();
       concurrentIndexible.set(idx, idx);
       start.countDown();
     }
     for (int i = 1; i < totalIndexSize; ++i) {
       final Integer idx = index.getAndIncrement();
       concurrentIndexible.set(idx, idx);
       queryableIndex.incrementAndGet();
     }
     done.set(true);

     Futures.allAsList(futures).get();
     executorService.shutdown();

     Assert.assertTrue(StringUtils.format("Index too small %d, expected %d across %d loops", index.get(), totalIndexSize, loops), index.get() >= totalIndexSize);
     for (int i = 0; i < index.get(); ++i) {
       Assert.assertEquals(i, concurrentIndexible.get(i).intValue());
     }
     concurrentIndexible.clear();
     futures.clear();
   }
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing,
	* software distributed under the License is distributed on an
	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	* KIND, either express or implied. See the License for the
	* specific language governing permissions and limitations
	* under the License.
	*/

	package org.apache.druid.segment.data;

	import com.carrotsearch.junitbenchmarks.AbstractBenchmark;
	import com.carrotsearch.junitbenchmarks.BenchmarkOptions;
	import com.carrotsearch.junitbenchmarks.Clock;
	import com.google.common.collect.ImmutableList;
	import com.google.common.util.concurrent.Futures;
	import com.google.common.util.concurrent.ListenableFuture;
	import com.google.common.util.concurrent.ListeningExecutorService;
	import com.google.common.util.concurrent.MoreExecutors;
	import com.google.common.util.concurrent.ThreadFactoryBuilder;
	import org.apache.druid.java.util.common.StringUtils;
	import org.junit.Assert;
	import org.junit.Ignore;
	import org.junit.Test;
	import org.junit.runner.RunWith;
	import org.junit.runners.Parameterized;

	import java.util.ArrayList;
	import java.util.Arrays;
	import java.util.Collection;
	import java.util.List;
	import java.util.Random;
	import java.util.concurrent.ConcurrentHashMap;
	import java.util.concurrent.CountDownLatch;
	import java.util.concurrent.ExecutionException;
	import java.util.concurrent.Executors;
	import java.util.concurrent.ThreadLocalRandom;
	import java.util.concurrent.atomic.AtomicBoolean;
	import java.util.concurrent.atomic.AtomicInteger;
	import java.util.concurrent.atomic.AtomicLong;
	import java.util.concurrent.atomic.AtomicReference;


	// AbstractBenchmark makes this ignored unless explicitly run
	@RunWith(Parameterized.class)
	public class BenchmarkIndexibleWrites extends AbstractBenchmark
	{
	@Parameterized.Parameters
	public static Collection<Object[]> getParameters()
	{
	return ImmutableList.of(
	new Object[]{new ConcurrentStandardMap<Integer>()},
	new Object[]{new ConcurrentExpandable<Integer>()}
	);
	}

	public BenchmarkIndexibleWrites(ConcurrentIndexible<Integer> concurrentIndexible)
	{
	this.concurrentIndexible = concurrentIndexible;
	}

	private interface ConcurrentIndexible<V>
	{
	void set(Integer index, V object);

	V get(Integer index);

	void clear();
	}

	private static class ConcurrentStandardMap<V> implements ConcurrentIndexible<V>
	{
	private final ConcurrentHashMap<Integer, V> delegate = new ConcurrentHashMap<>();

	@Override
	public void set(Integer index, V object)
	{
	delegate.put(index, object);
	}

	@Override
	public V get(Integer index)
	{
	return delegate.get(index);
	}

	@Override
	public void clear()
	{
	delegate.clear();
	}
	}

	private static class ConcurrentExpandable<V> implements ConcurrentIndexible<V>
	{
	private static Integer INIT_SIZE = 1 << 10;
	private final AtomicReference<V[]> reference = new AtomicReference<>();
	private final AtomicLong resizeCount = new AtomicLong(0);
	private final Integer initSize;

	public ConcurrentExpandable()
	{
	this(INIT_SIZE);
	}

	@SuppressWarnings("unchecked")
	public ConcurrentExpandable(Integer initSize)
	{
	reference.set((V[]) new Object[initSize]);
	this.initSize = initSize;
	}

	@Override
	public V get(Integer index)
	{
	return reference.get()[index];
	}

	@SuppressWarnings("unchecked")
	@Override
	public void clear()
	{
	reference.set((V[]) new Object[initSize]);
	}

	private static Boolean wasCopying(Long val)
	{
	return (val & 1L) > 0;
	}

	@Override
	public void set(Integer index, V object)
	{
	ensureCapacity(index + 1);
	Long pre, post;
	do {
	pre = resizeCount.get();
	reference.get()[index] = object;
	post = resizeCount.get();
	} while (wasCopying(pre) \|\| wasCopying(post) \|\| (!pre.equals(post)));
	}

	private final Object resizeMutex = new Object();

	private void ensureCapacity(int capacity)
	{
	synchronized (resizeMutex) {
	if (reference.get().length < capacity) {
	// We increment twice per resize. Once before the copy starts and once after the swap.
	//
	// Any task who sees a resizeCount which is odd between the start and stop of their critical section
	// has access to a nebulous aggList and should try again
	//
	// Any task who sees a resizeCount which changes between the start and stop of their critical section
	// should also try again
	resizeCount.incrementAndGet();
	reference.set(Arrays.copyOf(reference.get(), reference.get().length << 1));
	resizeCount.incrementAndGet();
	}
	}
	}
	}

	private final ConcurrentIndexible<Integer> concurrentIndexible;
	private final Integer concurrentThreads = 1 << 2;
	private final Integer totalIndexSize = 1 << 20;

	@BenchmarkOptions(warmupRounds = 100, benchmarkRounds = 100, clock = Clock.REAL_TIME, callgc = true)
	@Ignore @Test
	/**
	* CALLEN - 2015-01-15 - OSX - Java 1.7.0_71-b14
	BenchmarkIndexibleWrites.testConcurrentWrites[0]: [measured 100 out of 200 rounds, threads: 1 (sequential)]
	round: 0.24 [+- 0.01], round.block: 0.00 [+- 0.00], round.gc: 0.02 [+- 0.00], GC.calls: 396, GC.time: 1.88, time.total: 50.60, time.warmup: 24.84, time.bench: 25.77
	BenchmarkIndexibleWrites.testConcurrentWrites[1]: [measured 100 out of 200 rounds, threads: 1 (sequential)]
	round: 0.15 [+- 0.01], round.block: 0.00 [+- 0.00], round.gc: 0.02 [+- 0.00], GC.calls: 396, GC.time: 2.11, time.total: 33.14, time.warmup: 16.09, time.bench: 17.05
	*/
	public void testConcurrentWrites() throws ExecutionException, InterruptedException
	{
	final ListeningExecutorService executorService = MoreExecutors.listeningDecorator(
	Executors.newFixedThreadPool(
	concurrentThreads,
	new ThreadFactoryBuilder()
	.setDaemon(false)
	.setNameFormat("indexible-writes-benchmark-%d")
	.build()
	)
	);
	final AtomicInteger index = new AtomicInteger(0);
	List<ListenableFuture<?>> futures = new ArrayList<>();

	final Integer loops = totalIndexSize / concurrentThreads;

	for (int i = 0; i < concurrentThreads; ++i) {
	futures.add(
	executorService.submit(
	new Runnable()
	{
	@Override
	public void run()
	{
	for (int i = 0; i < loops; ++i) {
	final Integer idx = index.getAndIncrement();
	concurrentIndexible.set(idx, idx);
	}
	}
	}
	)
	);
	}
	Futures.allAsList(futures).get();
	Assert.assertTrue(StringUtils.format("Index too small %d, expected %d across %d loops", index.get(), totalIndexSize, loops), index.get() >= totalIndexSize);
	for (int i = 0; i < index.get(); ++i) {
	Assert.assertEquals(i, concurrentIndexible.get(i).intValue());
	}
	concurrentIndexible.clear();
	futures.clear();
	executorService.shutdown();
	}

	/**
	BenchmarkIndexibleWrites.TestConcurrentReads[0]: [measured 100 out of 200 rounds, threads: 1 (sequential)]
	round: 0.28 [+- 0.02], round.block: 0.00 [+- 0.00], round.gc: 0.02 [+- 0.00], GC.calls: 396, GC.time: 1.84, time.total: 59.98, time.warmup: 30.51, time.bench: 29.48
	BenchmarkIndexibleWrites.TestConcurrentReads[1]: [measured 100 out of 200 rounds, threads: 1 (sequential)]
	round: 0.12 [+- 0.01], round.block: 0.00 [+- 0.00], round.gc: 0.02 [+- 0.00], GC.calls: 396, GC.time: 2.05, time.total: 29.21, time.warmup: 14.65, time.bench: 14.55

	*/
	@BenchmarkOptions(warmupRounds = 100, benchmarkRounds = 100, clock = Clock.REAL_TIME, callgc = true)
	@Ignore @Test
	public void testConcurrentReads() throws ExecutionException, InterruptedException
	{
	final ListeningExecutorService executorService = MoreExecutors.listeningDecorator(
	Executors.newFixedThreadPool(
	concurrentThreads,
	new ThreadFactoryBuilder()
	.setDaemon(false)
	.setNameFormat("indexible-writes-benchmark-reader-%d")
	.build()
	)
	);
	final AtomicInteger index = new AtomicInteger(0);
	final AtomicInteger queryableIndex = new AtomicInteger(0);
	List<ListenableFuture<?>> futures = new ArrayList<>();

	final Integer loops = totalIndexSize / concurrentThreads;

	final AtomicBoolean done = new AtomicBoolean(false);

	final CountDownLatch start = new CountDownLatch(1);

	for (int i = 0; i < concurrentThreads; ++i) {
	futures.add(
	executorService.submit(
	new Runnable()
	{
	@Override
	public void run()
	{
	try {
	start.await();
	}
	catch (InterruptedException e) {
	throw new RuntimeException(e);
	}
	final Random rndGen = ThreadLocalRandom.current();
	while (!done.get()) {
	Integer idx = rndGen.nextInt(queryableIndex.get() + 1);
	Assert.assertEquals(idx, concurrentIndexible.get(idx));
	}
	}
	}
	)
	);
	}

	{
	final Integer idx = index.getAndIncrement();
	concurrentIndexible.set(idx, idx);
	start.countDown();
	}
	for (int i = 1; i < totalIndexSize; ++i) {
	final Integer idx = index.getAndIncrement();
	concurrentIndexible.set(idx, idx);
	queryableIndex.incrementAndGet();
	}
	done.set(true);

	Futures.allAsList(futures).get();
	executorService.shutdown();

	Assert.assertTrue(StringUtils.format("Index too small %d, expected %d across %d loops", index.get(), totalIndexSize, loops), index.get() >= totalIndexSize);
	for (int i = 0; i < index.get(); ++i) {
	Assert.assertEquals(i, concurrentIndexible.get(i).intValue());
	}
	concurrentIndexible.clear();
	futures.clear();
	}
	}