Isso-W commented on PR #2542:
URL: https://github.com/apache/systemds/pull/2542#issuecomment-5085173978
## Update: review feedback addressed
Thanks for the detailed review. All points are addressed:
**1. `AggregateUnary` always called the single-threaded implementation.**
`AggregateUnaryCPInstruction` now passes `op.getNumThreads()` into
`LibMatrixSketch.getUniqueValues(...)`. The thread count was already plumbed
into `UnarySketchOperator` by `InstructionUtils`, it just was not forwarded, so
the multi-threaded path was unreachable from DML and the 2-argument overload is
now only the `k = 1` fallback.
**2. Int overflow in the batched task ranges.**
All three occurrences now compute `int end = (int) Math.min((long) pos +
config._taskLen, config._len);`.
**3. Safety condition for `Row` and `Col`.**
The row-wise and column-wise workers reuse a single set that is cleared per
row/column, so only one live set per thread has to be charged. The guard is now
`numThreads * cellsPerIndex * 64 <= budget` instead of charging every index,
which no longer forces batched execution where it can safely be avoided.
**4. Batched `RowCol` and the merged set.**
The `RowCol` workers accumulate over their whole range and are merged into a
set that grows toward the total distinct count. The batched configuration now
reserves the all-distinct worst case for the merged set up front and sizes
`taskLen` against the remaining budget only. If not even the merged set fits,
batching cannot help and the code falls back to sequential execution.
**5. The batched paths were not covered in CI.**
There is now a `getUniqueValues(blkIn, dir, k, maxLocalBytes)` overload, so
tests can inject a small budget instead of relying on a low JVM heap. The test
also moved from `org.apache.sysds.runtime.matrix.data` to
`org.apache.sysds.test.functions.unique`, because the CI test selection only
scans `org.apache.sysds.test.*` and the test was therefore never executed.
Added cases now exercise all three batched paths and their sequential fallbacks
deterministically.
Additionally, the memory budget is now derived from
`InfrastructureAnalyzer.getLocalMaxMemory()` rather than
`Runtime.getRuntime().maxMemory()`, so it reflects the local JVM memory that
parfor adjusts per worker, matching the convention used in the neighbouring
matrix libraries.
The branch is rebased on current `main`, and `dev/format-changed.sh` has
been applied.
## Testing
`src/test/java/org/apache/sysds/test/functions/unique/LibMatrixSketchUniqueParallelTest.java`
```
mvn -Dtest=LibMatrixSketchUniqueParallelTest test -> 10 tests, 0 failures
mvn -Dtest=UniqueRow,UniqueRowCol test -> 9 tests, 0 failures
```
19 tests pass in total. Checkstyle reports 0 violations and the format check
on the PR-edited lines is clean.
## Runtime experiments
The benchmark is included as
`src/test/java/org/apache/sysds/performance/matrix/UniquePerf.java`, so the
numbers below can be reproduced:
```
java -Xmx3g --add-modules jdk.incubator.vector \
-cp target/classes:target/test-classes:<deps> \
org.apache.sysds.performance.matrix.UniquePerf 4 5
```
The experiments were run locally on an Intel Core i7-13700H (20 logical
cores) using Java 17, with the JVM heap fixed at 3 GB and dense `MatrixBlock`
inputs throughout.
For each configuration I used three warm-up iterations followed by five
measured iterations. The order of the `k=1` and `k=4` runs is alternated
between iterations to reduce ordering effects, and the median runtime is
reported. Input generation is not included in the measurements.
The experiments cover the directions `RowCol`, `Row` and `Col`, a small and
a large input, unique shares of 1%, 10%, 50% and 100%, and thread counts `k=1`
and `k=4`. `Unique Values` is the number of distinct scalar values in the whole
input for `RowCol`, per row for `Row` and per column for `Col`. Speedup is
`k=1` time divided by `k=4` time, so a value above 1 means the multi-threaded
path was faster.
These are local measurements and are intended as an initial comparison
rather than a full benchmark.
### RowCol
| Size | Input | Unique Share | Unique Values | k=1 (ms) | k=4 (ms) |
Speedup |
|---|---|---|---|---|---|---|
| small | 100000x1 | 1% | 1000 | 3.186 | 21.061 | 0.15 |
| small | 100000x1 | 10% | 10000 | 3.164 | 7.690 | 0.41 |
| small | 100000x1 | 50% | 50000 | 2.407 | 6.757 | 0.36 |
| small | 100000x1 | 100% | 100000 | 3.228 | 5.975 | 0.54 |
| large | 2000000x1 | 1% | 20000 | 23.134 | 23.172 | 1.00 |
| large | 2000000x1 | 10% | 200000 | 31.379 | 58.022 | 0.54 |
| large | 2000000x1 | 50% | 1000000 | 180.886 | 248.605 | 0.73 |
| large | 2000000x1 | 100% | 2000000 | 248.350 | 325.830 | 0.76 |
### Row
| Size | Input | Unique Share | Unique Values per Row | k=1 (ms) | k=4 (ms)
| Speedup |
|---|---|---|---|---|---|---|
| small | 10000x64 | 1% | 1 | 41.429 | 13.344 | 3.10 |
| small | 10000x64 | 10% | 6 | 44.689 | 15.397 | 2.90 |
| small | 10000x64 | 50% | 32 | 64.297 | 32.751 | 1.96 |
| small | 10000x64 | 100% | 64 | 88.566 | 55.402 | 1.60 |
| large | 200000x64 | 1% | 1 | 821.775 | 220.942 | 3.72 |
| large | 200000x64 | 10% | 6 | 921.537 | 260.261 | 3.54 |
| large | 200000x64 | 50% | 32 | 1290.565 | 568.493 | 2.27 |
| large | 200000x64 | 100% | 64 | 1837.356 | 1175.834 | 1.56 |
### Col
| Size | Input | Unique Share | Unique Values per Column | k=1 (ms) | k=4
(ms) | Speedup |
|---|---|---|---|---|---|---|
| small | 64x10000 | 1% | 1 | 46.585 | 15.993 | 2.91 |
| small | 64x10000 | 10% | 6 | 52.774 | 16.794 | 3.14 |
| small | 64x10000 | 50% | 32 | 71.740 | 34.172 | 2.10 |
| small | 64x10000 | 100% | 64 | 105.375 | 58.113 | 1.81 |
| large | 64x200000 | 1% | 1 | 995.851 | 289.124 | 3.44 |
| large | 64x200000 | 10% | 6 | 1117.002 | 335.624 | 3.33 |
| large | 64x200000 | 50% | 32 | 1553.361 | 798.953 | 1.94 |
| large | 64x200000 | 100% | 64 | 2253.123 | 1512.326 | 1.49 |
`Row` and `Col` show consistent speedups, decreasing as the share of unique
values grows: at a low share the output is narrow and the runtime is dominated
by the parallel scan, while at 100% the second pass writes a full-size output
and becomes memory-bound.
`RowCol` still does not benefit. Its thread-local sets must be merged into
one global set, and the merge cost grows with the total distinct count, so at a
100% share the merge re-inserts every distinct value and effectively repeats
the sequential work on top of the parallel pass. At low shares the merge is
cheap, but the concurrent construction of the per-thread `HashSet<Double>`
instances costs more than the saved scan time. Making `RowCol` scale would need
a different strategy — a shared concurrent set, a tree-shaped merge, or a
primitive-keyed hash map to avoid boxing — which seems out of scope for this PR.
--
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]
