diff --git a/src/test/java/examples/portfolio/PortfolioAssetInterface.java b/src/test/java/examples/portfolio/PortfolioAssetInterface.java new file mode 100644 index 000000000..dbc7fd742 --- /dev/null +++ b/src/test/java/examples/portfolio/PortfolioAssetInterface.java @@ -0,0 +1,13 @@ +package examples.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); +} \ No newline at end of file diff --git a/src/test/java/examples/portfolio/PortfolioValueAccumulator.java b/src/test/java/examples/portfolio/PortfolioValueAccumulator.java new file mode 100644 index 000000000..23f0364f3 --- /dev/null +++ b/src/test/java/examples/portfolio/PortfolioValueAccumulator.java @@ -0,0 +1,26 @@ +package examples.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> +{ + final MutableDouble total; + final PortfolioAssetInterface asset; + public PortfolioValueAccumulator(MutableDouble total, PortfolioAssetInterface asset) + { + this.total = total; + this.asset = asset; + } + + @Override + public void accept(MapEntry e) + { + e.value().getUsing(asset); + total.add(asset.getShares() * asset.getPrice()); + } +} \ No newline at end of file diff --git a/src/test/java/examples/portfolio/PortfolioValueTest.java b/src/test/java/examples/portfolio/PortfolioValueTest.java new file mode 100644 index 000000000..92177d83c --- /dev/null +++ b/src/test/java/examples/portfolio/PortfolioValueTest.java @@ -0,0 +1,163 @@ +package examples.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 mapBuilder = ChronicleMapBuilder.of(LongValue.class, PortfolioAssetInterface.class).entries(nAssets); + + try (ChronicleMap 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 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 cache) throws ExecutionException, InterruptedException + { + computeValueUsingIterator(cache); + } + + private static void computeValueUsingIterator(final ChronicleMap cache) throws ExecutionException, InterruptedException + { + long startTime = System.currentTimeMillis(); + + ExecutorService executor = Executors.newFixedThreadPool(nThreads); + @SuppressWarnings("unchecked") + Future[] 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 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 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 context = cache.segmentContext(s)) + { + context.forEachSegmentEntry(accumulator); + } + } + return accumulator.total.doubleValue(); + } + return 0; + } + + protected static double computeTotalUsingKeys(final ChronicleMap 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; + } +}