Isso-W commented on code in PR #2542:
URL: https://github.com/apache/systemds/pull/2542#discussion_r3694731341
##########
src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixSketch.java:
##########
@@ -19,89 +19,807 @@
package org.apache.sysds.runtime.matrix.data;
-import org.apache.sysds.common.Types;
-
+import java.util.ArrayList;
import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+
+import org.apache.sysds.common.Types;
+import org.apache.sysds.runtime.DMLRuntimeException;
+import org.apache.sysds.runtime.util.CommonThreadPool;
+import org.apache.sysds.runtime.util.UtilFunctions;
+import org.apache.sysds.utils.stats.InfrastructureAnalyzer;
public class LibMatrixSketch {
+ private static final long PAR_UNIQUE_NUMCELL_THRESHOLD = 1024 * 16;
+ private static final long PAR_UNIQUE_MAX_LOCAL_BYTES_FRACTION = 4;
+ /**
+ * Conservative footprint of one value retained in a
HashSet<Double>: the boxed Double plus amortized hash-map
+ * node and backing-array overhead.
+ */
+ private static final long PAR_UNIQUE_BYTES_PER_CELL = Double.BYTES * 8;
+ /**
+ * Computes unique values with the original single-threaded behavior.
The overload with a parallelism argument keeps
+ * this path as the k=1 baseline.
+ *
+ * @param blkIn input matrix block
+ * @param dir unique direction
+ * @return matrix block containing unique values
+ */
public static MatrixBlock getUniqueValues(MatrixBlock blkIn,
Types.Direction dir) {
- //similar to R's unique, this operation takes a matrix and
computes the
- //unique values (or rows in case of multiple column inputs)
-
+ return getUniqueValues(blkIn, dir, 1);
+ }
+
+ /**
+ * Computes unique values. For sufficiently large inputs and k > 1,
this uses parallel local deduplication or its
+ * batched variant.
+ *
+ * @param blkIn input matrix block
+ * @param dir unique direction
+ * @param k requested degree of parallelism
+ * @return matrix block containing unique values
+ */
+ public static MatrixBlock getUniqueValues(MatrixBlock blkIn,
Types.Direction dir, int k) {
+ return getUniqueValues(blkIn, dir, k,
getDefaultLocalBytesBudget());
+ }
+
+ /**
+ * Computes unique values with an explicit budget for the transient
deduplication structures. The budget decides
+ * between the full parallel path, its batched variant, and the
sequential fallback. This overload exists so tests
+ * can inject a small budget and deterministically exercise the batched
path, which the heap-derived default would
+ * not trigger.
+ *
+ * @param blkIn input matrix block
+ * @param dir unique direction
+ * @param k requested degree of parallelism
+ * @param maxLocalBytes budget in bytes for transient deduplication
structures
+ * @return matrix block containing unique values
+ */
+ public static MatrixBlock getUniqueValues(MatrixBlock blkIn,
Types.Direction dir, int k, long maxLocalBytes) {
+ // Similar to R's unique, this operation computes unique values
according
+ // to the requested direction.
+ if(!satisfiesMultiThreadingConstraints(blkIn, dir, k))
+ return getUniqueValuesSequential(blkIn, dir);
+
+ boolean localDedupMemorySafe =
isLocalDedupMemoryBudgetSafe(blkIn, dir, k, maxLocalBytes);
+ switch(dir) {
+ case RowCol:
Review Comment:
I prototyped an alternative before giving up on it.
Today each worker builds a local set and the caller merges them serially,
and that merge grows with the total distinct count. At a 100% share it
re-inserts everything, which caps the speedup at `k/(k+1)`. A tree-shaped merge
does not help, as the last merge is still `O(d)` on one thread.
The prototype partitions by value hash, so each partition can be merged
independently and in parallel, and uses `DoubleCountHashMap` to avoid boxing.
Same machine, k=8, results verified against sequential:
input | share | distinct | sequential [ms] | current [ms] | partitioned [ms]
| current | partitioned
-- | -- | -- | -- | -- | -- | -- | --
2000000x1 | 1% | 20000 | 77.0 | 15.5 | 18.0 | 4.97 | 4.28
2000000x1 | 10% | 200000 | 145.6 | 153.2 | 52.4 | 0.95 | 2.78
2000000x1 | 50% | 1000000 | 128.4 | 204.1 | 118.2 | 0.63 | 1.09
2000000x1 | 100% | 2000000 | 170.0 | 241.7 | 129.7 | 0.70 | 1.31
200000x64 | 1% | 128000 | 667.5 | 238.5 | 161.6 | 2.80 | 4.13
200000x64 | 10% | 1280000 | 803.8 | 1127.3 | 734.4 | 0.71 | 1.09
200000x64 | 50% | 6400000 | 1078.8 | 1572.8 | 1000.9 | 0.69 | 1.08
200000x64 | 100% | 12800000 | 1437.7 | 1928.6 | 1148.2 | 0.75 | 1.25
So the partitioned variant is never slower than sequential, while the
current one drops to 0.63. But it only gains about 1.1 at high distinct counts,
where the work becomes memory bound.
Does that result worth a multi-threading? Or I can also return to
single-threaded.
--
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]
