Skip to content

[SYSTEMDS-???] Parameterserver aggregation optimization #1211

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
Baunsgaard wants to merge 2 commits into from
+37 −10
Closed

[SYSTEMDS-???] Parameterserver aggregation optimization #1211

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7cf4b23
[MINOR] Fix native warning on sparse MM
Baunsgaard Mar 24, 2021
2dfecc6
[MINOR] AggregateFinalResults uniform format
Baunsgaard Mar 24, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 10 additions & 2 deletions src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixAgg.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -274,7 +274,7 @@ public static void aggregateUnaryMatrix(MatrixBlock in, MatrixBlock out, Aggrega
pool.shutdown();
//aggregate partial results
if( !(uaop.indexFn instanceof ReduceCol) ) {
out.copy(((PartialAggTask)tasks.get(0)).getResult()); //for init
out.copy(((PartialAggTask)tasks.get(0)).getResult(), false); //for init
for( int i=1; i<tasks.size(); i++ )
aggregateFinalResult(uaop.aggOp, out, ((PartialAggTask)tasks.get(i)).getResult());
}
Expand Down Expand Up @@ -468,7 +468,7 @@ public static MatrixBlock aggregateTernary(MatrixBlock in1, MatrixBlock in2, Mat
List<Future<MatrixBlock>> rtasks = pool.invokeAll(tasks);
pool.shutdown();
//aggregate partial results and error handling
ret.copy(rtasks.get(0).get()); //for init
ret.copy(rtasks.get(0).get(), false); //for init
for( int i=1; i<rtasks.size(); i++ )
aggregateFinalResult(op.aggOp, ret, rtasks.get(i).get());
}
Expand Down Expand Up @@ -674,6 +674,14 @@ private static void aggregateFinalResult( AggregateOperator aop, MatrixBlock out
laop = new AggregateOperator(0, KahanPlus.getKahanPlusFnObject(), aop.correction);
}

if( out.isInSparseFormat() != partout.isInSparseFormat() ){
if(partout.isInSparseFormat())
partout.sparseToDense();

if(out.isInSparseFormat())
out.sparseToDense();
}

//incremental aggregation of final results
if( laop.existsCorrection() )
out.incrementalAggregate(laop, partout);
Expand Down
20 changes: 12 additions & 8 deletions src/main/java/org/apache/sysds/runtime/matrix/data/LibMatrixNative.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,11 +79,12 @@ public static void matrixMult(MatrixBlock m1, MatrixBlock m2, MatrixBlock ret, i
return;
}

if( NativeHelper.isNativeLibraryLoaded()
&& !isMatMultMemoryBound(m1.rlen, m1.clen, m2.clen)
boolean isValidForNative = !isMatMultMemoryBound(m1.rlen, m1.clen, m2.clen)
&& !m1.isInSparseFormat() && !m2.isInSparseFormat()
&& m1.getDenseBlock().isContiguous() && m2.getDenseBlock().isContiguous()
&& 8L * ret.getLength() < Integer.MAX_VALUE ) //contiguous but not allocated
&& m1.getDenseBlock().isContiguous() && m2.getDenseBlock().isContiguous() //contiguous but not allocated
&& 8L * ret.getLength() < Integer.MAX_VALUE;

if( NativeHelper.isNativeLibraryLoaded() && isValidForNative )
{
ret.sparse = false;
ret.allocateDenseBlock();
Expand Down Expand Up @@ -114,11 +115,14 @@ public static void matrixMult(MatrixBlock m1, MatrixBlock m2, MatrixBlock ret, i
}
//else record failure and fallback to java
Statistics.incrementNativeFailuresCounter();
LOG.warn("matrixMult: Native mat mult failed. Falling back to java version ("
+ "loaded=" + NativeHelper.isNativeLibraryLoaded()
+ ", sparse=" + (m1.isInSparseFormat() | m2.isInSparseFormat()) + ")");
}
//fallback to default java implementation
LOG.warn("matrixMult: Native mat mult failed. Falling back to java version ("
+ "loaded=" + NativeHelper.isNativeLibraryLoaded()
+ ", sparse=" + (m1.isInSparseFormat() | m2.isInSparseFormat()) + ")");

if(isValidForNative)
LOG.debug("Was valid for native MM but native lib was not loaded");

if (k == 1)
LibMatrixMult.matrixMult(m1, m2, ret, !examSparsity);
else
Expand Down
15 changes: 15 additions & 0 deletions src/main/java/org/apache/sysds/runtime/matrix/data/MatrixValue.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -96,7 +96,22 @@ public MatrixValue(MatrixValue that) {
public abstract void reset(int rl, int cl, boolean sp, long nnzs);
public abstract void reset(int rl, int cl, double v);

/**
* Copy this MatrixValue into that MatrixValue.
*
* If the MatrixValue is a MatrixBlock evaluate the sparsity of the original matrix,
* and copy into either a sparse or a dense matrix.
*
* @param that object to copy the values into.
*/
public abstract void copy(MatrixValue that);

/**
* Copy this MatrixValue into that MatrixValue. But select sparse destination block depending on boolean parameter.
*
* @param that object to copy the values into.
* @param sp boolean specifying if output should be forced sparse or dense. (only applicable if the 'that' is a MatrixBlock)
*/
public abstract void copy(MatrixValue that, boolean sp);

public abstract MatrixValue scalarOperations(ScalarOperator op, MatrixValue result);
Expand Down