belliottsmith commented on code in PR #6:
URL: https://github.com/apache/cassandra-accord/pull/6#discussion_r945106839
##########
accord-core/src/main/java/accord/primitives/Deps.java:
##########
@@ -0,0 +1,1095 @@
+package accord.primitives;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.function.Predicate;
+
+import accord.utils.ArrayBuffers;
+import java.util.stream.Collector;
+import java.util.stream.Collectors;
+
+import accord.api.Key;
+import accord.local.Command;
+import accord.utils.SortedArrays;
+import com.carrotsearch.hppc.ObjectIntHashMap;
+import com.google.common.base.Preconditions;
+
+import static accord.utils.ArrayBuffers.*;
+import static accord.utils.SortedArrays.*;
+
+// TODO (now): switch to RoutingKey
+public class Deps implements Iterable<Map.Entry<Key, TxnId>>
+{
+ private static final boolean DEBUG_CHECKS = true;
+
+ private static final TxnId[] NO_TXNIDS = new TxnId[0];
+ private static final int[] NO_INTS = new int[0];
+ public static final Deps NONE = new Deps(Keys.EMPTY, NO_TXNIDS, NO_INTS);
+
+ public static Collector<Command, Builder, Builder> collector(Keys keys)
+ {
+ return Collector.of(() -> Deps.builder(keys), Deps.Builder::add, (a,
b) -> { throw new IllegalStateException(); });
+ }
+
+ public static class Builder
+ {
+ final Keys keys;
+ final ObjectIntHashMap<TxnId> txnIdLookup = new ObjectIntHashMap<>();
+ TxnId[] txnIds = new TxnId[4];
+ // Key -> [TxnId]
+ final int[][] keysToTxnId;
+ // Key -> Counter
+ final int[] keysToTxnIdCounts;
+
+ public Builder(Keys keys)
+ {
+ this.keys = keys;
+ this.keysToTxnId = new int[keys.size()][4];
+ this.keysToTxnIdCounts = new int[keys.size()];
+ }
+
+ public boolean isEmpty()
+ {
+ return txnIdLookup.isEmpty();
+ }
+
+ public void add(Command command)
+ {
+ int idx = ensureTxnIdx(command.txnId());
+
+ long count = keys.foldlIntersect(command.txn().keys, (li, ri, k,
p, v) -> {
+ if (keysToTxnId[li].length == keysToTxnIdCounts[li])
+ keysToTxnId[li] = Arrays.copyOf(keysToTxnId[li],
keysToTxnId[li].length * 2);
+ keysToTxnId[li][keysToTxnIdCounts[li]++] = idx;
+ return v + 1;
+ }, 0, 0, -1);
+
+ if (count == 0)
+ throw new IllegalArgumentException(command + " does not
intersect " + keys);
+ }
+
+ public void add(Key key, TxnId txnId)
+ {
+ int txnIdx = ensureTxnIdx(txnId);
+ int keyIdx = keys.indexOf(key);
+ if (keysToTxnIdCounts[keyIdx] == keysToTxnId[keyIdx].length)
+ keysToTxnId[keyIdx] = Arrays.copyOf(keysToTxnId[keyIdx],
keysToTxnIdCounts[keyIdx] * 2);
+ keysToTxnId[keyIdx][keysToTxnIdCounts[keyIdx]++] = txnIdx;
+ }
+
+ public boolean contains(TxnId txnId)
+ {
+ return txnIdLookup.containsKey(txnId);
+ }
+
+ private int ensureTxnIdx(TxnId txnId)
+ {
+ int i = txnIdLookup.getOrDefault(txnId, -1);
+ if (i < 0)
+ {
+ txnIdLookup.put(txnId, i = txnIdLookup.size());
+ if (i == txnIds.length)
+ txnIds = Arrays.copyOf(txnIds, txnIds.length * 2);
+ }
+ return i;
+ }
+
+ public Deps build()
+ {
+ TxnId[] txnIds = txnIdLookup.keys().toArray(TxnId.class);
+ Arrays.sort(txnIds, TxnId::compareTo);
+ // all data is indexed based off insertion order, but now that
build is happening its desirable to have ids
+ // in tx order, in order to map from sorted -> insert order, need
"txnIdMap" as the bridge
+ int[] txnIdMap = new int[txnIds.length];
+ for (int i = 0 ; i < txnIdMap.length ; i++)
+ txnIdMap[txnIdLookup.get(txnIds[i])] = i;
+
+ int keyCount = 0;
+ // Key -> distinct([*TxnID]])
+ // at the key position is the OFFSET into the array where the
txnIds are for the key
+ // Example
+ // OFFSET(3), OFFSET(5), *tx1, *tx1, *tx2
+ int[] result; {
+ int count = 0;
+ for (int i = 0 ; i < keys.size() ; ++i)
+ {
+ keyCount += keysToTxnIdCounts[i] > 0 ? 1 : 0;
+ count += keysToTxnIdCounts[i];
+ }
+ result = new int[count + keyCount];
+ }
+
+ int keyIndex = 0;
+ int offset = keyCount;
+ for (int i = 0 ; i < keys.size() ; ++i)
+ {
+ if (keysToTxnIdCounts[i] > 0)
+ {
+ int count = keysToTxnIdCounts[i];
+ int[] src = keysToTxnId[i];
+ // store all txnIds to results at offset
+ for (int j = 0 ; j < count ; ++j)
+ result[j + offset] = txnIdMap[src[j]];
+ Arrays.sort(result, offset, count + offset);
+ // the same txnId may be present multiple times, attempt
to detect this and push all dups to the end
+ int dups = 0;
+ for (int j = offset + 1 ; j < offset + count ; ++j)
+ {
+ if (result[j] == result[j - 1]) ++dups;
+ // positions i == i - 1 but dups were found; replace
the "first" dup with the current value, removing it
+ else if (dups > 0) result[j - dups] = result[j];
+ }
+ result[keyIndex] = offset += count - dups;
+ ++keyIndex;
+ }
+ }
+ if (offset < result.length)
+ result = Arrays.copyOf(result, offset);
+
+ Keys keys = this.keys;
+ if (keyCount < keys.size())
+ {
+ keyIndex = 0;
+ Key[] newKeys = new Key[keyCount];
+ for (int i = 0 ; i < keys.size() ; ++i)
+ {
+ if (keysToTxnIdCounts[i] > 0)
+ newKeys[keyIndex++] = keys.get(i);
+ }
+ keys = Keys.of(newKeys);
+ }
+
+ return new Deps(keys, txnIds, result);
+ }
+ }
+
+ public static Builder builder(Keys keys)
+ {
+ return new Builder(keys);
+ }
+
+ private static class LinearMerger extends ArrayBuffers.SavingManager
implements DepsConstructor<Object>
Review Comment:
As discussed offline, an updated patch using the latest caching logic and a
similar split or enabled/disabled shows 50-90% memory reduction using caching.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
