From 9e98bcc4edeb1c773d02dd2b28666b0bfa973955 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Thu, 24 Jul 2014 15:19:30 +0100 Subject: [PATCH] HCOLL-119 removed the use of " public SharedHashMap create(File file, Class kClass, Class vClass) " --- .../collections/ConcurrentHashMap.java | 834 +----------------- .../openhft/collections/VanillaSharedMap.java | 794 ++++++++++++++++- 2 files changed, 792 insertions(+), 836 deletions(-) diff --git a/src/main/java/net/openhft/collections/ConcurrentHashMap.java b/src/main/java/net/openhft/collections/ConcurrentHashMap.java index bc6ad8f2d..8cbe16c5c 100644 --- a/src/main/java/net/openhft/collections/ConcurrentHashMap.java +++ b/src/main/java/net/openhft/collections/ConcurrentHashMap.java @@ -1470,133 +1470,10 @@ public class ConcurrentHashMap extends AbstractMap return (v = get(key)) == null ? defaultValue : v; } - /* public void forEach(BiConsumer action) { - if (action == null) throw new NullPointerException(); - Node[] t; - if ((t = table) != null) { - Traverser it = new Traverser(t, t.length, 0, t.length); - for (Node p; (p = it.advance()) != null; ) { - action.accept(p.key, p.val); - } - } - }*/ - /* public void replaceAll(BiFunction function) { - if (function == null) throw new NullPointerException(); - Node[] t; - if ((t = table) != null) { - Traverser it = new Traverser(t, t.length, 0, t.length); - for (Node p; (p = it.advance()) != null; ) { - V oldValue = p.val; - for (K key = p.key; ; ) { - V newValue = function.apply(key, oldValue); - if (newValue == null) - throw new NullPointerException(); - if (replaceNode(key, newValue, oldValue) != null || - (oldValue = get(key)) == null) - break; - } - } - } - }*/ - /** - * If the specified key is not already associated with a value, attempts to compute its value using the - * given mapping function and enters it into this map unless {@code null}. The entire method invocation - * is performed atomically, so the function is applied at most once per key. Some attempted update - * operations on this map by other threads may be blocked while computation is in progress, so the - * computation should be short and simple, and must not attempt to update any other mappings of this map. - * - * @param key key with which the specified value is to be associated - * @param mappingFunction the function to compute a value - * @return the current (existing or computed) value associated with the specified key, or null if the - * computed value is null - * @throws NullPointerException if the specified key or mappingFunction is null - * @throws IllegalStateException if the computation detectably attempts a recursive update to this map - * that would otherwise never complete - * @throws RuntimeException or Error if the mappingFunction does so, in which case the mapping is - * left unestablished - */ - public V computeIfAbsent(K key, Function mappingFunction) { - if (key == null || mappingFunction == null) - throw new NullPointerException(); - int h = spread(key.hashCode()); - V val = null; - int binCount = 0; - for (Node[] tab = table; ; ) { - Node f; - int n, i, fh; - if (tab == null || (n = tab.length) == 0) - tab = initTable(); - else if ((f = tabAt(tab, i = (n - 1) & h)) == null) { - Node r = new ReservationNode(); - synchronized (r) { - if (casTabAt(tab, i, null, r)) { - binCount = 1; - Node node = null; - try { - if ((val = mappingFunction.apply(key)) != null) - node = new Node(h, key, val, null); - } finally { - setTabAt(tab, i, node); - } - } - } - if (binCount != 0) - break; - } else if ((fh = f.hash) == MOVED) - tab = helpTransfer(tab, f); - else { - boolean added = false; - synchronized (f) { - if (tabAt(tab, i) == f) { - if (fh >= 0) { - binCount = 1; - for (Node e = f; ; ++binCount) { - K ek; - V ev; - if (e.hash == h && - ((ek = e.key) == key || - (ek != null && key.equals(ek)))) { - val = e.val; - break; - } - Node pred = e; - if ((e = e.next) == null) { - if ((val = mappingFunction.apply(key)) != null) { - added = true; - pred.next = new Node(h, key, val, null); - } - break; - } - } - } else if (f instanceof TreeBin) { - binCount = 2; - TreeBin t = (TreeBin) f; - TreeNode r, p; - if ((r = t.root) != null && - (p = r.findTreeNode(h, key, null)) != null) - val = p.val; - else if ((val = mappingFunction.apply(key)) != null) { - added = true; - t.putTreeVal(h, key, val); - } - } - } - } - if (binCount != 0) { - if (binCount >= TREEIFY_THRESHOLD) - treeifyBin(tab, i); - if (!added) - return val; - break; - } - } - } - if (val != null) - addCount(1L, binCount); - return val; - } + + /** * If the value for the specified key is present, attempts to compute a new mapping given the key and its @@ -3534,669 +3411,10 @@ public class ConcurrentHashMap extends AbstractMap // Parallel bulk operations - /** - * Computes initial batch value for bulk tasks. The returned value is approximately exp2 of the number of - * times (minus one) to split task by two before executing leaf action. This value is faster to compute - * and more convenient to use as a guide to splitting than is the depth, since it is used while dividing - * by two anyway. - */ - final int batchFor(long b) { - long n; - if (b == Long.MAX_VALUE || (n = sumCount()) <= 1L || n < b) - return 0; - int sp = ForkJoinPool.getCommonPoolParallelism() << 2; // slack of 4 - return (b <= 0L || (n /= b) >= sp) ? sp : (int) n; - } - /** - * Performs the given action for each (key, value). - * - * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed - * in parallel - * @param action the action - * @since 1.8 - */ - public void forEach(long parallelismThreshold, - BiConsumer action) { - if (action == null) throw new NullPointerException(); - new ForEachMappingTask - (null, batchFor(parallelismThreshold), 0, 0, table, - action).invoke(); - } - /** - * Performs the given action for each non-null transformation of each (key, value). - * - * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed - * in parallel - * @param transformer a function returning the transformation for an element, or null if there is - * no transformation (in which case the action is not applied) - * @param action the action - * @param the return type of the transformer - * @since 1.8 - */ - public void forEach(long parallelismThreshold, - BiFunction transformer, - Consumer action) { - if (transformer == null || action == null) - throw new NullPointerException(); - new ForEachTransformedMappingTask - (null, batchFor(parallelismThreshold), 0, 0, table, - transformer, action).invoke(); - } - /** - * Returns a non-null result from applying the given search function on each (key, value), or null if - * none. Upon success, further element processing is suppressed and the results of any other parallel - * invocations of the search function are ignored. - * - * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed - * in parallel - * @param searchFunction a function returning a non-null result on success, else null - * @param the return type of the search function - * @return a non-null result from applying the given search function on each (key, value), or null if none - * @since 1.8 - */ - public U search(long parallelismThreshold, - BiFunction searchFunction) { - if (searchFunction == null) throw new NullPointerException(); - return new SearchMappingsTask - (null, batchFor(parallelismThreshold), 0, 0, table, - searchFunction, new AtomicReference()).invoke(); - } - /** - * Returns the result of accumulating the given transformation of all (key, value) pairs using the given - * reducer to combine values, or null if none. - * - * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed - * in parallel - * @param transformer a function returning the transformation for an element, or null if there is - * no transformation (in which case it is not combined) - * @param reducer a commutative associative combining function - * @param the return type of the transformer - * @return the result of accumulating the given transformation of all (key, value) pairs - * @since 1.8 - */ - public U reduce(long parallelismThreshold, - BiFunction transformer, - BiFunction reducer) { - if (transformer == null || reducer == null) - throw new NullPointerException(); - return new MapReduceMappingsTask - (null, batchFor(parallelismThreshold), 0, 0, table, - null, transformer, reducer).invoke(); - } - - /** - * Returns the result of accumulating the given transformation of all (key, value) pairs using the given - * reducer to combine values, and the given basis as an identity value. - * - * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed - * in parallel - * @param transformer a function returning the transformation for an element - * @param basis the identity (initial default value) for the reduction - * @param reducer a commutative associative combining function - * @return the result of accumulating the given transformation of all (key, value) pairs - * @since 1.8 - */ - public double reduceToDouble(long parallelismThreshold, - ToDoubleBiFunction transformer, - double basis, - DoubleBinaryOperator reducer) { - if (transformer == null || reducer == null) - throw new NullPointerException(); - return new MapReduceMappingsToDoubleTask - (null, batchFor(parallelismThreshold), 0, 0, table, - null, transformer, basis, reducer).invoke(); - } - - /** - * Returns the result of accumulating the given transformation of all (key, value) pairs using the given - * reducer to combine values, and the given basis as an identity value. - * - * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed - * in parallel - * @param transformer a function returning the transformation for an element - * @param basis the identity (initial default value) for the reduction - * @param reducer a commutative associative combining function - * @return the result of accumulating the given transformation of all (key, value) pairs - * @since 1.8 - */ - public long reduceToLong(long parallelismThreshold, - ToLongBiFunction transformer, - long basis, - LongBinaryOperator reducer) { - if (transformer == null || reducer == null) - throw new NullPointerException(); - return new MapReduceMappingsToLongTask - (null, batchFor(parallelismThreshold), 0, 0, table, - null, transformer, basis, reducer).invoke(); - } - - /** - * Returns the result of accumulating the given transformation of all (key, value) pairs using the given - * reducer to combine values, and the given basis as an identity value. - * - * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed - * in parallel - * @param transformer a function returning the transformation for an element - * @param basis the identity (initial default value) for the reduction - * @param reducer a commutative associative combining function - * @return the result of accumulating the given transformation of all (key, value) pairs - * @since 1.8 - */ - public int reduceToInt(long parallelismThreshold, - ToIntBiFunction transformer, - int basis, - IntBinaryOperator reducer) { - if (transformer == null || reducer == null) - throw new NullPointerException(); - return new MapReduceMappingsToIntTask - (null, batchFor(parallelismThreshold), 0, 0, table, - null, transformer, basis, reducer).invoke(); - } - - /** - * Performs the given action for each key. - * - * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed - * in parallel - * @param action the action - * @since 1.8 - */ - public void forEachKey(long parallelismThreshold, - Consumer action) { - if (action == null) throw new NullPointerException(); - new ForEachKeyTask - (null, batchFor(parallelismThreshold), 0, 0, table, - action).invoke(); - } - - /** - * Performs the given action for each non-null transformation of each key. - * - * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed - * in parallel - * @param transformer a function returning the transformation for an element, or null if there is - * no transformation (in which case the action is not applied) - * @param action the action - * @param the return type of the transformer - * @since 1.8 - */ - public void forEachKey(long parallelismThreshold, - Function transformer, - Consumer action) { - if (transformer == null || action == null) - throw new NullPointerException(); - new ForEachTransformedKeyTask - (null, batchFor(parallelismThreshold), 0, 0, table, - transformer, action).invoke(); - } - - /** - * Returns a non-null result from applying the given search function on each key, or null if none. Upon - * success, further element processing is suppressed and the results of any other parallel invocations of - * the search function are ignored. - * - * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed - * in parallel - * @param searchFunction a function returning a non-null result on success, else null - * @param the return type of the search function - * @return a non-null result from applying the given search function on each key, or null if none - * @since 1.8 - */ - public U searchKeys(long parallelismThreshold, - Function searchFunction) { - if (searchFunction == null) throw new NullPointerException(); - return new SearchKeysTask - (null, batchFor(parallelismThreshold), 0, 0, table, - searchFunction, new AtomicReference()).invoke(); - } - - /** - * Returns the result of accumulating all keys using the given reducer to combine values, or null if - * none. - * - * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed - * in parallel - * @param reducer a commutative associative combining function - * @return the result of accumulating all keys using the given reducer to combine values, or null if none - * @since 1.8 - */ - public K reduceKeys(long parallelismThreshold, - BiFunction reducer) { - if (reducer == null) throw new NullPointerException(); - return new ReduceKeysTask - (null, batchFor(parallelismThreshold), 0, 0, table, - null, reducer).invoke(); - } - - /** - * Returns the result of accumulating the given transformation of all keys using the given reducer to - * combine values, or null if none. - * - * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed - * in parallel - * @param transformer a function returning the transformation for an element, or null if there is - * no transformation (in which case it is not combined) - * @param reducer a commutative associative combining function - * @param the return type of the transformer - * @return the result of accumulating the given transformation of all keys - * @since 1.8 - */ - public U reduceKeys(long parallelismThreshold, - Function transformer, - BiFunction reducer) { - if (transformer == null || reducer == null) - throw new NullPointerException(); - return new MapReduceKeysTask - (null, batchFor(parallelismThreshold), 0, 0, table, - null, transformer, reducer).invoke(); - } - - /** - * Returns the result of accumulating the given transformation of all keys using the given reducer to - * combine values, and the given basis as an identity value. - * - * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed - * in parallel - * @param transformer a function returning the transformation for an element - * @param basis the identity (initial default value) for the reduction - * @param reducer a commutative associative combining function - * @return the result of accumulating the given transformation of all keys - * @since 1.8 - */ - public double reduceKeysToDouble(long parallelismThreshold, - ToDoubleFunction transformer, - double basis, - DoubleBinaryOperator reducer) { - if (transformer == null || reducer == null) - throw new NullPointerException(); - return new MapReduceKeysToDoubleTask - (null, batchFor(parallelismThreshold), 0, 0, table, - null, transformer, basis, reducer).invoke(); - } - - /** - * Returns the result of accumulating the given transformation of all keys using the given reducer to - * combine values, and the given basis as an identity value. - * - * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed - * in parallel - * @param transformer a function returning the transformation for an element - * @param basis the identity (initial default value) for the reduction - * @param reducer a commutative associative combining function - * @return the result of accumulating the given transformation of all keys - * @since 1.8 - */ - public long reduceKeysToLong(long parallelismThreshold, - ToLongFunction transformer, - long basis, - LongBinaryOperator reducer) { - if (transformer == null || reducer == null) - throw new NullPointerException(); - return new MapReduceKeysToLongTask - (null, batchFor(parallelismThreshold), 0, 0, table, - null, transformer, basis, reducer).invoke(); - } - - /** - * Returns the result of accumulating the given transformation of all keys using the given reducer to - * combine values, and the given basis as an identity value. - * - * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed - * in parallel - * @param transformer a function returning the transformation for an element - * @param basis the identity (initial default value) for the reduction - * @param reducer a commutative associative combining function - * @return the result of accumulating the given transformation of all keys - * @since 1.8 - */ - public int reduceKeysToInt(long parallelismThreshold, - ToIntFunction transformer, - int basis, - IntBinaryOperator reducer) { - if (transformer == null || reducer == null) - throw new NullPointerException(); - return new MapReduceKeysToIntTask - (null, batchFor(parallelismThreshold), 0, 0, table, - null, transformer, basis, reducer).invoke(); - } - - /** - * Performs the given action for each value. - * - * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed - * in parallel - * @param action the action - * @since 1.8 - */ - public void forEachValue(long parallelismThreshold, - Consumer action) { - if (action == null) - throw new NullPointerException(); - new ForEachValueTask - (null, batchFor(parallelismThreshold), 0, 0, table, - action).invoke(); - } - - /** - * Performs the given action for each non-null transformation of each value. - * - * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed - * in parallel - * @param transformer a function returning the transformation for an element, or null if there is - * no transformation (in which case the action is not applied) - * @param action the action - * @param the return type of the transformer - * @since 1.8 - */ - public void forEachValue(long parallelismThreshold, - Function transformer, - Consumer action) { - if (transformer == null || action == null) - throw new NullPointerException(); - new ForEachTransformedValueTask - (null, batchFor(parallelismThreshold), 0, 0, table, - transformer, action).invoke(); - } - - /** - * Returns a non-null result from applying the given search function on each value, or null if none. Upon - * success, further element processing is suppressed and the results of any other parallel invocations of - * the search function are ignored. - * - * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed - * in parallel - * @param searchFunction a function returning a non-null result on success, else null - * @param the return type of the search function - * @return a non-null result from applying the given search function on each value, or null if none - * @since 1.8 - */ - public U searchValues(long parallelismThreshold, - Function searchFunction) { - if (searchFunction == null) throw new NullPointerException(); - return new SearchValuesTask - (null, batchFor(parallelismThreshold), 0, 0, table, - searchFunction, new AtomicReference()).invoke(); - } - - /** - * Returns the result of accumulating all values using the given reducer to combine values, or null if - * none. - * - * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed - * in parallel - * @param reducer a commutative associative combining function - * @return the result of accumulating all values - * @since 1.8 - */ - public V reduceValues(long parallelismThreshold, - BiFunction reducer) { - if (reducer == null) throw new NullPointerException(); - return new ReduceValuesTask - (null, batchFor(parallelismThreshold), 0, 0, table, - null, reducer).invoke(); - } - - /** - * Returns the result of accumulating the given transformation of all values using the given reducer to - * combine values, or null if none. - * - * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed - * in parallel - * @param transformer a function returning the transformation for an element, or null if there is - * no transformation (in which case it is not combined) - * @param reducer a commutative associative combining function - * @param the return type of the transformer - * @return the result of accumulating the given transformation of all values - * @since 1.8 - */ - public U reduceValues(long parallelismThreshold, - Function transformer, - BiFunction reducer) { - if (transformer == null || reducer == null) - throw new NullPointerException(); - return new MapReduceValuesTask - (null, batchFor(parallelismThreshold), 0, 0, table, - null, transformer, reducer).invoke(); - } - - /** - * Returns the result of accumulating the given transformation of all values using the given reducer to - * combine values, and the given basis as an identity value. - * - * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed - * in parallel - * @param transformer a function returning the transformation for an element - * @param basis the identity (initial default value) for the reduction - * @param reducer a commutative associative combining function - * @return the result of accumulating the given transformation of all values - * @since 1.8 - */ - public double reduceValuesToDouble(long parallelismThreshold, - ToDoubleFunction transformer, - double basis, - DoubleBinaryOperator reducer) { - if (transformer == null || reducer == null) - throw new NullPointerException(); - return new MapReduceValuesToDoubleTask - (null, batchFor(parallelismThreshold), 0, 0, table, - null, transformer, basis, reducer).invoke(); - } - - /** - * Returns the result of accumulating the given transformation of all values using the given reducer to - * combine values, and the given basis as an identity value. - * - * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed - * in parallel - * @param transformer a function returning the transformation for an element - * @param basis the identity (initial default value) for the reduction - * @param reducer a commutative associative combining function - * @return the result of accumulating the given transformation of all values - * @since 1.8 - */ - public long reduceValuesToLong(long parallelismThreshold, - ToLongFunction transformer, - long basis, - LongBinaryOperator reducer) { - if (transformer == null || reducer == null) - throw new NullPointerException(); - return new MapReduceValuesToLongTask - (null, batchFor(parallelismThreshold), 0, 0, table, - null, transformer, basis, reducer).invoke(); - } - - /** - * Returns the result of accumulating the given transformation of all values using the given reducer to - * combine values, and the given basis as an identity value. - * - * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed - * in parallel - * @param transformer a function returning the transformation for an element - * @param basis the identity (initial default value) for the reduction - * @param reducer a commutative associative combining function - * @return the result of accumulating the given transformation of all values - * @since 1.8 - */ - public int reduceValuesToInt(long parallelismThreshold, - ToIntFunction transformer, - int basis, - IntBinaryOperator reducer) { - if (transformer == null || reducer == null) - throw new NullPointerException(); - return new MapReduceValuesToIntTask - (null, batchFor(parallelismThreshold), 0, 0, table, - null, transformer, basis, reducer).invoke(); - } - - /** - * Performs the given action for each entry. - * - * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed - * in parallel - * @param action the action - * @since 1.8 - */ - public void forEachEntry(long parallelismThreshold, - Consumer> action) { - if (action == null) throw new NullPointerException(); - new ForEachEntryTask(null, batchFor(parallelismThreshold), 0, 0, table, - action).invoke(); - } - - /** - * Performs the given action for each non-null transformation of each entry. - * - * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed - * in parallel - * @param transformer a function returning the transformation for an element, or null if there is - * no transformation (in which case the action is not applied) - * @param action the action - * @param the return type of the transformer - * @since 1.8 - */ - public void forEachEntry(long parallelismThreshold, - Function, ? extends U> transformer, - Consumer action) { - if (transformer == null || action == null) - throw new NullPointerException(); - new ForEachTransformedEntryTask - (null, batchFor(parallelismThreshold), 0, 0, table, - transformer, action).invoke(); - } - - /** - * Returns a non-null result from applying the given search function on each entry, or null if none. Upon - * success, further element processing is suppressed and the results of any other parallel invocations of - * the search function are ignored. - * - * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed - * in parallel - * @param searchFunction a function returning a non-null result on success, else null - * @param the return type of the search function - * @return a non-null result from applying the given search function on each entry, or null if none - * @since 1.8 - */ - public U searchEntries(long parallelismThreshold, - Function, ? extends U> searchFunction) { - if (searchFunction == null) throw new NullPointerException(); - return new SearchEntriesTask - (null, batchFor(parallelismThreshold), 0, 0, table, - searchFunction, new AtomicReference()).invoke(); - } - - /** - * Returns the result of accumulating all entries using the given reducer to combine values, or null if - * none. - * - * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed - * in parallel - * @param reducer a commutative associative combining function - * @return the result of accumulating all entries - * @since 1.8 - */ - public Map.Entry reduceEntries(long parallelismThreshold, - BiFunction, Map.Entry, ? extends Map.Entry> reducer) { - if (reducer == null) throw new NullPointerException(); - return new ReduceEntriesTask - (null, batchFor(parallelismThreshold), 0, 0, table, - null, reducer).invoke(); - } - - /** - * Returns the result of accumulating the given transformation of all entries using the given reducer to - * combine values, or null if none. - * - * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed - * in parallel - * @param transformer a function returning the transformation for an element, or null if there is - * no transformation (in which case it is not combined) - * @param reducer a commutative associative combining function - * @param the return type of the transformer - * @return the result of accumulating the given transformation of all entries - * @since 1.8 - */ - public U reduceEntries(long parallelismThreshold, - Function, ? extends U> transformer, - BiFunction reducer) { - if (transformer == null || reducer == null) - throw new NullPointerException(); - return new MapReduceEntriesTask - (null, batchFor(parallelismThreshold), 0, 0, table, - null, transformer, reducer).invoke(); - } - - /** - * Returns the result of accumulating the given transformation of all entries using the given reducer to - * combine values, and the given basis as an identity value. - * - * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed - * in parallel - * @param transformer a function returning the transformation for an element - * @param basis the identity (initial default value) for the reduction - * @param reducer a commutative associative combining function - * @return the result of accumulating the given transformation of all entries - * @since 1.8 - */ - public double reduceEntriesToDouble(long parallelismThreshold, - ToDoubleFunction> transformer, - double basis, - DoubleBinaryOperator reducer) { - if (transformer == null || reducer == null) - throw new NullPointerException(); - return new MapReduceEntriesToDoubleTask - (null, batchFor(parallelismThreshold), 0, 0, table, - null, transformer, basis, reducer).invoke(); - } - - /** - * Returns the result of accumulating the given transformation of all entries using the given reducer to - * combine values, and the given basis as an identity value. - * - * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed - * in parallel - * @param transformer a function returning the transformation for an element - * @param basis the identity (initial default value) for the reduction - * @param reducer a commutative associative combining function - * @return the result of accumulating the given transformation of all entries - * @since 1.8 - */ - public long reduceEntriesToLong(long parallelismThreshold, - ToLongFunction> transformer, - long basis, - LongBinaryOperator reducer) { - if (transformer == null || reducer == null) - throw new NullPointerException(); - return new MapReduceEntriesToLongTask - (null, batchFor(parallelismThreshold), 0, 0, table, - null, transformer, basis, reducer).invoke(); - } - - /** - * Returns the result of accumulating the given transformation of all entries using the given reducer to - * combine values, and the given basis as an identity value. - * - * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed - * in parallel - * @param transformer a function returning the transformation for an element - * @param basis the identity (initial default value) for the reduction - * @param reducer a commutative associative combining function - * @return the result of accumulating the given transformation of all entries - * @since 1.8 - */ - public int reduceEntriesToInt(long parallelismThreshold, - ToIntFunction> transformer, - int basis, - IntBinaryOperator reducer) { - if (transformer == null || reducer == null) - throw new NullPointerException(); - return new MapReduceEntriesToIntTask - (null, batchFor(parallelismThreshold), 0, 0, table, - null, transformer, basis, reducer).invoke(); - } /* ----------------Views -------------- */ @@ -5165,55 +4383,7 @@ public class ConcurrentHashMap extends AbstractMap } } - @SuppressWarnings("serial") - static final class SearchMappingsTask - extends BulkTask { - final BiFunction searchFunction; - final AtomicReference result; - SearchMappingsTask - (BulkTask p, int b, int i, int f, Node[] t, - BiFunction searchFunction, - AtomicReference result) { - super(p, b, i, f, t); - this.searchFunction = searchFunction; - this.result = result; - } - - public final U getRawResult() { - return result.get(); - } - - public final void compute() { - final BiFunction searchFunction; - final AtomicReference result; - if ((searchFunction = this.searchFunction) != null && - (result = this.result) != null) { - for (int i = baseIndex, f, h; batch > 0 && - (h = ((f = baseLimit) + i) >>> 1) > i; ) { - if (result.get() != null) - return; - addToPendingCount(1); - new SearchMappingsTask - (this, batch >>>= 1, baseLimit = h, f, tab, - searchFunction, result).fork(); - } - while (result.get() == null) { - U u; - Node p; - if ((p = advance()) == null) { - propagateCompletion(); - break; - } - if ((u = searchFunction.apply(p.key, p.val)) != null) { - if (result.compareAndSet(null, u)) - quietlyCompleteRoot(); - break; - } - } - } - } - } @SuppressWarnings("serial") static final class ReduceKeysTask diff --git a/src/main/java/net/openhft/collections/VanillaSharedMap.java b/src/main/java/net/openhft/collections/VanillaSharedMap.java index 470428838..0b6342658 100644 --- a/src/main/java/net/openhft/collections/VanillaSharedMap.java +++ b/src/main/java/net/openhft/collections/VanillaSharedMap.java @@ -38,7 +38,9 @@ import java.io.IOException; import java.lang.reflect.Array; import java.nio.channels.FileChannel; import java.util.*; -import java.util.function.BiConsumer; +import java.util.concurrent.ForkJoinPool; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.*; import static java.lang.Thread.currentThread; @@ -83,6 +85,765 @@ abstract class AbstractVanillaSharedMap extends net.openhft.collections.Co } + /** + * If the specified key is not already associated with a value, attempts to compute its value using the + * given mapping function and enters it into this map unless {@code null}. The entire method invocation + * is performed atomically, so the function is applied at most once per key. Some attempted update + * operations on this map by other threads may be blocked while computation is in progress, so the + * computation should be short and simple, and must not attempt to update any other mappings of this map. + * + * @param key key with which the specified value is to be associated + * @param mappingFunction the function to compute a value + * @return the current (existing or computed) value associated with the specified key, or null if the + * computed value is null + * @throws NullPointerException if the specified key or mappingFunction is null + * @throws IllegalStateException if the computation detectably attempts a recursive update to this map + * that would otherwise never complete + * @throws RuntimeException or Error if the mappingFunction does so, in which case the mapping is + * left unestablished + */ + public V computeIfAbsent(K key, Function mappingFunction) { + checkKey(key); + + Bytes keyBytes = getKeyAsBytes(key); + long hash = Hasher.hash(keyBytes); + int segmentNum = hasher.getSegment(hash); + int segmentHash = hasher.segmentHash(hash); + return segments[segmentNum].put(keyBytes, key, null, segmentHash, false, mappingFunction); + + } + + + /** + * Returns the value to which the specified key is mapped, or the given default value if this map contains + * no mapping for the key. + * + * @param key the key whose associated value is to be returned + * @param defaultValue the value to return if this map contains no mapping for the given key + * @return the mapping for the key, if present; else the default value + * @throws NullPointerException if the specified key is null + */ + public V getOrDefault(Object key, V defaultValue) { + V v; + return (v = get(key)) == null ? defaultValue : v; + } + + /** + * Computes initial batch value for bulk tasks. The returned value is approximately exp2 of the number of + * times (minus one) to split task by two before executing leaf action. This value is faster to compute + * and more convenient to use as a guide to splitting than is the depth, since it is used while dividing + * by two anyway. + */ + final int batchFor(long b) { + long n; + if (b == Long.MAX_VALUE || (n = sumCount()) <= 1L || n < b) + return 0; + int sp = ForkJoinPool.getCommonPoolParallelism() << 2; // slack of 4 + return (b <= 0L || (n /= b) >= sp) ? sp : (int) n; + } + + + /** + * Performs the given action for each non-null transformation of each (key, value). + * + * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed + * in parallel + * @param transformer a function returning the transformation for an element, or null if there is + * no transformation (in which case the action is not applied) + * @param action the action + * @param the return type of the transformer + * @since 1.8 + */ + public void forEach(long parallelismThreshold, + BiFunction transformer, + Consumer action) { + if (transformer == null || action == null) + throw new NullPointerException(); + new ForEachTransformedMappingTask + (null, batchFor(parallelismThreshold), 0, 0, table, + transformer, action).invoke(); + } + + /** + * Returns a non-null result from applying the given search function on each (key, value), or null if + * none. Upon success, further element processing is suppressed and the results of any other parallel + * invocations of the search function are ignored. + * + * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed + * in parallel + * @param searchFunction a function returning a non-null result on success, else null + * @param the return type of the search function + * @return a non-null result from applying the given search function on each (key, value), or null if none + * @since 1.8 + */ + public U search(long parallelismThreshold, + BiFunction searchFunction) { + if (searchFunction == null) throw new NullPointerException(); + return new SearchMappingsTask + (null, batchFor(parallelismThreshold), 0, 0, table, + searchFunction, new AtomicReference()).invoke(); + } + + /** + * Returns the result of accumulating the given transformation of all (key, value) pairs using the given + * reducer to combine values, or null if none. + * + * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed + * in parallel + * @param transformer a function returning the transformation for an element, or null if there is + * no transformation (in which case it is not combined) + * @param reducer a commutative associative combining function + * @param the return type of the transformer + * @return the result of accumulating the given transformation of all (key, value) pairs + * @since 1.8 + */ + public U reduce(long parallelismThreshold, + BiFunction transformer, + BiFunction reducer) { + if (transformer == null || reducer == null) + throw new NullPointerException(); + return new MapReduceMappingsTask + (null, batchFor(parallelismThreshold), 0, 0, table, + null, transformer, reducer).invoke(); + } + + /** + * Returns the result of accumulating the given transformation of all (key, value) pairs using the given + * reducer to combine values, and the given basis as an identity value. + * + * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed + * in parallel + * @param transformer a function returning the transformation for an element + * @param basis the identity (initial default value) for the reduction + * @param reducer a commutative associative combining function + * @return the result of accumulating the given transformation of all (key, value) pairs + * @since 1.8 + */ + public double reduceToDouble(long parallelismThreshold, + ToDoubleBiFunction transformer, + double basis, + DoubleBinaryOperator reducer) { + if (transformer == null || reducer == null) + throw new NullPointerException(); + return new MapReduceMappingsToDoubleTask + (null, batchFor(parallelismThreshold), 0, 0, table, + null, transformer, basis, reducer).invoke(); + } + + /** + * Returns the result of accumulating the given transformation of all (key, value) pairs using the given + * reducer to combine values, and the given basis as an identity value. + * + * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed + * in parallel + * @param transformer a function returning the transformation for an element + * @param basis the identity (initial default value) for the reduction + * @param reducer a commutative associative combining function + * @return the result of accumulating the given transformation of all (key, value) pairs + * @since 1.8 + */ + public long reduceToLong(long parallelismThreshold, + ToLongBiFunction transformer, + long basis, + LongBinaryOperator reducer) { + if (transformer == null || reducer == null) + throw new NullPointerException(); + return new MapReduceMappingsToLongTask + (null, batchFor(parallelismThreshold), 0, 0, table, + null, transformer, basis, reducer).invoke(); + } + + /** + * Returns the result of accumulating the given transformation of all (key, value) pairs using the given + * reducer to combine values, and the given basis as an identity value. + * + * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed + * in parallel + * @param transformer a function returning the transformation for an element + * @param basis the identity (initial default value) for the reduction + * @param reducer a commutative associative combining function + * @return the result of accumulating the given transformation of all (key, value) pairs + * @since 1.8 + */ + public int reduceToInt(long parallelismThreshold, + ToIntBiFunction transformer, + int basis, + IntBinaryOperator reducer) { + if (transformer == null || reducer == null) + throw new NullPointerException(); + return new MapReduceMappingsToIntTask + (null, batchFor(parallelismThreshold), 0, 0, table, + null, transformer, basis, reducer).invoke(); + } + + /** + * Performs the given action for each key. + * + * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed + * in parallel + * @param action the action + * @since 1.8 + */ + public void forEachKey(long parallelismThreshold, + Consumer action) { + if (action == null) throw new NullPointerException(); + new ForEachKeyTask + (null, batchFor(parallelismThreshold), 0, 0, table, + action).invoke(); + } + + /** + * Performs the given action for each non-null transformation of each key. + * + * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed + * in parallel + * @param transformer a function returning the transformation for an element, or null if there is + * no transformation (in which case the action is not applied) + * @param action the action + * @param the return type of the transformer + * @since 1.8 + */ + public void forEachKey(long parallelismThreshold, + Function transformer, + Consumer action) { + if (transformer == null || action == null) + throw new NullPointerException(); + new ForEachTransformedKeyTask + (null, batchFor(parallelismThreshold), 0, 0, table, + transformer, action).invoke(); + } + + /** + * Returns a non-null result from applying the given search function on each key, or null if none. Upon + * success, further element processing is suppressed and the results of any other parallel invocations of + * the search function are ignored. + * + * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed + * in parallel + * @param searchFunction a function returning a non-null result on success, else null + * @param the return type of the search function + * @return a non-null result from applying the given search function on each key, or null if none + * @since 1.8 + */ + public U searchKeys(long parallelismThreshold, + Function searchFunction) { + if (searchFunction == null) throw new NullPointerException(); + return new SearchKeysTask + (null, batchFor(parallelismThreshold), 0, 0, table, + searchFunction, new AtomicReference()).invoke(); + } + + /** + * Returns the result of accumulating all keys using the given reducer to combine values, or null if + * none. + * + * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed + * in parallel + * @param reducer a commutative associative combining function + * @return the result of accumulating all keys using the given reducer to combine values, or null if none + * @since 1.8 + */ + public K reduceKeys(long parallelismThreshold, + BiFunction reducer) { + if (reducer == null) throw new NullPointerException(); + return new ReduceKeysTask + (null, batchFor(parallelismThreshold), 0, 0, table, + null, reducer).invoke(); + } + + /** + * Returns the result of accumulating the given transformation of all keys using the given reducer to + * combine values, or null if none. + * + * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed + * in parallel + * @param transformer a function returning the transformation for an element, or null if there is + * no transformation (in which case it is not combined) + * @param reducer a commutative associative combining function + * @param the return type of the transformer + * @return the result of accumulating the given transformation of all keys + * @since 1.8 + */ + public U reduceKeys(long parallelismThreshold, + Function transformer, + BiFunction reducer) { + if (transformer == null || reducer == null) + throw new NullPointerException(); + return new MapReduceKeysTask + (null, batchFor(parallelismThreshold), 0, 0, table, + null, transformer, reducer).invoke(); + } + + /** + * Returns the result of accumulating the given transformation of all keys using the given reducer to + * combine values, and the given basis as an identity value. + * + * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed + * in parallel + * @param transformer a function returning the transformation for an element + * @param basis the identity (initial default value) for the reduction + * @param reducer a commutative associative combining function + * @return the result of accumulating the given transformation of all keys + * @since 1.8 + */ + public double reduceKeysToDouble(long parallelismThreshold, + ToDoubleFunction transformer, + double basis, + DoubleBinaryOperator reducer) { + if (transformer == null || reducer == null) + throw new NullPointerException(); + return new MapReduceKeysToDoubleTask + (null, batchFor(parallelismThreshold), 0, 0, table, + null, transformer, basis, reducer).invoke(); + } + + /** + * Returns the result of accumulating the given transformation of all keys using the given reducer to + * combine values, and the given basis as an identity value. + * + * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed + * in parallel + * @param transformer a function returning the transformation for an element + * @param basis the identity (initial default value) for the reduction + * @param reducer a commutative associative combining function + * @return the result of accumulating the given transformation of all keys + * @since 1.8 + */ + public long reduceKeysToLong(long parallelismThreshold, + ToLongFunction transformer, + long basis, + LongBinaryOperator reducer) { + if (transformer == null || reducer == null) + throw new NullPointerException(); + return new MapReduceKeysToLongTask + (null, batchFor(parallelismThreshold), 0, 0, table, + null, transformer, basis, reducer).invoke(); + } + + /** + * Returns the result of accumulating the given transformation of all keys using the given reducer to + * combine values, and the given basis as an identity value. + * + * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed + * in parallel + * @param transformer a function returning the transformation for an element + * @param basis the identity (initial default value) for the reduction + * @param reducer a commutative associative combining function + * @return the result of accumulating the given transformation of all keys + * @since 1.8 + */ + public int reduceKeysToInt(long parallelismThreshold, + ToIntFunction transformer, + int basis, + IntBinaryOperator reducer) { + if (transformer == null || reducer == null) + throw new NullPointerException(); + return new MapReduceKeysToIntTask + (null, batchFor(parallelismThreshold), 0, 0, table, + null, transformer, basis, reducer).invoke(); + } + + /** + * Performs the given action for each value. + * + * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed + * in parallel + * @param action the action + * @since 1.8 + */ + public void forEachValue(long parallelismThreshold, + Consumer action) { + if (action == null) + throw new NullPointerException(); + new ForEachValueTask + (null, batchFor(parallelismThreshold), 0, 0, table, + action).invoke(); + } + + /** + * Performs the given action for each non-null transformation of each value. + * + * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed + * in parallel + * @param transformer a function returning the transformation for an element, or null if there is + * no transformation (in which case the action is not applied) + * @param action the action + * @param the return type of the transformer + * @since 1.8 + */ + public void forEachValue(long parallelismThreshold, + Function transformer, + Consumer action) { + if (transformer == null || action == null) + throw new NullPointerException(); + new ForEachTransformedValueTask + (null, batchFor(parallelismThreshold), 0, 0, table, + transformer, action).invoke(); + } + + /** + * Returns a non-null result from applying the given search function on each value, or null if none. Upon + * success, further element processing is suppressed and the results of any other parallel invocations of + * the search function are ignored. + * + * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed + * in parallel + * @param searchFunction a function returning a non-null result on success, else null + * @param the return type of the search function + * @return a non-null result from applying the given search function on each value, or null if none + * @since 1.8 + */ + public U searchValues(long parallelismThreshold, + Function searchFunction) { + if (searchFunction == null) throw new NullPointerException(); + return new SearchValuesTask + (null, batchFor(parallelismThreshold), 0, 0, table, + searchFunction, new AtomicReference()).invoke(); + } + + /** + * Returns the result of accumulating all values using the given reducer to combine values, or null if + * none. + * + * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed + * in parallel + * @param reducer a commutative associative combining function + * @return the result of accumulating all values + * @since 1.8 + */ + public V reduceValues(long parallelismThreshold, + BiFunction reducer) { + if (reducer == null) throw new NullPointerException(); + return new ReduceValuesTask + (null, batchFor(parallelismThreshold), 0, 0, table, + null, reducer).invoke(); + } + + /** + * Returns the result of accumulating the given transformation of all values using the given reducer to + * combine values, or null if none. + * + * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed + * in parallel + * @param transformer a function returning the transformation for an element, or null if there is + * no transformation (in which case it is not combined) + * @param reducer a commutative associative combining function + * @param the return type of the transformer + * @return the result of accumulating the given transformation of all values + * @since 1.8 + */ + public U reduceValues(long parallelismThreshold, + Function transformer, + BiFunction reducer) { + if (transformer == null || reducer == null) + throw new NullPointerException(); + return new MapReduceValuesTask + (null, batchFor(parallelismThreshold), 0, 0, table, + null, transformer, reducer).invoke(); + } + + /** + * Returns the result of accumulating the given transformation of all values using the given reducer to + * combine values, and the given basis as an identity value. + * + * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed + * in parallel + * @param transformer a function returning the transformation for an element + * @param basis the identity (initial default value) for the reduction + * @param reducer a commutative associative combining function + * @return the result of accumulating the given transformation of all values + * @since 1.8 + */ + public double reduceValuesToDouble(long parallelismThreshold, + ToDoubleFunction transformer, + double basis, + DoubleBinaryOperator reducer) { + if (transformer == null || reducer == null) + throw new NullPointerException(); + return new MapReduceValuesToDoubleTask + (null, batchFor(parallelismThreshold), 0, 0, table, + null, transformer, basis, reducer).invoke(); + } + + /** + * Returns the result of accumulating the given transformation of all values using the given reducer to + * combine values, and the given basis as an identity value. + * + * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed + * in parallel + * @param transformer a function returning the transformation for an element + * @param basis the identity (initial default value) for the reduction + * @param reducer a commutative associative combining function + * @return the result of accumulating the given transformation of all values + * @since 1.8 + */ + public long reduceValuesToLong(long parallelismThreshold, + ToLongFunction transformer, + long basis, + LongBinaryOperator reducer) { + if (transformer == null || reducer == null) + throw new NullPointerException(); + return new MapReduceValuesToLongTask + (null, batchFor(parallelismThreshold), 0, 0, table, + null, transformer, basis, reducer).invoke(); + } + + /** + * Returns the result of accumulating the given transformation of all values using the given reducer to + * combine values, and the given basis as an identity value. + * + * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed + * in parallel + * @param transformer a function returning the transformation for an element + * @param basis the identity (initial default value) for the reduction + * @param reducer a commutative associative combining function + * @return the result of accumulating the given transformation of all values + * @since 1.8 + */ + public int reduceValuesToInt(long parallelismThreshold, + ToIntFunction transformer, + int basis, + IntBinaryOperator reducer) { + if (transformer == null || reducer == null) + throw new NullPointerException(); + return new MapReduceValuesToIntTask + (null, batchFor(parallelismThreshold), 0, 0, table, + null, transformer, basis, reducer).invoke(); + } + + /** + * Performs the given action for each entry. + * + * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed + * in parallel + * @param action the action + * @since 1.8 + */ + public void forEachEntry(long parallelismThreshold, + Consumer> action) { + if (action == null) throw new NullPointerException(); + new ForEachEntryTask(null, batchFor(parallelismThreshold), 0, 0, table, + action).invoke(); + } + + /** + * Performs the given action for each non-null transformation of each entry. + * + * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed + * in parallel + * @param transformer a function returning the transformation for an element, or null if there is + * no transformation (in which case the action is not applied) + * @param action the action + * @param the return type of the transformer + * @since 1.8 + */ + public void forEachEntry(long parallelismThreshold, + Function, ? extends U> transformer, + Consumer action) { + if (transformer == null || action == null) + throw new NullPointerException(); + new ForEachTransformedEntryTask + (null, batchFor(parallelismThreshold), 0, 0, table, + transformer, action).invoke(); + } + + /** + * Returns a non-null result from applying the given search function on each entry, or null if none. Upon + * success, further element processing is suppressed and the results of any other parallel invocations of + * the search function are ignored. + * + * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed + * in parallel + * @param searchFunction a function returning a non-null result on success, else null + * @param the return type of the search function + * @return a non-null result from applying the given search function on each entry, or null if none + * @since 1.8 + */ + public U searchEntries(long parallelismThreshold, + Function, ? extends U> searchFunction) { + if (searchFunction == null) throw new NullPointerException(); + return new SearchEntriesTask + (null, batchFor(parallelismThreshold), 0, 0, table, + searchFunction, new AtomicReference()).invoke(); + } + + /** + * Returns the result of accumulating all entries using the given reducer to combine values, or null if + * none. + * + * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed + * in parallel + * @param reducer a commutative associative combining function + * @return the result of accumulating all entries + * @since 1.8 + */ + public Map.Entry reduceEntries(long parallelismThreshold, + BiFunction, Map.Entry, ? extends Map.Entry> reducer) { + if (reducer == null) throw new NullPointerException(); + return new ReduceEntriesTask + (null, batchFor(parallelismThreshold), 0, 0, table, + null, reducer).invoke(); + } + + /** + * Returns the result of accumulating the given transformation of all entries using the given reducer to + * combine values, or null if none. + * + * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed + * in parallel + * @param transformer a function returning the transformation for an element, or null if there is + * no transformation (in which case it is not combined) + * @param reducer a commutative associative combining function + * @param the return type of the transformer + * @return the result of accumulating the given transformation of all entries + * @since 1.8 + */ + public U reduceEntries(long parallelismThreshold, + Function, ? extends U> transformer, + BiFunction reducer) { + if (transformer == null || reducer == null) + throw new NullPointerException(); + return new MapReduceEntriesTask + (null, batchFor(parallelismThreshold), 0, 0, table, + null, transformer, reducer).invoke(); + } + + /** + * Returns the result of accumulating the given transformation of all entries using the given reducer to + * combine values, and the given basis as an identity value. + * + * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed + * in parallel + * @param transformer a function returning the transformation for an element + * @param basis the identity (initial default value) for the reduction + * @param reducer a commutative associative combining function + * @return the result of accumulating the given transformation of all entries + * @since 1.8 + */ + public double reduceEntriesToDouble(long parallelismThreshold, + ToDoubleFunction> transformer, + double basis, + DoubleBinaryOperator reducer) { + if (transformer == null || reducer == null) + throw new NullPointerException(); + return new MapReduceEntriesToDoubleTask + (null, batchFor(parallelismThreshold), 0, 0, table, + null, transformer, basis, reducer).invoke(); + } + + /** + * Returns the result of accumulating the given transformation of all entries using the given reducer to + * combine values, and the given basis as an identity value. + * + * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed + * in parallel + * @param transformer a function returning the transformation for an element + * @param basis the identity (initial default value) for the reduction + * @param reducer a commutative associative combining function + * @return the result of accumulating the given transformation of all entries + * @since 1.8 + */ + public long reduceEntriesToLong(long parallelismThreshold, + ToLongFunction> transformer, + long basis, + LongBinaryOperator reducer) { + if (transformer == null || reducer == null) + throw new NullPointerException(); + return new MapReduceEntriesToLongTask + (null, batchFor(parallelismThreshold), 0, 0, table, + null, transformer, basis, reducer).invoke(); + } + + /** + * Returns the result of accumulating the given transformation of all entries using the given reducer to + * combine values, and the given basis as an identity value. + * + * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed + * in parallel + * @param transformer a function returning the transformation for an element + * @param basis the identity (initial default value) for the reduction + * @param reducer a commutative associative combining function + * @return the result of accumulating the given transformation of all entries + * @since 1.8 + */ + public int reduceEntriesToInt(long parallelismThreshold, + ToIntFunction> transformer, + int basis, + IntBinaryOperator reducer) { + if (transformer == null || reducer == null) + throw new NullPointerException(); + return new MapReduceEntriesToIntTask + (null, batchFor(parallelismThreshold), 0, 0, table, + null, transformer, basis, reducer).invoke(); + } + + /** + * Performs the given action for each (key, value). + * + * @param parallelismThreshold the (estimated) number of elements needed for this operation to be executed + * in parallel + * @param action the action + * @since 1.8 + */ + public void forEach(long parallelismThreshold, + BiConsumer action) { + if (action == null) throw new NullPointerException(); + new ForEachMappingTask + (null, batchFor(parallelismThreshold), 0, 0, table, + action).invoke(); + } + + + @SuppressWarnings("serial") + static final class SearchMappingsTask + extends BulkTask { + final BiFunction searchFunction; + final AtomicReference result; + + SearchMappingsTask + (BulkTask p, int b, int i, int f, Node[] t, + BiFunction searchFunction, + AtomicReference result) { + super(p, b, i, f, t); + this.searchFunction = searchFunction; + this.result = result; + } + + public final U getRawResult() { + return result.get(); + } + + public final void compute() { + final BiFunction searchFunction; + final AtomicReference result; + if ((searchFunction = this.searchFunction) != null && + (result = this.result) != null) { + for (int i = baseIndex, f, h; batch > 0 && + (h = ((f = baseLimit) + i) >>> 1) > i; ) { + if (result.get() != null) + return; + addToPendingCount(1); + new SearchMappingsTask + (this, batch >>>= 1, baseLimit = h, f, tab, + searchFunction, result).fork(); + } + while (result.get() == null) { + U u; + Node p; + if ((p = advance()) == null) { + propagateCompletion(); + break; + } + if ((u = searchFunction.apply(p.key, p.val)) != null) { + if (result.compareAndSet(null, u)) + quietlyCompleteRoot(); + break; + } + } + } + } + } + private static int figureBufferAllocationFactor(SharedHashMapBuilder builder) { // if expected map size is about 1000, seems rather wasteful to allocate // key and value serialization buffers each x64 of expected entry size.. @@ -341,6 +1102,7 @@ abstract class AbstractVanillaSharedMap extends net.openhft.collections.Co return put0(key, value, true); } + /** * {@inheritDoc} */ @@ -356,9 +1118,10 @@ abstract class AbstractVanillaSharedMap extends net.openhft.collections.Co long hash = Hasher.hash(keyBytes); int segmentNum = hasher.getSegment(hash); int segmentHash = hasher.segmentHash(hash); - return segments[segmentNum].put(keyBytes, key, value, segmentHash, replaceIfPresent); + return segments[segmentNum].put(keyBytes, key, value, segmentHash, replaceIfPresent, null); } + DirectBytes getKeyAsBytes(K key) { DirectBytes buffer = acquireBufferForKey(); if (generatedKeyType) @@ -825,7 +1588,18 @@ abstract class AbstractVanillaSharedMap extends net.openhft.collections.Co } } - V put(Bytes keyBytes, K key, V value, int hash2, boolean replaceIfPresent) { + + /** + * @param keyBytes the bytes of the key + * @param key the key to the entry you wish to add or update + * @param value the value in the entry you wish to add or update + * @param hash2 the hash of the key in this segment + * @param replaceIfPresent true if the item is to be replaced + * @param mappingFunction if non null, this takes presidents over the value, the value calculated via + * the mappingFunction is used instead + * @return + */ + V put(Bytes keyBytes, K key, V value, int hash2, boolean replaceIfPresent, Function mappingFunction) { lock(); try { long keyLen = keyBytes.remaining(); @@ -838,11 +1612,22 @@ abstract class AbstractVanillaSharedMap extends net.openhft.collections.Co // key is found entry.skip(keyLen); if (replaceIfPresent) { - return replaceValueOnPut(key, value, entry, pos, offset, !putReturnsNull, hashLookup); + { + if (mappingFunction != null) + value = mappingFunction.apply(key); + if (value == null) + return putReturnsNull ? null : readValue(entry, null); + return replaceValueOnPut(key, value, entry, pos, offset, !putReturnsNull, hashLookup); + } } else { return putReturnsNull ? null : readValue(entry, null); } } + + if (mappingFunction != null) + value = mappingFunction.apply(key); + if (value == null) + return null; // key is not found long offset = putEntry(keyBytes, value, false); incrementSize(); @@ -1339,6 +2124,7 @@ abstract class AbstractVanillaSharedMap extends net.openhft.collections.Co return hashLookup; } + private class PosPresentOnce implements IntIntMultiMap.EntryConsumer { int pos, count = 0;