From a2c3aa676f48b1c91285b148df62a5b7c110bcaf Mon Sep 17 00:00:00 2001 From: Janardhan Pulivarthi Date: Wed, 3 Sep 2025 22:11:03 +0530 Subject: [PATCH 1/2] tsmm operation --- .../org/apache/sysds/hops/AggBinaryOp.java | 18 +-- .../instructions/OOCInstructionParser.java | 3 + .../instructions/ooc/TSMMOOCInstruction.java | 112 ++++++++++++++++++ .../sysds/test/functions/ooc/TSMMTest.java | 107 +++++++++++++++++ src/test/scripts/functions/ooc/TSMM.dml | 28 +++++ 5 files changed, 259 insertions(+), 9 deletions(-) create mode 100644 src/main/java/org/apache/sysds/runtime/instructions/ooc/TSMMOOCInstruction.java create mode 100644 src/test/java/org/apache/sysds/test/functions/ooc/TSMMTest.java create mode 100644 src/test/scripts/functions/ooc/TSMM.dml diff --git a/src/main/java/org/apache/sysds/hops/AggBinaryOp.java b/src/main/java/org/apache/sysds/hops/AggBinaryOp.java index fb20cc41d01..0ca5a752765 100644 --- a/src/main/java/org/apache/sysds/hops/AggBinaryOp.java +++ b/src/main/java/org/apache/sysds/hops/AggBinaryOp.java @@ -179,7 +179,7 @@ public Lop constructLops() { et = ExecType.CP; } - if (et == ExecType.CP || et == ExecType.GPU || et == ExecType.FED) { + if (et == ExecType.CP || et == ExecType.GPU || et == ExecType.FED || et == ExecType.OOC) { //matrix mult operation selection part 3 (CP type) _method = optFindMMultMethodCP(input1.getDim1(), input1.getDim2(), input2.getDim1(), input2.getDim2(), mmtsj, chain, _hasLeftPMInput); @@ -240,14 +240,14 @@ public Lop constructLops() { default: throw new HopsException(this.printErrorLocation() + "Invalid Matrix Mult Method (" + _method + ") while constructing SPARK lops."); } - } else if (et == ExecType.OOC) { - Lop in1 = getInput().get(0).constructLops(); - Lop in2 = getInput().get(1).constructLops(); - MatMultCP matmult = new MatMultCP(in1, in2, getDataType(), getValueType(), - et, OptimizerUtils.getConstrainedNumThreads(_maxNumThreads)); - setOutputDimensions(matmult); - setLineNumbers(matmult); - setLops(matmult); +// } else if (et == ExecType.OOC) { +// Lop in1 = getInput().get(0).constructLops(); +// Lop in2 = getInput().get(1).constructLops(); +// MatMultCP matmult = new MatMultCP(in1, in2, getDataType(), getValueType(), +// et, OptimizerUtils.getConstrainedNumThreads(_maxNumThreads)); +// setOutputDimensions(matmult); +// setLineNumbers(matmult); +// setLops(matmult); } } else throw new HopsException(this.printErrorLocation() + "Invalid operation in AggBinary Hop, aggBin(" + innerOp + "," + outerOp + ") while constructing lops."); diff --git a/src/main/java/org/apache/sysds/runtime/instructions/OOCInstructionParser.java b/src/main/java/org/apache/sysds/runtime/instructions/OOCInstructionParser.java index 73b5ca02618..e898800aba1 100644 --- a/src/main/java/org/apache/sysds/runtime/instructions/OOCInstructionParser.java +++ b/src/main/java/org/apache/sysds/runtime/instructions/OOCInstructionParser.java @@ -27,6 +27,7 @@ import org.apache.sysds.runtime.instructions.ooc.BinaryOOCInstruction; import org.apache.sysds.runtime.instructions.ooc.OOCInstruction; import org.apache.sysds.runtime.instructions.ooc.ReblockOOCInstruction; +import org.apache.sysds.runtime.instructions.ooc.TSMMOOCInstruction; import org.apache.sysds.runtime.instructions.ooc.UnaryOOCInstruction; import org.apache.sysds.runtime.instructions.ooc.MatrixVectorBinaryOOCInstruction; import org.apache.sysds.runtime.instructions.ooc.TransposeOOCInstruction; @@ -61,6 +62,8 @@ public static OOCInstruction parseSingleInstruction(InstructionType ooctype, Str case AggregateBinary: case MAPMM: return MatrixVectorBinaryOOCInstruction.parseInstruction(str); + case MMTSJ: + return TSMMOOCInstruction.parseInstruction(str); case Reorg: return TransposeOOCInstruction.parseInstruction(str); diff --git a/src/main/java/org/apache/sysds/runtime/instructions/ooc/TSMMOOCInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/ooc/TSMMOOCInstruction.java new file mode 100644 index 00000000000..080df44b08c --- /dev/null +++ b/src/main/java/org/apache/sysds/runtime/instructions/ooc/TSMMOOCInstruction.java @@ -0,0 +1,112 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.sysds.runtime.instructions.ooc; + +import org.apache.sysds.common.Opcodes; +import org.apache.sysds.conf.ConfigurationManager; +import org.apache.sysds.lops.MMTSJ; +import org.apache.sysds.runtime.DMLRuntimeException; +import org.apache.sysds.runtime.controlprogram.caching.MatrixObject; +import org.apache.sysds.runtime.controlprogram.context.ExecutionContext; +import org.apache.sysds.runtime.controlprogram.parfor.LocalTaskQueue; +import org.apache.sysds.runtime.functionobjects.Multiply; +import org.apache.sysds.runtime.functionobjects.Plus; +import org.apache.sysds.runtime.instructions.InstructionUtils; +import org.apache.sysds.runtime.instructions.cp.CPOperand; +import org.apache.sysds.runtime.instructions.spark.data.IndexedMatrixValue; +import org.apache.sysds.runtime.matrix.data.MatrixBlock; +import org.apache.sysds.runtime.matrix.data.MatrixIndexes; +import org.apache.sysds.runtime.matrix.operators.AggregateBinaryOperator; +import org.apache.sysds.runtime.matrix.operators.AggregateOperator; +import org.apache.sysds.runtime.matrix.operators.BinaryOperator; +import org.apache.sysds.runtime.matrix.operators.Operator; +import org.apache.sysds.runtime.util.CommonThreadPool; + +import java.util.HashMap; +import java.util.concurrent.ExecutorService; + +public class TSMMOOCInstruction extends ComputationOOCInstruction { + private final MMTSJ.MMTSJType _type; + + protected TSMMOOCInstruction(OOCType type, Operator op, CPOperand in1, CPOperand out, MMTSJ.MMTSJType mmtsjType, String opcode, String istr) { + super(type, op, in1, out, opcode, istr); + _type = mmtsjType; + } + + public static TSMMOOCInstruction parseInstruction(String str) { + String[] parts = InstructionUtils.getInstructionPartsWithValueType(str); + InstructionUtils.checkNumFields(parts, 3); + String opcode = parts[0]; + CPOperand in1 = new CPOperand(parts[1]); // the larget matrix (streamed) + CPOperand out = new CPOperand(parts[2]); + MMTSJ.MMTSJType mmtsjType = MMTSJ.MMTSJType.valueOf(parts[3]); + + AggregateOperator agg = new AggregateOperator(0, Plus.getPlusFnObject()); + AggregateBinaryOperator ba = new AggregateBinaryOperator(Multiply.getMultiplyFnObject(), agg); + + return new TSMMOOCInstruction(OOCType.MAPMM, ba, in1, out, mmtsjType, opcode, str); + } + + @Override + public void processInstruction( ExecutionContext ec ) { + // 1. Identify the inputs + MatrixObject min = ec.getMatrixObject(input1); // big matrix + + // number of colBlocks for early block output + long nBlocks = min.getDataCharacteristics().getNumColBlocks(); + int nCols = (int) min.getDataCharacteristics().getCols(); + + LocalTaskQueue qIn = min.getStreamHandle(); + LocalTaskQueue qOut = new LocalTaskQueue<>(); + BinaryOperator plus = InstructionUtils.parseBinaryOperator(Opcodes.PLUS.toString()); + ec.getMatrixObject(output).setStreamHandle(qOut); + + ExecutorService pool = CommonThreadPool.get(); + try { + // Core logic: background thread + pool.submit(() -> { + IndexedMatrixValue tmp = null; + try { + MatrixBlock resultBlock = new MatrixBlock(nCols, nCols, false); + while((tmp = qIn.dequeueTask()) != LocalTaskQueue.NO_MORE_TASKS) { + MatrixBlock matrixBlock = (MatrixBlock) tmp.getValue(); + + // Now, call the operation with the correct, specific operator. + MatrixBlock partialResult = matrixBlock.transposeSelfMatrixMultOperations(new MatrixBlock(), _type); + resultBlock.binaryOperationsInPlace(plus, partialResult); + + } + qOut.enqueueTask(new IndexedMatrixValue(new MatrixIndexes(1, 1), resultBlock)); + } + catch(Exception ex) { + throw new DMLRuntimeException(ex); + } + finally { + qOut.closeInput(); + } + }); + } catch (Exception e) { + throw new DMLRuntimeException(e); + } + finally { + pool.shutdown(); + } + } +} diff --git a/src/test/java/org/apache/sysds/test/functions/ooc/TSMMTest.java b/src/test/java/org/apache/sysds/test/functions/ooc/TSMMTest.java new file mode 100644 index 00000000000..7424bed50fe --- /dev/null +++ b/src/test/java/org/apache/sysds/test/functions/ooc/TSMMTest.java @@ -0,0 +1,107 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.sysds.test.functions.ooc; + +import org.apache.sysds.common.Types; +import org.apache.sysds.lops.MMTSJ; +import org.apache.sysds.runtime.io.MatrixWriter; +import org.apache.sysds.runtime.io.MatrixWriterFactory; +import org.apache.sysds.runtime.matrix.data.LibMatrixMult; +import org.apache.sysds.runtime.matrix.data.MatrixBlock; +import org.apache.sysds.runtime.meta.MatrixCharacteristics; +import org.apache.sysds.runtime.util.DataConverter; +import org.apache.sysds.runtime.util.HDFSTool; +import org.apache.sysds.test.AutomatedTestBase; +import org.apache.sysds.test.TestConfiguration; +import org.apache.sysds.test.TestUtils; +import org.checkerframework.checker.units.qual.A; +import org.junit.Test; + +import java.io.IOException; + +public class TSMMTest extends AutomatedTestBase { + private final static String TEST_NAME1 = "TSMM"; + private final static String TEST_DIR = "functions/ooc/"; + private final static String TEST_CLASS_DIR = TEST_DIR + TSMMTest.class.getSimpleName() + "/"; + private final static double eps = 1e-10; + private static final String INPUT_NAME = "X"; + private static final String OUTPUT_NAME = "res"; + + private final static int rows = 1000; + private final static int cols_wide = 1000; + private final static double sparsity1 = 0.7; + private final static double sparsity2 = 0.1; + private final int k = 1; + + @Override + public void setUp() { + TestUtils.clearAssertionInformation(); + TestConfiguration config = new TestConfiguration(TEST_CLASS_DIR, TEST_NAME1); + addTestConfiguration(TEST_NAME1, config); + } + + @Test + public void testTSMM1() { + runTSMMTest(cols_wide, false); + } + + private void runTSMMTest(int cols, boolean sparse ) + { + Types.ExecMode platformOld = setExecMode(Types.ExecMode.SINGLE_NODE); + + try + { + getAndLoadTestConfiguration(TEST_NAME1); + String HOME = SCRIPT_DIR + TEST_DIR; + fullDMLScriptName = HOME + TEST_NAME1 + ".dml"; + programArgs = new String[]{"-explain", "-stats", "-ooc", + "-args", input(INPUT_NAME), output(OUTPUT_NAME)}; + + // 1. Generate the data in-memory as MatrixBlock objects + double[][] A_data = getRandomMatrix(rows, cols, 0, 1, sparse?sparsity2:sparsity1, 10); + + // 2. Convert the double arrays to MatrixBlock objects + MatrixBlock A_mb = DataConverter.convertToMatrixBlock(A_data); + + // 3. Create a binary matrix writer + MatrixWriter writer = MatrixWriterFactory.createMatrixWriter(Types.FileFormat.BINARY); + + // 4. Write matrix A to a binary SequenceFile + writer.writeMatrixToHDFS(A_mb, input(INPUT_NAME), rows, cols, 1000, A_mb.getNonZeros()); + HDFSTool.writeMetaDataFile(input(INPUT_NAME + ".mtd"), Types.ValueType.FP64, + new MatrixCharacteristics(rows, cols, 1000, A_mb.getNonZeros()), Types.FileFormat.BINARY); + + boolean exceptionExpected = false; + runTest(true, exceptionExpected, null, -1); + + MatrixBlock ret1 = DataConverter.readMatrixFromHDFS(output(OUTPUT_NAME), Types.FileFormat.BINARY, rows, cols, 1000, 1000); + MatrixBlock ret2 = new MatrixBlock(rows, rows, false); + A_mb.transposeSelfMatrixMultOperations(ret2, MMTSJ.MMTSJType.LEFT, k); + TestUtils.compareMatrices(ret1, ret2, 1e-8); + } + catch (IOException e) { + throw new RuntimeException(e); + } + finally { + resetExecMode(platformOld); + } + } + +} \ No newline at end of file diff --git a/src/test/scripts/functions/ooc/TSMM.dml b/src/test/scripts/functions/ooc/TSMM.dml new file mode 100644 index 00000000000..432d2d9daab --- /dev/null +++ b/src/test/scripts/functions/ooc/TSMM.dml @@ -0,0 +1,28 @@ +#------------------------------------------------------------- +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +# +#------------------------------------------------------------- + +# Read input matrix and operator from command line args +X = read($1); + +# Operation under test +res = t(X) %*% X; + +write(res, $2, format="binary") From 7cd9d18376b5b6a7347d6d8c2c3d4e0cf26e7faf Mon Sep 17 00:00:00 2001 From: Janardhan Pulivarthi Date: Thu, 4 Sep 2025 09:31:07 +0530 Subject: [PATCH 2/2] fix comments --- src/main/java/org/apache/sysds/hops/AggBinaryOp.java | 8 -------- .../sysds/runtime/instructions/ooc/OOCInstruction.java | 2 +- .../runtime/instructions/ooc/TSMMOOCInstruction.java | 5 ++--- 3 files changed, 3 insertions(+), 12 deletions(-) diff --git a/src/main/java/org/apache/sysds/hops/AggBinaryOp.java b/src/main/java/org/apache/sysds/hops/AggBinaryOp.java index 0ca5a752765..8685524a3fa 100644 --- a/src/main/java/org/apache/sysds/hops/AggBinaryOp.java +++ b/src/main/java/org/apache/sysds/hops/AggBinaryOp.java @@ -240,14 +240,6 @@ public Lop constructLops() { default: throw new HopsException(this.printErrorLocation() + "Invalid Matrix Mult Method (" + _method + ") while constructing SPARK lops."); } -// } else if (et == ExecType.OOC) { -// Lop in1 = getInput().get(0).constructLops(); -// Lop in2 = getInput().get(1).constructLops(); -// MatMultCP matmult = new MatMultCP(in1, in2, getDataType(), getValueType(), -// et, OptimizerUtils.getConstrainedNumThreads(_maxNumThreads)); -// setOutputDimensions(matmult); -// setLineNumbers(matmult); -// setLops(matmult); } } else throw new HopsException(this.printErrorLocation() + "Invalid operation in AggBinary Hop, aggBin(" + innerOp + "," + outerOp + ") while constructing lops."); diff --git a/src/main/java/org/apache/sysds/runtime/instructions/ooc/OOCInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/ooc/OOCInstruction.java index 0495dcfde51..60244fd7de5 100644 --- a/src/main/java/org/apache/sysds/runtime/instructions/ooc/OOCInstruction.java +++ b/src/main/java/org/apache/sysds/runtime/instructions/ooc/OOCInstruction.java @@ -30,7 +30,7 @@ public abstract class OOCInstruction extends Instruction { protected static final Log LOG = LogFactory.getLog(OOCInstruction.class.getName()); public enum OOCType { - Reblock, AggregateUnary, Binary, Unary, MAPMM, Reorg, AggregateBinary + Reblock, AggregateUnary, Binary, Unary, MAPMM, Reorg, AggregateBinary, MMTSJ } protected final OOCInstruction.OOCType _ooctype; diff --git a/src/main/java/org/apache/sysds/runtime/instructions/ooc/TSMMOOCInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/ooc/TSMMOOCInstruction.java index 080df44b08c..a36d18067a8 100644 --- a/src/main/java/org/apache/sysds/runtime/instructions/ooc/TSMMOOCInstruction.java +++ b/src/main/java/org/apache/sysds/runtime/instructions/ooc/TSMMOOCInstruction.java @@ -54,14 +54,14 @@ public static TSMMOOCInstruction parseInstruction(String str) { String[] parts = InstructionUtils.getInstructionPartsWithValueType(str); InstructionUtils.checkNumFields(parts, 3); String opcode = parts[0]; - CPOperand in1 = new CPOperand(parts[1]); // the larget matrix (streamed) + CPOperand in1 = new CPOperand(parts[1]); // the large matrix (streamed), columns <= blocksize CPOperand out = new CPOperand(parts[2]); MMTSJ.MMTSJType mmtsjType = MMTSJ.MMTSJType.valueOf(parts[3]); AggregateOperator agg = new AggregateOperator(0, Plus.getPlusFnObject()); AggregateBinaryOperator ba = new AggregateBinaryOperator(Multiply.getMultiplyFnObject(), agg); - return new TSMMOOCInstruction(OOCType.MAPMM, ba, in1, out, mmtsjType, opcode, str); + return new TSMMOOCInstruction(OOCType.MMTSJ, ba, in1, out, mmtsjType, opcode, str); } @Override @@ -70,7 +70,6 @@ public void processInstruction( ExecutionContext ec ) { MatrixObject min = ec.getMatrixObject(input1); // big matrix // number of colBlocks for early block output - long nBlocks = min.getDataCharacteristics().getNumColBlocks(); int nCols = (int) min.getDataCharacteristics().getCols(); LocalTaskQueue qIn = min.getStreamHandle();