janniklinde commented on code in PR #2529:
URL: https://github.com/apache/systemds/pull/2529#discussion_r3518320710
##########
src/main/java/org/apache/sysds/runtime/functionobjects/COV.java:
##########
@@ -63,6 +63,8 @@ private COV() {
public Data execute(Data in1, double u, double v, double w2)
{
CmCovObject cov1=(CmCovObject) in1;
+ if(w2 == 0)
+ return cov1;
Review Comment:
Why is this change necessary?
##########
src/main/java/org/apache/sysds/runtime/instructions/ooc/TSMMOOCInstruction.java:
##########
@@ -59,39 +64,90 @@ public static TSMMOOCInstruction parseInstruction(String
str) {
}
@Override
- public void processInstruction( ExecutionContext ec ) {
+ public void processInstruction(ExecutionContext ec) {
MatrixObject min = ec.getMatrixObject(input1);
- int nRows = (int) min.getDataCharacteristics().getRows();
- int nCols = (int) min.getDataCharacteristics().getCols();
- int bLen = min.getDataCharacteristics().getBlocksize();
-
- OOCStream<IndexedMatrixValue> qIn = min.getStreamHandle();
+ int numRowBlocks =
Math.toIntExact(min.getDataCharacteristics().getNumRowBlocks());
+ int numColBlocks =
Math.toIntExact(min.getDataCharacteristics().getNumColBlocks());
+ int blocksPerJoinGroup = _type.isLeft() ? numColBlocks :
numRowBlocks;
+ int partialsPerOutput = _type.isLeft() ? numRowBlocks :
numColBlocks;
+
+ OOCStreamable<IndexedMatrixValue> inputStreamable =
min.getStreamable();
+ final boolean createdCache = !inputStreamable.hasStreamCache();
+ final CachingStream inputCache = createdCache ? new
CachingStream(min.getStreamHandle())
+ : inputStreamable.getStreamCache();
+
+ OOCStream<List<IndexedMatrixValue>> groupedPartials =
createWritableStream();
+ OOCStream<IndexedMatrixValue> partials = createWritableStream();
+ OOCStream<IndexedMatrixValue> out = createWritableStream();
+ addOutStream(out);
+ ec.getMatrixObject(output).setStreamHandle(out);
+
+ CompletableFuture<Void> joinFuture =
joinManyOOC(inputCache.getReadStream(), inputCache.getReadStream(),
groupedPartials,
+ this::createPartialOutputTiles, this::getJoinIndex,
this::getJoinIndex,
+ blocksPerJoinGroup, blocksPerJoinGroup);
+ CompletableFuture<Void> expandFuture =
expandOOC(groupedPartials, partials, values -> values);
+
BinaryOperator plus =
InstructionUtils.parseBinaryOperator(Opcodes.PLUS.toString());
+ CompletableFuture<Void> outFuture = groupedReduceOOC(partials,
out, (left, right) -> {
+ MatrixBlock result = ((MatrixBlock)
left.getValue()).binaryOperations(plus, right.getValue());
+ left.setValue(result);
+ return left;
+ }, partialsPerOutput);
+
+ propagateFailuresToOutput(out, List.of(joinFuture,
expandFuture, outFuture));
+
+ outFuture.whenComplete((result, error) -> {
+ if(createdCache)
+ inputCache.scheduleDeletion();
+ });
+ }
+
+ private long getJoinIndex(IndexedMatrixValue value) {
+ return _type.isLeft() ? value.getIndexes().getRowIndex() :
value.getIndexes().getColumnIndex();
+ }
- //validation check TODO extend compiler to not create OOC
otherwise
- if( (_type.isLeft() && nCols > bLen)
- || (_type.isRight() && nRows > bLen) )
- {
- throw new UnsupportedOperationException();
+ private long getOutputIndex(IndexedMatrixValue value) {
+ return _type.isLeft() ? value.getIndexes().getColumnIndex() :
value.getIndexes().getRowIndex();
+ }
+
+ private List<IndexedMatrixValue>
createPartialOutputTiles(IndexedMatrixValue left, IndexedMatrixValue right) {
+ long leftIndex = getOutputIndex(left);
+ long rightIndex = getOutputIndex(right);
+ if(leftIndex > rightIndex)
+ return List.of();
+
+ MatrixBlock leftBlock = (MatrixBlock) left.getValue();
+ MatrixBlock rightBlock = (MatrixBlock) right.getValue();
+ if(leftIndex == rightIndex) {
+ MatrixBlock diagonal =
leftBlock.transposeSelfMatrixMultOperations(new MatrixBlock(), _type);
+ return List.of(new IndexedMatrixValue(new
MatrixIndexes(leftIndex, rightIndex), diagonal));
}
-
- //int dim = _type.isLeft() ? nCols : nRows;
- MatrixBlock resultBlock = null;
-
- OOCStream<MatrixBlock> tmpStream = createWritableStream();
-
- mapOOC(qIn, tmpStream,
- tmp -> ((MatrixBlock) tmp.getValue())
- .transposeSelfMatrixMultOperations(new
MatrixBlock(), _type));
-
- MatrixBlock tmp;
- while ((tmp = tmpStream.dequeue()) !=
LocalTaskQueue.NO_MORE_TASKS) {
- if (resultBlock == null)
- resultBlock = tmp;
- else
- resultBlock.binaryOperationsInPlace(plus, tmp);
Review Comment:
These changes look good for the general case. However, the more general
execution path is more expensive. I'd like you to keep the special case where
we can avoid creating a `CachingStream` and using the heavier primitives (when
only a single output tile is produced).
##########
src/test/java/org/apache/sysds/test/functions/ooc/CovarianceTest.java:
##########
Review Comment:
Please assert that OOC operators were used (similar to the tsmm test)
##########
src/test/java/org/apache/sysds/test/functions/ooc/CovarianceWeightsTest.java:
##########
Review Comment:
Assert that OOC op was used
--
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]
