Skip to content

[SYSTEMDS-3891] OOC Statistics and Improved Error Propagation #2572

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

Merged
janniklinde merged 1 commit into from
Jul 31, 2026
Merged

[SYSTEMDS-3891] OOC Statistics and Improved Error Propagation #2572

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
14 changes: 11 additions & 3 deletions src/main/java/org/apache/sysds/runtime/instructions/ooc/SubscribableTaskQueue.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@ public class SubscribableTaskQueue<T> extends LocalTaskQueue<OOCStream.QueueCall

private final AtomicInteger _availableCtr = new AtomicInteger(1);
private final AtomicBoolean _closed = new AtomicBoolean(false);
private final AtomicBoolean _terminalDelivered = new AtomicBoolean(false);
private final AtomicInteger _blockCount = new AtomicInteger(0);
private QueueCallback<T> _lastDequeued = null;
private CacheableData<?> _cdata;
Expand Down Expand Up @@ -146,6 +147,7 @@ public T dequeue() {
_lastDequeued = deq;
return deq.get();
}
_terminalDelivered.set(true);
return null;
}
catch(InterruptedException e) {
Expand All @@ -167,6 +169,8 @@ public OOCStream.QueueCallback<T> dequeueCB() {
onDeliveryFinished();
_lastDequeued = deq;
}
else
_terminalDelivered.set(true);
return deq == NO_MORE_TASKS ? null : deq;
}
catch(InterruptedException e) {
Expand Down Expand Up @@ -239,8 +243,10 @@ private void onDeliveryFinished() {
if(ctr == 0) {
validateBlockCountOnClose();
Consumer<QueueCallback<T>> s = _subscriber;
if(s != null)
if(s != null) {
s.accept(OOCStream.eos(_failure));
_terminalDelivered.set(true);
}

if(OOCWatchdog.WATCH)
OOCWatchdog.registerClose(_watchdogId);
Expand All @@ -250,12 +256,14 @@ private void onDeliveryFinished() {
@Override
public synchronized void propagateFailure(DMLRuntimeException re) {
// Ignore late failures
if(_closed.get() && _availableCtr.get() == 0)
if(_terminalDelivered.get())
return;
super.propagateFailure(re);
Consumer<QueueCallback<T>> s = _subscriber;
if(s != null)
if(s != null) {
s.accept(new SimpleQueueCallback<>(null, re));
_terminalDelivered.set(true);
}
}

@Override
Expand Down
6 changes: 6 additions & 0 deletions src/main/java/org/apache/sysds/runtime/ooc/memory/GlobalMemoryBroker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,8 @@

package org.apache.sysds.runtime.ooc.memory;

import org.apache.sysds.utils.Statistics;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
Expand Down Expand Up @@ -137,11 +139,14 @@ public void reservationBlocked(MemoryAllowance allowance, long bytes) {
}

private void runReclaim() {
Statistics.incrementOOCMemoryReclaimRun();
long nanos = System.nanoTime();
try {
long reclaimed = 0;
for(MemoryAllowance allowance : _allowances)
if(!allowance.isShutdown())
reclaimed += allowance.reclaimUnused();
Statistics.accumulateOOCMemoryReclaimBytes(reclaimed);
if(reclaimed == 0)
return;

Expand All @@ -155,6 +160,7 @@ private void runReclaim() {
notifyReservationWaiters();
}
finally {
Statistics.accumulateOOCMemoryReclaimTime(System.nanoTime() - nanos);
if(shouldRetryReclaim())
RECLAIM_EXECUTOR.schedule(this::runReclaim, RECLAIM_RETRY_DELAY_MS, TimeUnit.MILLISECONDS);
else {
Expand Down
66 changes: 51 additions & 15 deletions src/main/java/org/apache/sysds/utils/Statistics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -233,6 +233,9 @@ public Object getMeta(String key) {
private static final LongAdder oocEvictionWriteCalls = new LongAdder();
private static final LongAdder oocEvictionWriteTimeNanos = new LongAdder();
private static final LongAdder oocEvictionWriteBytesSize = new LongAdder();
private static final LongAdder oocMemoryReclaimRuns = new LongAdder();
private static final LongAdder oocMemoryReclaimTime = new LongAdder();
private static final LongAdder oocMemoryReclaimBytes = new LongAdder();
private static final AtomicLong oocStatsStartTime = new AtomicLong(System.nanoTime());

public static long getNoOfExecutedSPInst() {
Expand Down Expand Up @@ -362,6 +365,9 @@ public static void resetOOCEvictionStats() {
oocEvictionWriteCalls.reset();
oocEvictionWriteTimeNanos.reset();
oocEvictionWriteBytesSize.reset();
oocMemoryReclaimRuns.reset();
oocMemoryReclaimTime.reset();
oocMemoryReclaimBytes.reset();
oocStatsStartTime.set(System.nanoTime());
}

Expand Down Expand Up @@ -481,6 +487,18 @@ public static void accumulateOOCEvictionWriteBytes(long bytes) {
oocEvictionWriteBytesSize.add(bytes);
}

public static void incrementOOCMemoryReclaimRun() {
oocMemoryReclaimRuns.increment();
}

public static void accumulateOOCMemoryReclaimTime(long nanos) {
oocMemoryReclaimTime.add(nanos);
}

public static void accumulateOOCMemoryReclaimBytes(long bytes) {
oocMemoryReclaimBytes.add(bytes);
}

public static String displayOOCEvictionStats() {
long elapsedNanos = Math.max(1, System.nanoTime() - oocStatsStartTime.get());
double elapsedSeconds = elapsedNanos / 1e9;
Expand All @@ -499,6 +517,9 @@ public static String displayOOCEvictionStats() {
oocLoadFromDiskCalls.longValue(), oocLoadFromDiskTimeNanos.longValue() / 1e9, oocLoadFromDiskBytesSize.longValue() / 1e9));
sb.append(String.format(Locale.US, " evict writes:\t\t%d (time %.3f sec, %.3f GB)\n",
oocEvictionWriteCalls.longValue(), oocEvictionWriteTimeNanos.longValue() / 1e9, oocEvictionWriteBytesSize.longValue() / 1e9));
sb.append(String.format(Locale.US, " reclaim runs:\t\t%d (time %.3f sec, %.3f GB)\n",
oocMemoryReclaimRuns.longValue(), oocMemoryReclaimTime.longValue() / 1e9,
oocMemoryReclaimBytes.longValue() / 1e9));
return sb.toString();
}

Expand Down Expand Up @@ -540,15 +561,15 @@ public static void reset()
}

public static void resetJITCompileTime(){
jitCompileTime = -1 * getJITCompileTime();
jitCompileTime = -1 * getCurrentJITCompileTime();
}

public static void resetJVMgcTime(){
jvmGCTime = -1 * getJVMgcTime();
jvmGCTime = -1 * getCurrentJVMgcTime();
}

public static void resetJVMgcCount(){
jvmGCTime = -1 * getJVMgcCount();
jvmGCCount = -1 * getCurrentJVMgcCount();
}

public static void resetCPHeavyHitters(){
Expand Down Expand Up @@ -1117,38 +1138,53 @@ private static String byteCountToDisplaySize(double numBytes) {
* @return JIT compile time
*/
public static long getJITCompileTime(){
long ret = -1; //unsupported
long ret = getCurrentJITCompileTime();
if(ret >= 0)
ret += jitCompileTime; // add from remote processes
return ret;
}

private static long getCurrentJITCompileTime() {
long ret = -1; // unsupported
CompilationMXBean cmx = ManagementFactory.getCompilationMXBean();
if( cmx.isCompilationTimeMonitoringSupported() ) {
if(cmx.isCompilationTimeMonitoringSupported())
ret = cmx.getTotalCompilationTime();
ret += jitCompileTime; //add from remote processes
}
return ret;
}

public static long getJVMgcTime(){
long ret = 0;
long ret = getCurrentJVMgcTime();
if(ret > 0)
ret += jvmGCTime;

return ret;
}

private static long getCurrentJVMgcTime() {
long ret = 0;

List<GarbageCollectorMXBean> gcxs = ManagementFactory.getGarbageCollectorMXBeans();

for( GarbageCollectorMXBean gcx : gcxs )
ret += gcx.getCollectionTime();
return ret;
}

public static long getJVMgcCount() {
long ret = getCurrentJVMgcCount();
if( ret>0 )
ret += jvmGCTime;
ret += jvmGCCount;

return ret;
}
public static long getJVMgcCount(){
long ret = 0;

private static long getCurrentJVMgcCount() {
long ret = 0;

List<GarbageCollectorMXBean> gcxs = ManagementFactory.getGarbageCollectorMXBeans();

for( GarbageCollectorMXBean gcx : gcxs )
ret += gcx.getCollectionCount();
if( ret>0 )
ret += jvmGCCount;

return ret;
}

Expand Down
16 changes: 14 additions & 2 deletions src/test/java/org/apache/sysds/test/component/ooc/OOCInstructionUtilsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,11 +26,16 @@
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

import org.apache.sysds.common.Types.FileFormat;
import org.apache.sysds.common.Types.ValueType;
import org.apache.sysds.runtime.DMLRuntimeException;
import org.apache.sysds.runtime.controlprogram.caching.MatrixObject;
import org.apache.sysds.runtime.instructions.ooc.SubscribableTaskQueue;
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.meta.MatrixCharacteristics;
import org.apache.sysds.runtime.meta.MetaDataFormat;
import org.apache.sysds.runtime.ooc.cache.OOCFuture;
import org.apache.sysds.runtime.ooc.store.MaterializedCallback;
import org.apache.sysds.runtime.ooc.store.StoreLease;
Expand Down Expand Up @@ -78,7 +83,9 @@ public void testSubmitTasksWaitsForAllStreams() throws Exception {

@Test
public void testSubmitTaskPropagatesFailure() throws Exception {
SubscribableTaskQueue<Integer> output = new SubscribableTaskQueue<>();
SubscribableTaskQueue<IndexedMatrixValue> output = new SubscribableTaskQueue<>();
output.setData(new MatrixObject(ValueType.FP64, "/dev/null",
new MetaDataFormat(new MatrixCharacteristics(1, 1, 1), FileFormat.BINARY)));
AtomicReference<DMLRuntimeException> propagated = new AtomicReference<>();
output.setSubscriber(callback -> {
try(callback) {
Expand All @@ -92,6 +99,12 @@ public void testSubmitTaskPropagatesFailure() throws Exception {
}
}
});
try {
output.closeInput();
Assert.fail("Expected block-count failure");
}
catch(DMLRuntimeException expected) {
}

OOCFuture<Void> completion = OOCInstructionUtils.submitOOCTask(() -> {
throw new DMLRuntimeException("injected failure");
Expand All @@ -103,7 +116,6 @@ public void testSubmitTaskPropagatesFailure() throws Exception {
catch(ExecutionException expected) {
Assert.assertTrue(expected.getCause() instanceof DMLRuntimeException);
}
output.closeInput();
Assert.assertNotNull(propagated.get());
Assert.assertEquals("injected failure", propagated.get().getMessage());
}
Expand Down
Loading