From d459589049e07d254f6a340ca335573a8180b2fa Mon Sep 17 00:00:00 2001 From: Jannik Lindemann Date: Tue, 7 Jul 2026 10:21:15 +0200 Subject: [PATCH 1/3] [OOC] Adapt OOCStream to Forward QueueCallbacks --- .../runtime/instructions/ooc/OOCStream.java | 4 ++ .../instructions/ooc/PlaybackStream.java | 21 ++++++ .../ooc/SubscribableTaskQueue.java | 66 ++++++++++++++----- .../runtime/ooc/stream/FilteredOOCStream.java | 25 +++++-- .../runtime/ooc/stream/MergedOOCStream.java | 19 +++++- .../ooc/stream/SplittingOOCStream.java | 10 +++ .../runtime/ooc/stream/SubOOCStream.java | 15 ++++- 7 files changed, 133 insertions(+), 27 deletions(-) diff --git a/src/main/java/org/apache/sysds/runtime/instructions/ooc/OOCStream.java b/src/main/java/org/apache/sysds/runtime/instructions/ooc/OOCStream.java index ce53e5f0949..b4ffbbbaedb 100644 --- a/src/main/java/org/apache/sysds/runtime/instructions/ooc/OOCStream.java +++ b/src/main/java/org/apache/sysds/runtime/instructions/ooc/OOCStream.java @@ -30,8 +30,12 @@ static QueueCallback eos(DMLRuntimeException e) { void enqueue(T t); + void enqueue(QueueCallback callback); + T dequeue(); + QueueCallback dequeueCB(); + void closeInput(); void propagateFailure(DMLRuntimeException re); diff --git a/src/main/java/org/apache/sysds/runtime/instructions/ooc/PlaybackStream.java b/src/main/java/org/apache/sysds/runtime/instructions/ooc/PlaybackStream.java index 6a67c6602b6..b17b4eb17f4 100644 --- a/src/main/java/org/apache/sysds/runtime/instructions/ooc/PlaybackStream.java +++ b/src/main/java/org/apache/sysds/runtime/instructions/ooc/PlaybackStream.java @@ -52,6 +52,11 @@ public void enqueue(IndexedMatrixValue t) { throw new DMLRuntimeException("Cannot enqueue to a playback stream"); } + @Override + public void enqueue(QueueCallback callback) { + throw new DMLRuntimeException("Cannot enqueue to a playback stream"); + } + @Override public void closeInput() { throw new DMLRuntimeException("Cannot close a playback stream"); @@ -72,6 +77,22 @@ public synchronized IndexedMatrixValue dequeue() { } } + @Override + public synchronized QueueCallback dequeueCB() { + if (_subscriberSet.get()) + throw new IllegalStateException("Cannot dequeue from a playback stream if a subscriber has been set"); + + try { + if (_lastDequeue != null) + _lastDequeue.close(); + _lastDequeue = _streamCache.get(_streamIdx.getAndIncrement()).get(); + return _lastDequeue; + } + catch (InterruptedException | ExecutionException e) { + throw new DMLRuntimeException(e); + } + } + @Override public OOCStream getReadStream() { return _streamCache.getReadStream(); 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 a2030f5a4a4..37a5007d558 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,11 +35,12 @@ import java.util.function.BiFunction; import java.util.function.Consumer; -public class SubscribableTaskQueue extends LocalTaskQueue implements OOCStream { +public class SubscribableTaskQueue extends LocalTaskQueue> implements OOCStream { private final AtomicInteger _availableCtr = new AtomicInteger(1); private final AtomicBoolean _closed = new AtomicBoolean(false); private final AtomicInteger _blockCount = new AtomicInteger(0); + private QueueCallback _lastDequeued = null; private CacheableData _cdata; private volatile Consumer> _subscriber = null; private volatile CopyOnWriteArrayList> _upstreamMsgRelays = null; @@ -71,9 +72,13 @@ private String getCtxMsg() { @Override public void enqueue(T t) { - if (t == NO_MORE_TASKS) - throw new DMLRuntimeException("Cannot enqueue NO_MORE_TASKS item"); + enqueue(new SimpleQueueCallback<>(t, _failure)); + } + @Override + public void enqueue(QueueCallback cb) { + if(cb == NO_MORE_TASKS) + throw new DMLRuntimeException("Cannot enqueue NO_MORE_TASKS item"); int cnt = _availableCtr.incrementAndGet(); if (cnt <= 1) { // Then the queue was already closed and we disallow further enqueues @@ -87,7 +92,7 @@ public void enqueue(T t) { final Consumer> fS = s; if (fS != null) { - fS.accept(new SimpleQueueCallback<>(t, _failure)); + fS.accept(cb); onDeliveryFinished(); return; } @@ -96,7 +101,7 @@ public void enqueue(T t) { // Re-check that subscriber is really null to avoid race conditions if (_subscriber == null) { try { - super.enqueueTask(t); + super.enqueueTask(cb); } catch(InterruptedException e) { throw new DMLRuntimeException(e); @@ -108,7 +113,7 @@ public void enqueue(T t) { } // Last case if due to race a subscriber has been set - s.accept(new SimpleQueueCallback<>(t, _failure)); + s.accept(cb); onDeliveryFinished(); } @@ -128,7 +133,7 @@ protected boolean tryDeliverCallback(QueueCallback cb, int blockCount) { } @Override - public synchronized void enqueueTask(T t) { + public synchronized void enqueueTask(OOCStream.QueueCallback t) { enqueue(t); } @@ -137,10 +142,38 @@ public T dequeue() { try { if (OOCWatchdog.WATCH) OOCWatchdog.addEvent(_watchdogId, "dequeue -- " + getCtxMsg()); - T deq = super.dequeueTask(); - if (deq != NO_MORE_TASKS) + if(_lastDequeued != null) { + _lastDequeued.close(); + _lastDequeued = null; + } + OOCStream.QueueCallback deq = super.dequeueTask(); + if (deq != NO_MORE_TASKS) { + onDeliveryFinished(); + _lastDequeued = deq; + return deq.get(); + } + return null; + } + catch(InterruptedException e) { + throw new DMLRuntimeException(e); + } + } + + @Override + public OOCStream.QueueCallback dequeueCB() { + try { + if (OOCWatchdog.WATCH) + OOCWatchdog.addEvent(_watchdogId, "dequeue -- " + getCtxMsg()); + if(_lastDequeued != null) { + _lastDequeued.close(); + _lastDequeued = null; + } + OOCStream.QueueCallback deq = super.dequeueTask(); + if (deq != NO_MORE_TASKS) { onDeliveryFinished(); - return deq; + _lastDequeued = deq; + } + return deq == NO_MORE_TASKS ? null : deq; } catch(InterruptedException e) { throw new DMLRuntimeException(e); @@ -148,8 +181,8 @@ public T dequeue() { } @Override - public synchronized T dequeueTask() { - return dequeue(); + public synchronized OOCStream.QueueCallback dequeueTask() { + return dequeueCB(); } @Override @@ -180,7 +213,7 @@ public void setSubscriber(Consumer> subscriber) { if(subscriber == null) throw new IllegalArgumentException("Cannot set subscriber to null"); - LinkedList data; + LinkedList> data; boolean needsEos; synchronized(this) { @@ -198,8 +231,8 @@ public void setSubscriber(Consumer> subscriber) { _availableCtr.incrementAndGet(); // route terminal emission via onDeliveryFinished } - for (T t : data) { - subscriber.accept(new SimpleQueueCallback<>(t, _failure)); + for (QueueCallback t : data) { + subscriber.accept(t); onDeliveryFinished(); } @@ -207,7 +240,6 @@ public void setSubscriber(Consumer> subscriber) { onDeliveryFinished(); } - @SuppressWarnings("unchecked") private void onDeliveryFinished() { int ctr = _availableCtr.decrementAndGet(); @@ -215,7 +247,7 @@ private void onDeliveryFinished() { validateBlockCountOnClose(); Consumer> s = _subscriber; if (s != null) - s.accept(new SimpleQueueCallback<>((T) LocalTaskQueue.NO_MORE_TASKS, _failure)); + s.accept(OOCStream.eos(_failure)); if (OOCWatchdog.WATCH) OOCWatchdog.registerClose(_watchdogId); diff --git a/src/main/java/org/apache/sysds/runtime/ooc/stream/FilteredOOCStream.java b/src/main/java/org/apache/sysds/runtime/ooc/stream/FilteredOOCStream.java index 4ad28ae6162..f7f57744390 100644 --- a/src/main/java/org/apache/sysds/runtime/ooc/stream/FilteredOOCStream.java +++ b/src/main/java/org/apache/sysds/runtime/ooc/stream/FilteredOOCStream.java @@ -35,6 +35,7 @@ public class FilteredOOCStream implements OOCStream { private final OOCStream _sourceStream; private final Function _predicate; private CacheableData _data; + private QueueCallback _last; public FilteredOOCStream(OOCStream sourceStream, Function predicate) { _sourceStream = sourceStream; @@ -46,12 +47,26 @@ public void enqueue(T t) { _sourceStream.enqueue(t); } + @Override + public void enqueue(QueueCallback callback) { + _sourceStream.enqueue(callback); + } + @Override public synchronized T dequeue() { - T next; - while((next = _sourceStream.dequeue()) != null) { - if(_predicate.apply(next)) - return next; + QueueCallback cb = dequeueCB(); + return cb == null ? null : cb.get(); + } + + @Override + public synchronized QueueCallback dequeueCB() { + if(_last != null) + _last.close(); + while((_last = _sourceStream.dequeueCB()) != null) { + if(_predicate.apply(_last.get())) + return _last; + _last.close(); + _last = null; } return null; } @@ -101,6 +116,8 @@ public void setSubscriber(Consumer> subscriber) { if(_predicate.apply(cb.get())) subscriber.accept(cb); + else + cb.close(); }); } diff --git a/src/main/java/org/apache/sysds/runtime/ooc/stream/MergedOOCStream.java b/src/main/java/org/apache/sysds/runtime/ooc/stream/MergedOOCStream.java index 8ea13842177..51e38c8cce9 100644 --- a/src/main/java/org/apache/sysds/runtime/ooc/stream/MergedOOCStream.java +++ b/src/main/java/org/apache/sysds/runtime/ooc/stream/MergedOOCStream.java @@ -37,7 +37,7 @@ public class MergedOOCStream implements OOCStream { private final List> _sources; - private final SubscribableTaskQueue> _taskQueue; + private final SubscribableTaskQueue _taskQueue; private final AtomicInteger _openSources; private final AtomicBoolean _failed; private final CachingStream _sharedCache; @@ -132,16 +132,29 @@ public void enqueue(T t) { throw new UnsupportedOperationException(); } + @Override + public void enqueue(QueueCallback callback) { + throw new UnsupportedOperationException(); + } + @Override public synchronized T dequeue() { if(_last != null) _last.close(); - _last = _taskQueue.dequeue(); + _last = _taskQueue.dequeueCB(); if(_last == null) return null; return _last.get(); } + @Override + public synchronized QueueCallback dequeueCB() { + if(_last != null) + _last.close(); + _last = _taskQueue.dequeueCB(); + return _last; + } + @Override public void closeInput() { throw new UnsupportedOperationException(); @@ -183,7 +196,7 @@ public void setSubscriber(Consumer> subscriber) { } return; } - subscriber.accept(cb.get()); + subscriber.accept(cb); }); } diff --git a/src/main/java/org/apache/sysds/runtime/ooc/stream/SplittingOOCStream.java b/src/main/java/org/apache/sysds/runtime/ooc/stream/SplittingOOCStream.java index fb19611c253..b7fd9a1d1a4 100644 --- a/src/main/java/org/apache/sysds/runtime/ooc/stream/SplittingOOCStream.java +++ b/src/main/java/org/apache/sysds/runtime/ooc/stream/SplittingOOCStream.java @@ -110,11 +110,21 @@ public void enqueue(T t) { throw new UnsupportedOperationException(); } + @Override + public void enqueue(QueueCallback callback) { + throw new UnsupportedOperationException(); + } + @Override public T dequeue() { throw new UnsupportedOperationException(); } + @Override + public QueueCallback dequeueCB() { + throw new UnsupportedOperationException(); + } + @Override public void closeInput() { throw new UnsupportedOperationException(); diff --git a/src/main/java/org/apache/sysds/runtime/ooc/stream/SubOOCStream.java b/src/main/java/org/apache/sysds/runtime/ooc/stream/SubOOCStream.java index e5908c18b04..9a231654c26 100644 --- a/src/main/java/org/apache/sysds/runtime/ooc/stream/SubOOCStream.java +++ b/src/main/java/org/apache/sysds/runtime/ooc/stream/SubOOCStream.java @@ -33,7 +33,7 @@ public class SubOOCStream implements OOCStream { private OOCStream _sourceStream; - private SubscribableTaskQueue> _taskQueue; + private SubscribableTaskQueue _taskQueue; private QueueCallback _last; public SubOOCStream(OOCStream sourceStream) { @@ -42,6 +42,7 @@ public SubOOCStream(OOCStream sourceStream) { _taskQueue.setUpstreamMessageRelay(_sourceStream::messageUpstream); } + @Override public void enqueue(QueueCallback callback) { _taskQueue.enqueue(callback); } @@ -55,12 +56,20 @@ public void enqueue(T t) { public synchronized T dequeue() { if(_last != null) _last.close(); - _last = _taskQueue.dequeue(); + _last = _taskQueue.dequeueCB(); if(_last != null) return _last.get(); return null; } + @Override + public synchronized QueueCallback dequeueCB() { + if(_last != null) + _last.close(); + _last = _taskQueue.dequeueCB(); + return _last; + } + @Override public void closeInput() { _taskQueue.closeInput(); @@ -98,7 +107,7 @@ public void setSubscriber(Consumer> subscriber) { } } else - subscriber.accept(cb.get()); + subscriber.accept(cb); }); } From ff790e02ad8584b59db254fcfa6bdfaa6e6f86a8 Mon Sep 17 00:00:00 2001 From: Jannik Lindemann Date: Tue, 7 Jul 2026 10:29:06 +0200 Subject: [PATCH 2/3] [OOC] Reformat --- .../instructions/ooc/PlaybackStream.java | 22 +++---- .../ooc/SubscribableTaskQueue.java | 58 ++++++++++--------- 2 files changed, 41 insertions(+), 39 deletions(-) diff --git a/src/main/java/org/apache/sysds/runtime/instructions/ooc/PlaybackStream.java b/src/main/java/org/apache/sysds/runtime/instructions/ooc/PlaybackStream.java index b17b4eb17f4..3de438bf17b 100644 --- a/src/main/java/org/apache/sysds/runtime/instructions/ooc/PlaybackStream.java +++ b/src/main/java/org/apache/sysds/runtime/instructions/ooc/PlaybackStream.java @@ -64,11 +64,11 @@ public void closeInput() { @Override public synchronized IndexedMatrixValue dequeue() { - if (_subscriberSet.get()) + if(_subscriberSet.get()) throw new IllegalStateException("Cannot dequeue from a playback stream if a subscriber has been set"); try { - if (_lastDequeue != null) + if(_lastDequeue != null) _lastDequeue.close(); _lastDequeue = _streamCache.get(_streamIdx.getAndIncrement()).get(); return _lastDequeue.get(); @@ -79,16 +79,16 @@ public synchronized IndexedMatrixValue dequeue() { @Override public synchronized QueueCallback dequeueCB() { - if (_subscriberSet.get()) + if(_subscriberSet.get()) throw new IllegalStateException("Cannot dequeue from a playback stream if a subscriber has been set"); try { - if (_lastDequeue != null) + if(_lastDequeue != null) _lastDequeue.close(); _lastDequeue = _streamCache.get(_streamIdx.getAndIncrement()).get(); return _lastDequeue; } - catch (InterruptedException | ExecutionException e) { + catch(InterruptedException | ExecutionException e) { throw new DMLRuntimeException(e); } } @@ -135,9 +135,9 @@ public void messageDownstream(OOCStreamMessage msg) { if(msg.isCancelled()) return; CopyOnWriteArrayList> relays = _downstreamRelays; - if (relays != null) { - for (Consumer relay : relays) { - if (msg.isCancelled()) + if(relays != null) { + for(Consumer relay : relays) { + if(msg.isCancelled()) break; relay.accept(msg); } @@ -146,7 +146,7 @@ public void messageDownstream(OOCStreamMessage msg) { @Override public void setSubscriber(Consumer> subscriber) { - if (!_subscriberSet.compareAndSet(false, true)) + if(!_subscriberSet.compareAndSet(false, true)) throw new IllegalArgumentException("Subscriber cannot be set multiple times"); _streamCache.setSubscriber(subscriber, false); @@ -184,10 +184,10 @@ public void addUpstreamMessageRelay(Consumer relay) { @Override public void addDownstreamMessageRelay(Consumer relay) { - if (relay == null) + if(relay == null) throw new IllegalArgumentException("Cannot set downstream relay to null"); CopyOnWriteArrayList> relays = _downstreamRelays; - if (relays == null) { + if(relays == null) { synchronized(this) { if (_downstreamRelays == null) _downstreamRelays = new CopyOnWriteArrayList<>(); 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 37a5007d558..30049b70a50 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 @@ -49,7 +49,7 @@ public class SubscribableTaskQueue extends LocalTaskQueue cb) { throw new DMLRuntimeException("Cannot enqueue NO_MORE_TASKS item"); int cnt = _availableCtr.incrementAndGet(); - if (cnt <= 1) { // Then the queue was already closed and we disallow further enqueues + if(cnt <= 1) { // Then the queue was already closed and we disallow further enqueues _availableCtr.decrementAndGet(); // Undo increment throw new DMLRuntimeException("Cannot enqueue into closed SubscribableTaskQueue"); } @@ -91,15 +91,15 @@ public void enqueue(QueueCallback cb) { Consumer> s = _subscriber; final Consumer> fS = s; - if (fS != null) { + if(fS != null) { fS.accept(cb); onDeliveryFinished(); return; } - synchronized (this) { + synchronized(this) { // Re-check that subscriber is really null to avoid race conditions - if (_subscriber == null) { + if(_subscriber == null) { try { super.enqueueTask(cb); } @@ -119,10 +119,10 @@ public void enqueue(QueueCallback cb) { protected boolean tryDeliverCallback(QueueCallback cb, int blockCount) { Consumer> s = _subscriber; - if (s == null) + if(s == null) return false; int cnt = _availableCtr.incrementAndGet(); - if (cnt <= 1) { // Then the queue was already closed and we disallow further enqueues + if(cnt <= 1) { // Then the queue was already closed and we disallow further enqueues _availableCtr.decrementAndGet(); // Undo increment throw new DMLRuntimeException("Cannot enqueue into closed SubscribableTaskQueue"); } @@ -140,14 +140,14 @@ public synchronized void enqueueTask(OOCStream.QueueCallback t) { @Override public T dequeue() { try { - if (OOCWatchdog.WATCH) + if(OOCWatchdog.WATCH) OOCWatchdog.addEvent(_watchdogId, "dequeue -- " + getCtxMsg()); if(_lastDequeued != null) { _lastDequeued.close(); _lastDequeued = null; } OOCStream.QueueCallback deq = super.dequeueTask(); - if (deq != NO_MORE_TASKS) { + if(deq != NO_MORE_TASKS) { onDeliveryFinished(); _lastDequeued = deq; return deq.get(); @@ -162,14 +162,14 @@ public T dequeue() { @Override public OOCStream.QueueCallback dequeueCB() { try { - if (OOCWatchdog.WATCH) + if(OOCWatchdog.WATCH) OOCWatchdog.addEvent(_watchdogId, "dequeue -- " + getCtxMsg()); if(_lastDequeued != null) { _lastDequeued.close(); _lastDequeued = null; } OOCStream.QueueCallback deq = super.dequeueTask(); - if (deq != NO_MORE_TASKS) { + if(deq != NO_MORE_TASKS) { onDeliveryFinished(); _lastDequeued = deq; } @@ -187,23 +187,25 @@ public synchronized OOCStream.QueueCallback dequeueTask() { @Override public synchronized void closeInput() { - if (_closed.compareAndSet(false, true)) { + if(_closed.compareAndSet(false, true)) { super.closeInput(); onDeliveryFinished(); _upstreamMsgRelays = null; _downstreamMsgRelays = null; - } else { + } + else { throw new IllegalStateException("Multiple close input calls"); } } private void validateBlockCountOnClose() { DataCharacteristics dc = getDataCharacteristics(); - if (dc != null && dc.dimsKnown() && dc.getBlocksize() > 0) { + if(dc != null && dc.dimsKnown() && dc.getBlocksize() > 0) { long expected = OOCUtils.getNumBlocks(dc); - if (expected >= 0 && _blockCount.get() != expected) { - throw new DMLRuntimeException("OOCStream block count mismatch: expected " - + expected + " but saw " + _blockCount.get() + " (" + dc.getRows() + "x" + dc.getCols() + ")"); + if(expected >= 0 && _blockCount.get() != expected) { + throw new DMLRuntimeException( + "OOCStream block count mismatch: expected " + expected + " but saw " + _blockCount.get() + " (" + + dc.getRows() + "x" + dc.getCols() + ")"); } } } @@ -231,7 +233,7 @@ public void setSubscriber(Consumer> subscriber) { _availableCtr.incrementAndGet(); // route terminal emission via onDeliveryFinished } - for (QueueCallback t : data) { + for(QueueCallback t : data) { subscriber.accept(t); onDeliveryFinished(); } @@ -243,13 +245,13 @@ public void setSubscriber(Consumer> subscriber) { private void onDeliveryFinished() { int ctr = _availableCtr.decrementAndGet(); - if (ctr == 0) { + if(ctr == 0) { validateBlockCountOnClose(); Consumer> s = _subscriber; - if (s != null) + if(s != null) s.accept(OOCStream.eos(_failure)); - if (OOCWatchdog.WATCH) + if(OOCWatchdog.WATCH) OOCWatchdog.registerClose(_watchdogId); } } @@ -280,17 +282,17 @@ public void messageUpstream(OOCStreamMessage msg) { if(msg.isCancelled()) return; msg.addIXTransform(_ixTransform); - if (msg.isCancelled()) + if(msg.isCancelled()) return; - if (msg instanceof OOCGetStreamTypeMessage) { - if (_cdata != null) + if(msg instanceof OOCGetStreamTypeMessage) { + if(_cdata != null) ((OOCGetStreamTypeMessage) msg).setInMemoryType(); return; } CopyOnWriteArrayList> relays = _upstreamMsgRelays; if(relays != null) { - for (Consumer relay : relays) { - if (msg.isCancelled()) + for(Consumer relay : relays) { + if(msg.isCancelled()) break; relay.accept(msg); } @@ -304,8 +306,8 @@ public void messageDownstream(OOCStreamMessage msg) { msg.addIXTransform(_ixTransform); CopyOnWriteArrayList> relays = _downstreamMsgRelays; if(relays != null) { - for (Consumer relay : relays) { - if (msg.isCancelled()) + for(Consumer relay : relays) { + if(msg.isCancelled()) break; relay.accept(msg); } From 976c61d99575761b6da304f0b21884cd061ca77f Mon Sep 17 00:00:00 2001 From: Jannik Lindemann Date: Tue, 7 Jul 2026 10:32:47 +0200 Subject: [PATCH 3/3] [OOC] Reformat --- .../runtime/instructions/ooc/SubscribableTaskQueue.java | 5 ++--- 1 file changed, 2 insertions(+), 3 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 30049b70a50..67746b64257 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 @@ -203,9 +203,8 @@ private void validateBlockCountOnClose() { if(dc != null && dc.dimsKnown() && dc.getBlocksize() > 0) { long expected = OOCUtils.getNumBlocks(dc); if(expected >= 0 && _blockCount.get() != expected) { - throw new DMLRuntimeException( - "OOCStream block count mismatch: expected " + expected + " but saw " + _blockCount.get() + " (" + - dc.getRows() + "x" + dc.getCols() + ")"); + throw new DMLRuntimeException("OOCStream block count mismatch: expected " + expected + " but saw " + + _blockCount.get() + " (" + dc.getRows() + "x" + dc.getCols() + ")"); } } }