Added files via upload

src/test/java/PortfolioAssetInterface.java

@@ -0,0 +1,13 @@
+package com.pci.cache.experimentation.chronicle.portfolio;
+
+public interface PortfolioAssetInterface
+{
+	public long getAssetId();
+	public void setAssetId(long assetId);
+
+	public int getShares();
+	public void setShares(int shares);
+
+	public double getPrice();
+	public void setPrice(double price);
+}

src/test/java/PortfolioValueAccumulator.java

@@ -0,0 +1,26 @@
+package com.pci.cache.experimentation.chronicle.portfolio;
+
+import java.util.function.Consumer;
+
+import org.apache.commons.lang3.mutable.MutableDouble;
+
+import net.openhft.chronicle.core.values.LongValue;
+import net.openhft.chronicle.map.MapEntry;
+
+public final class PortfolioValueAccumulator implements Consumer<MapEntry<LongValue, PortfolioAssetInterface>>
+{
+	final MutableDouble total;
+	final PortfolioAssetInterface asset;
+	public PortfolioValueAccumulator(MutableDouble total, PortfolioAssetInterface asset)
+	{
+		this.total = total;
+		this.asset = asset;
+	}
+
+	@Override
+	public void accept(MapEntry<LongValue, PortfolioAssetInterface> e)
+	{
+		e.value().getUsing(asset);
+		total.add(asset.getShares() * asset.getPrice());
+	}
+}

src/test/java/PortfolioValueTest.java

@@ -0,0 +1,163 @@
+package com.pci.cache.experimentation.chronicle.portfolio;
+
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.apache.commons.lang3.mutable.MutableDouble;
+import org.junit.Test;
+
+import net.openhft.chronicle.core.values.LongValue;
+import net.openhft.chronicle.map.ChronicleMap;
+import net.openhft.chronicle.map.ChronicleMapBuilder;
+import net.openhft.chronicle.map.MapSegmentContext;
+import net.openhft.chronicle.values.Values;
+
+public class PortfolioValueTest
+{
+	private static final boolean useIterator = true;
+	private static final long nAssets = 10_000_000;
+	private static final int nThreads = Runtime.getRuntime().availableProcessors();
+	private static final int nRepetitions = 11; // 10 for computing average, throwing away the first one (warmup)
+
+	@Test
+	public void test() throws ExecutionException, InterruptedException
+	{
+		ChronicleMapBuilder<LongValue, PortfolioAssetInterface> mapBuilder = ChronicleMapBuilder.of(LongValue.class, PortfolioAssetInterface.class).entries(nAssets);
+
+		try (ChronicleMap<LongValue, PortfolioAssetInterface> cache = mapBuilder.create())
+		{
+			createData(cache);
+
+			// Compute multiple times to get an reasonable average compute time
+			for (int i = 0; i < nRepetitions; i++)
+			{
+				computeValue(cache);
+			}
+		}
+	}
+
+	private void createData(final ChronicleMap<LongValue, PortfolioAssetInterface> cache) throws ExecutionException, InterruptedException
+	{
+		long startTime = System.currentTimeMillis();
+
+		ExecutorService executor = Executors.newFixedThreadPool(nThreads);
+		Future<?>[] futures = new Future[nThreads];
+		long batchSize = nAssets / nThreads;
+
+		for (int t = 0; t < nThreads; t++)
+		{
+			final long batch = t;
+			futures[t] = executor.submit(() -> {
+				final LongValue key = Values.newHeapInstance(LongValue.class);
+				final PortfolioAssetInterface value = Values.newHeapInstance(PortfolioAssetInterface.class);
+
+				long start = batch * batchSize;
+				long end = Math.min(nAssets, (batch + 1) * batchSize);
+				long n = (end - start);
+
+				if (end > start)
+				{
+					System.out.println("Inserting batch " + (batch + 1) + "/" + nThreads + " of " + n + " records");
+
+					for (long k = start; k < end; k++)
+					{
+						key.setValue(k);
+						value.setAssetId(k);
+						value.setShares(1);
+						value.setPrice(2.0);
+						cache.put(key, value);
+					}
+				}
+			});
+		}
+
+		for (Future<?> future : futures)
+		{
+			future.get();
+		}
+
+		long elapsedTime = (System.currentTimeMillis() - startTime);
+		System.out.println("Data inserted in " + elapsedTime + " ms");
+	}
+
+	private static void computeValue(final ChronicleMap<LongValue, PortfolioAssetInterface> cache) throws ExecutionException, InterruptedException
+	{
+		computeValueUsingIterator(cache);
+	}
+
+	private static void computeValueUsingIterator(final ChronicleMap<LongValue, PortfolioAssetInterface> cache) throws ExecutionException, InterruptedException
+	{
+		long startTime = System.currentTimeMillis();
+
+		ExecutorService executor = Executors.newFixedThreadPool(nThreads);
+		@SuppressWarnings("unchecked")
+		Future<Double>[] futures = new Future[nThreads];
+		long batchSize = (useIterator ? cache.segments() : nAssets) / nThreads;
+
+		// Map
+		for (int t = 0; t < nThreads; t++)
+		{
+			final int batch = t;
+
+			futures[t] = executor.submit(() -> {
+				if (useIterator)
+				{
+					int start = (int) (batch * batchSize);
+					int end = (int) Math.min(cache.segments(), (batch + 1) * batchSize);
+					return computeTotalUsingIterator(cache, start, end);
+				}
+				else
+				{
+					long start = batch * batchSize;
+					long end = Math.min(nAssets, (batch + 1) * batchSize);
+					return computeTotalUsingKeys(cache, start, end);
+				}
+			});
+		}
+
+		// Reduce
+		double total = 0;
+		for (Future<Double> future : futures)
+		{
+			total += future.get();
+		}
+
+		long elapsedTime = (System.currentTimeMillis() - startTime);
+		System.out.println("Total Portfolio Value: " + total + " for " + cache.longSize() + ", computed in " + elapsedTime + " ms, using " + (useIterator ? "Iterator" : "Keys"));
+	}
+
+	protected static double computeTotalUsingIterator(final ChronicleMap<LongValue, PortfolioAssetInterface> cache, int start, int end)
+	{
+		if (end > start)
+		{
+			final PortfolioAssetInterface asset = Values.newHeapInstance(PortfolioAssetInterface.class);
+			PortfolioValueAccumulator accumulator = new PortfolioValueAccumulator(new MutableDouble(), asset);
+			for (int s = start; s < end; s++)
+			{
+				try (MapSegmentContext<LongValue, PortfolioAssetInterface, ?> context = cache.segmentContext(s))
+				{
+					context.forEachSegmentEntry(accumulator);
+				}
+			}
+			return accumulator.total.doubleValue();
+		}
+		return 0;
+	}
+
+	protected static double computeTotalUsingKeys(final ChronicleMap<LongValue, PortfolioAssetInterface> cache, long start, long end)
+	{
+		final LongValue key = Values.newHeapInstance(LongValue.class);
+		PortfolioAssetInterface asset = Values.newHeapInstance(PortfolioAssetInterface.class);
+
+		double total = 0;
+		for (long k = start; k < end; k++)
+		{
+			key.setValue(k);
+			asset = cache.getUsing(key, asset);
+			total += asset.getShares() * asset.getPrice();
+		}
+		return total;
+	}
+}