[SYSTEMDS-3524] Add multi-threaded unique in LibMatrixSketch#2542
[SYSTEMDS-3524] Add multi-threaded unique in LibMatrixSketch#2542shieru1214 wants to merge 5 commits into
Conversation
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #2542 +/- ##
============================================
+ Coverage 71.41% 71.48% +0.06%
- Complexity 48774 49849 +1075
============================================
Files 1571 1602 +31
Lines 188912 193298 +4386
Branches 37067 37836 +769
============================================
+ Hits 134917 138184 +3267
- Misses 43544 44317 +773
- Partials 10451 10797 +346 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
There was a problem hiding this comment.
Thank you for the contribution @shieru1214. It looks good overall, but there are some issues I would like you to go over.
First, the most critical problem is that the AggregateUnary instruction will always call the single-threaded implementation. With multi-threading in place, it should be only used as a fallback.
Second, the batched multi-threading path has a number of problems related to memory estimation and guards against excessive usage. Below is the summary of suggested fixes:
- Int overflow: compute as
int end = (int) Math.min((long) pos + config._taskLen, config._len); - The safety condition for
RowandColisnumThreads × cellsPerIndex × 64 ≤ budget(one live row/column set per thread). Otherwise, you force batched execution where it can be safely avoided - You have to keep in mind that for batched
RowColthe mergedHashSetgrows toward the distinct count. So you should sizetaskLenagainst a fraction of the budget with the remainder reserved for the merged set. If the worst-case simply doesn't fit, fall back to sequential execution. - I'm pretty sure batched tests do not test the respective path in our CI since the JVM heap is never sufficiently low (the worst-case requirement is ~77 MiB). You can, e.g., add an overload like
maxLocalBytes, or account for it in any other way you prefer
Also, the formatting job is failing, you need to run the dev/format-changed.sh script. Your fork doesn't have it, so please rebase on the current main so that it's present in your repository as well.
| //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); |
There was a problem hiding this comment.
The degree of parallelism is hard-coded here, so AggregateUnaryCPInstruction will always call the single-threaded implementation.
| for( int pos = 0; pos < config._len; ) { | ||
| ArrayList<UniqueValueTask> tasks = new ArrayList<>(); | ||
| for( int i = 0; i < config._numThreads && pos < config._len; i++ ) { | ||
| int end = Math.min(pos + config._taskLen, config._len); |
There was a problem hiding this comment.
This will overflow when pos is close to Integer.MAX_VALUE, i.e., when the number of rows in the matrix is very large
| for( int pos = 0; pos < config._len; ) { | ||
| ArrayList<UniqueCountTask> tasks = new ArrayList<>(); | ||
| for( int i = 0; i < config._numThreads && pos < config._len; i++ ) { | ||
| int end = Math.min(pos + config._taskLen, config._len); |
| for( int pos = 0; pos < config._len; ) { | ||
| ArrayList<UniqueOutputTask> tasks = new ArrayList<>(); | ||
| for( int i = 0; i < config._numThreads && pos < config._len; i++ ) { | ||
| int end = Math.min(pos + config._taskLen, config._len); |
|
|
||
| ExecutorService pool = CommonThreadPool.get(config._numThreads); | ||
| try { | ||
| HashSet<Double> hashSet = new HashSet<>(); |
There was a problem hiding this comment.
This HashSet is unbounded. When almost all values are unique, the set grows to rlen x clen, so it in principle matches the sequential case.
| if( numThreads <= 1 ) | ||
| return null; | ||
|
|
||
| int taskLen = Math.max(1, (int) Math.min(Integer.MAX_VALUE, maxBatchIndexes / numThreads)); |
There was a problem hiding this comment.
Then numThreads x taskLen consumes the entire 1/4 of the heap in the worst case
| * @return true if local deduplication is small enough for the parallel path | ||
| */ | ||
| private static boolean isLocalDedupMemoryBudgetSafe(MatrixBlock blkIn, Types.Direction dir) { | ||
| return getMaxLocalDedupIndexes(blkIn, dir) >= getPartitionLength(blkIn, dir); |
There was a problem hiding this comment.
If I understand correctly, you check here whether you can hold in memory each dedup set for every row at the same time. This is correct for RowCol, but due to hashSet.clear() it is unnecessarily conservative for Row and Col, because the peak memory is then numThreads × clen × 64 bytes which doesn't depend on rlen.
| * under the License. | ||
| */ | ||
|
|
||
| package org.apache.sysds.runtime.matrix.data; |
There was a problem hiding this comment.
Please, move the tests to respective files in src/test/java/org/apache/sysds/test/functions/unique, feel free to extend for the multi-threaded case if necessary
| return 0; | ||
|
|
||
| long bytesPerIndex = cellsPerIndex * Double.BYTES * PAR_UNIQUE_LOCAL_BYTES_OVERHEAD; | ||
| long maxLocalBytes = Runtime.getRuntime().maxMemory() / PAR_UNIQUE_MAX_LOCAL_BYTES_FRACTION; |
There was a problem hiding this comment.
Read from InfrastructureAnalyzer.getLocalMaxMemory() instead
github-project-automation
Bot
moved this from In Progress
to In Review
in SystemDS PR Queue
2 participants
Summary
This PR adds a multi-threaded path for
uniqueinLibMatrixSketch.The existing single-threaded implementation is kept and is still used as the baseline/fallback. I added a new overload:
getUniqueValues(MatrixBlock blkIn, Types.Direction dir, int k), wherekcontrols the requested number of threads.The implementation follows the existing SystemDS
uniquedirection semantics:RowCol: unique scalar values over the input matrix, returned as one columnRow: row-wise unique scalar values, preserving the number of rowsCol: column-wise unique scalar values, preserving the number of columnsImplementation
For
k > 1and sufficiently large inputs, the code uses balanced row or column ranges andCommonThreadPool.The parallel logic is direction-specific:
RowCol: each worker builds a localHashSet<Double>over its row range, and the local sets are merged afterwards.Row: rows are partitioned across workers. The code first computes the maximum number of unique values per row, allocates the output, and then fills the row-wise unique values in parallel.Col: columns are partitioned across workers. The code first computes the maximum number of unique values per column, allocates the output, and then fills the column-wise unique values in parallel.There are two parallel paths:
The batched path processes a limited number of ranges at a time, clears temporary local data earlier, and then continues with the next batch. If the input is small,
k <= 1, or batching is not useful, the code falls back to the original single-threaded path.Testing
I added a focused JUnit test:
src/test/java/org/apache/sysds/runtime/matrix/data/LibMatrixSketchUniqueParallelTest.javaIt checks
RowCol,Row, andCol, and compares the multi-threaded result with the single-threaded baseline.I ran the following tests:
mvn -Dtest=LibMatrixSketchUniqueParallelTest testmvn -Dtest=UniqueRow,UniqueRowCol testThe focused test completed with 4 tests passing, and the existing unique tests completed with 9 tests passing. I also ran the focused test with
-Xmx64mto exercise the memory-aware batched path.Runtime experiments
I ran local runtime experiments with fixed
k=4, comparing against thek=1baseline.The experiments were run locally on an Apple Silicon Mac using Java 17. The JVM heap was fixed at 3 GB, and all inputs used dense
MatrixBlockstorage.For each case, I used three warm-up iterations followed by five measured iterations. The order of
k=1andk=4was alternated between iterations to reduce ordering effects, and the median runtime is reported. Input generation is not included in the measurements.The experiments cover:
RowCol,Row, andCol1%,10%,50%, and100%k=1andk=4In the tables,
Unique Valueshas a different meaning for each direction:RowCol, it is the total number of distinct scalar values in the inputRow, it is the number of distinct scalar values within each rowCol, it is the number of distinct scalar values within each columnThe exact integer count used for each case is shown in the tables. Speedup is calculated as
k=1 time / k=4 time, so a value above 1 means that the multi-threaded path was faster.The 3 GB heap limit was chosen to keep the large dense inputs within memory while allowing the memory guard to select the batched path for the large
RowandColcases. With this setting, the smaller cases use the regular parallel path, while the largeRowandColcases use batched parallel processing.These are local measurements and are intended as an initial comparison rather than a full benchmark.
RowCol
Row
Col
The row-wise and column-wise cases show consistent speedups because rows and columns can be processed independently. Their speedup generally decreases as the number of unique values grows, since larger sets and outputs require more work.
RowColbehaves differently because its thread-local sets must still be merged into one global set. The large input with a 1% unique share shows a small improvement, but in the other tested cases the allocation and merge overhead is greater than the saved parallel work.