From 65dd3200f467d60cf95d629443e12276c8a13cad Mon Sep 17 00:00:00 2001 From: Jannik Lindemann Date: Fri, 17 Jul 2026 14:28:53 +0200 Subject: [PATCH] [OOC] Bugfixes and Improved Error Propagation --- .../ooc/SubscribableTaskQueue.java | 14 +++- .../ooc/memory/GlobalMemoryBroker.java | 6 ++ .../org/apache/sysds/utils/Statistics.java | 66 ++++++++++++++----- .../ooc/OOCInstructionUtilsTest.java | 16 ++++- 4 files changed, 82 insertions(+), 20 deletions(-) diff --git a/src/main/java/org/apache/sysds/runtime/instructions/ooc/SubscribableTaskQueue.java b/src/main/java/org/apache/sysds/runtime/instructions/ooc/SubscribableTaskQueue.java index 5400b6ba98f..9a449e8b331 100644 --- a/src/main/java/org/apache/sysds/runtime/instructions/ooc/SubscribableTaskQueue.java +++ b/src/main/java/org/apache/sysds/runtime/instructions/ooc/SubscribableTaskQueue.java @@ -35,6 +35,7 @@ public class SubscribableTaskQueue extends LocalTaskQueue _lastDequeued = null; private CacheableData _cdata; @@ -146,6 +147,7 @@ public T dequeue() { _lastDequeued = deq; return deq.get(); } + _terminalDelivered.set(true); return null; } catch(InterruptedException e) { @@ -167,6 +169,8 @@ public OOCStream.QueueCallback dequeueCB() { onDeliveryFinished(); _lastDequeued = deq; } + else + _terminalDelivered.set(true); return deq == NO_MORE_TASKS ? null : deq; } catch(InterruptedException e) { @@ -239,8 +243,10 @@ private void onDeliveryFinished() { if(ctr == 0) { validateBlockCountOnClose(); Consumer> s = _subscriber; - if(s != null) + if(s != null) { s.accept(OOCStream.eos(_failure)); + _terminalDelivered.set(true); + } if(OOCWatchdog.WATCH) OOCWatchdog.registerClose(_watchdogId); @@ -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> s = _subscriber; - if(s != null) + if(s != null) { s.accept(new SimpleQueueCallback<>(null, re)); + _terminalDelivered.set(true); + } } @Override diff --git a/src/main/java/org/apache/sysds/runtime/ooc/memory/GlobalMemoryBroker.java b/src/main/java/org/apache/sysds/runtime/ooc/memory/GlobalMemoryBroker.java index f7ad7b28577..a4e847deeef 100644 --- a/src/main/java/org/apache/sysds/runtime/ooc/memory/GlobalMemoryBroker.java +++ b/src/main/java/org/apache/sysds/runtime/ooc/memory/GlobalMemoryBroker.java @@ -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; @@ -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; @@ -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 { diff --git a/src/main/java/org/apache/sysds/utils/Statistics.java b/src/main/java/org/apache/sysds/utils/Statistics.java index 5102933911a..c257495c575 100644 --- a/src/main/java/org/apache/sysds/utils/Statistics.java +++ b/src/main/java/org/apache/sysds/utils/Statistics.java @@ -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() { @@ -362,6 +365,9 @@ public static void resetOOCEvictionStats() { oocEvictionWriteCalls.reset(); oocEvictionWriteTimeNanos.reset(); oocEvictionWriteBytesSize.reset(); + oocMemoryReclaimRuns.reset(); + oocMemoryReclaimTime.reset(); + oocMemoryReclaimBytes.reset(); oocStatsStartTime.set(System.nanoTime()); } @@ -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; @@ -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(); } @@ -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(){ @@ -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 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 gcxs = ManagementFactory.getGarbageCollectorMXBeans(); for( GarbageCollectorMXBean gcx : gcxs ) ret += gcx.getCollectionCount(); - if( ret>0 ) - ret += jvmGCCount; - return ret; } diff --git a/src/test/java/org/apache/sysds/test/component/ooc/OOCInstructionUtilsTest.java b/src/test/java/org/apache/sysds/test/component/ooc/OOCInstructionUtilsTest.java index f4bab5d9813..42ca04cbbb6 100644 --- a/src/test/java/org/apache/sysds/test/component/ooc/OOCInstructionUtilsTest.java +++ b/src/test/java/org/apache/sysds/test/component/ooc/OOCInstructionUtilsTest.java @@ -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; @@ -78,7 +83,9 @@ public void testSubmitTasksWaitsForAllStreams() throws Exception { @Test public void testSubmitTaskPropagatesFailure() throws Exception { - SubscribableTaskQueue output = new SubscribableTaskQueue<>(); + SubscribableTaskQueue output = new SubscribableTaskQueue<>(); + output.setData(new MatrixObject(ValueType.FP64, "/dev/null", + new MetaDataFormat(new MatrixCharacteristics(1, 1, 1), FileFormat.BINARY))); AtomicReference propagated = new AtomicReference<>(); output.setSubscriber(callback -> { try(callback) { @@ -92,6 +99,12 @@ public void testSubmitTaskPropagatesFailure() throws Exception { } } }); + try { + output.closeInput(); + Assert.fail("Expected block-count failure"); + } + catch(DMLRuntimeException expected) { + } OOCFuture completion = OOCInstructionUtils.submitOOCTask(() -> { throw new DMLRuntimeException("injected failure"); @@ -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()); }