From a2103bcb13ab31a352aa61b6c773f5503c26f3cd Mon Sep 17 00:00:00 2001 From: afbarbaro Date: Mon, 4 Apr 2016 23:21:35 -0500 Subject: [PATCH 1/8] Added files via upload --- src/test/java/PortfolioAssetInterface.java | 13 ++ src/test/java/PortfolioValueAccumulator.java | 26 +++ src/test/java/PortfolioValueTest.java | 163 +++++++++++++++++++ 3 files changed, 202 insertions(+) create mode 100644 src/test/java/PortfolioAssetInterface.java create mode 100644 src/test/java/PortfolioValueAccumulator.java create mode 100644 src/test/java/PortfolioValueTest.java diff --git a/src/test/java/PortfolioAssetInterface.java b/src/test/java/PortfolioAssetInterface.java new file mode 100644 index 000000000..4570555d4 --- /dev/null +++ b/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); +} \ No newline at end of file diff --git a/src/test/java/PortfolioValueAccumulator.java b/src/test/java/PortfolioValueAccumulator.java new file mode 100644 index 000000000..da7948ea7 --- /dev/null +++ b/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> +{ + 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/PortfolioValueTest.java b/src/test/java/PortfolioValueTest.java new file mode 100644 index 000000000..384a0f2f2 --- /dev/null +++ b/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 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; + } +} \ No newline at end of file From 63d22349bd853224e6e8d892c41751ceb101c0ad Mon Sep 17 00:00:00 2001 From: afbarbaro Date: Mon, 4 Apr 2016 23:24:28 -0500 Subject: [PATCH 2/8] Create portfolio --- src/test/java/examples/portfolio | 1 + 1 file changed, 1 insertion(+) create mode 100644 src/test/java/examples/portfolio diff --git a/src/test/java/examples/portfolio b/src/test/java/examples/portfolio new file mode 100644 index 000000000..8b1378917 --- /dev/null +++ b/src/test/java/examples/portfolio @@ -0,0 +1 @@ + From b7154837f0573b06654244c00140dff2ea187819 Mon Sep 17 00:00:00 2001 From: afbarbaro Date: Mon, 4 Apr 2016 23:27:41 -0500 Subject: [PATCH 3/8] Delete portfolio --- src/test/java/examples/portfolio | 1 - 1 file changed, 1 deletion(-) delete mode 100644 src/test/java/examples/portfolio diff --git a/src/test/java/examples/portfolio b/src/test/java/examples/portfolio deleted file mode 100644 index 8b1378917..000000000 --- a/src/test/java/examples/portfolio +++ /dev/null @@ -1 +0,0 @@ - From f47b13e3ccd6dee1650f8484ddc8ca9dabdebd94 Mon Sep 17 00:00:00 2001 From: afbarbaro Date: Mon, 4 Apr 2016 23:27:50 -0500 Subject: [PATCH 4/8] Delete PortfolioAssetInterface.java --- src/test/java/PortfolioAssetInterface.java | 13 ------------- 1 file changed, 13 deletions(-) delete mode 100644 src/test/java/PortfolioAssetInterface.java diff --git a/src/test/java/PortfolioAssetInterface.java b/src/test/java/PortfolioAssetInterface.java deleted file mode 100644 index 4570555d4..000000000 --- a/src/test/java/PortfolioAssetInterface.java +++ /dev/null @@ -1,13 +0,0 @@ -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); -} \ No newline at end of file From 0143bffedfbbf5580c48d790bc0c090174bd0638 Mon Sep 17 00:00:00 2001 From: afbarbaro Date: Mon, 4 Apr 2016 23:27:55 -0500 Subject: [PATCH 5/8] Delete PortfolioValueAccumulator.java --- src/test/java/PortfolioValueAccumulator.java | 26 -------------------- 1 file changed, 26 deletions(-) delete mode 100644 src/test/java/PortfolioValueAccumulator.java diff --git a/src/test/java/PortfolioValueAccumulator.java b/src/test/java/PortfolioValueAccumulator.java deleted file mode 100644 index da7948ea7..000000000 --- a/src/test/java/PortfolioValueAccumulator.java +++ /dev/null @@ -1,26 +0,0 @@ -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> -{ - 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 From 44b2b0aa8b035a57805ce7630756f2045f316e73 Mon Sep 17 00:00:00 2001 From: afbarbaro Date: Mon, 4 Apr 2016 23:28:01 -0500 Subject: [PATCH 6/8] Delete PortfolioValueTest.java --- src/test/java/PortfolioValueTest.java | 163 -------------------------- 1 file changed, 163 deletions(-) delete mode 100644 src/test/java/PortfolioValueTest.java diff --git a/src/test/java/PortfolioValueTest.java b/src/test/java/PortfolioValueTest.java deleted file mode 100644 index 384a0f2f2..000000000 --- a/src/test/java/PortfolioValueTest.java +++ /dev/null @@ -1,163 +0,0 @@ -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 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; - } -} \ No newline at end of file From 6df4782bdb0f88647479509a24c7472dd73e9068 Mon Sep 17 00:00:00 2001 From: afbarbaro Date: Mon, 4 Apr 2016 23:28:46 -0500 Subject: [PATCH 7/8] Create PortfolioValueTest.java --- .../portfolio/PortfolioValueTest.java | 163 ++++++++++++++++++ 1 file changed, 163 insertions(+) create mode 100644 src/test/java/examples/portfolio/PortfolioValueTest.java 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; + } +} From 4e77615f4fda716a0275ec3b4acd710a3d6c50a5 Mon Sep 17 00:00:00 2001 From: afbarbaro Date: Mon, 4 Apr 2016 23:29:00 -0500 Subject: [PATCH 8/8] Added files via upload --- .../portfolio/PortfolioAssetInterface.java | 13 ++++++++++ .../portfolio/PortfolioValueAccumulator.java | 26 +++++++++++++++++++ 2 files changed, 39 insertions(+) create mode 100644 src/test/java/examples/portfolio/PortfolioAssetInterface.java create mode 100644 src/test/java/examples/portfolio/PortfolioValueAccumulator.java 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