diff --git a/src/main/java/org/apache/sysds/runtime/instructions/ooc/AggregateTernaryOOCInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/ooc/AggregateTernaryOOCInstruction.java index 2573ede14e4..3e3fafba7b8 100644 --- a/src/main/java/org/apache/sysds/runtime/instructions/ooc/AggregateTernaryOOCInstruction.java +++ b/src/main/java/org/apache/sysds/runtime/instructions/ooc/AggregateTernaryOOCInstruction.java @@ -39,7 +39,6 @@ import org.apache.sysds.runtime.matrix.operators.AggregateTernaryOperator; import org.apache.sysds.runtime.matrix.operators.Operator; import org.apache.sysds.runtime.meta.DataCharacteristics; -import org.apache.sysds.runtime.util.IndexRange; import java.util.ArrayList; import java.util.HashMap; @@ -155,19 +154,6 @@ private void processReduceRow(ExecutionContext ec, AggregateTernaryOperator abOp if(qIn3 != null) streams.add(qIn3); - for (OOCStream stream : streams) - stream.setDownstreamMessageRelay(qOut::messageDownstream); - - qOut.setUpstreamMessageRelay(msg -> - streams.forEach(stream -> stream.messageUpstream(streams.size() > 1 ? msg.split() : msg))); - - qOut.setIXTransform((downstream, range) -> { - if (downstream) - return new IndexRange(1, 1, range.colStart, range.colEnd); - else - return new IndexRange(1, dc.getRows(), range.colStart, range.colEnd); - }); - CompletableFuture fut = joinOOC(streams, qMid, blocks -> { MatrixBlock b1 = (MatrixBlock) blocks.get(0).getValue(); MatrixBlock b2 = (MatrixBlock) blocks.get(1).getValue(); diff --git a/src/main/java/org/apache/sysds/runtime/instructions/ooc/AggregateUnaryOOCInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/ooc/AggregateUnaryOOCInstruction.java index 38b228ccb1d..ac4e9bac919 100644 --- a/src/main/java/org/apache/sysds/runtime/instructions/ooc/AggregateUnaryOOCInstruction.java +++ b/src/main/java/org/apache/sysds/runtime/instructions/ooc/AggregateUnaryOOCInstruction.java @@ -35,7 +35,6 @@ import org.apache.sysds.runtime.matrix.operators.AggregateUnaryOperator; import org.apache.sysds.runtime.matrix.operators.Operator; import org.apache.sysds.runtime.meta.DataCharacteristics; -import org.apache.sysds.runtime.util.IndexRange; import java.util.HashMap; @@ -91,21 +90,6 @@ public void processInstruction( ExecutionContext ec ) { ec.getMatrixObject(output).setStreamHandle(qOut); - qIn.setDownstreamMessageRelay(qOut::messageDownstream); - qOut.setUpstreamMessageRelay(qIn::messageUpstream); - qOut.setIXTransform((downstream, range) -> { - if (downstream) { - if (aggun.isRowAggregate()) - return new IndexRange(range.rowStart, range.rowEnd, 1, 1); - else - return new IndexRange(1, 1, range.colStart, range.colEnd); - } - if (aggun.isRowAggregate()) - return new IndexRange(range.rowStart, range.rowEnd, 1, min.getNumColumns() - 1); - else - return new IndexRange(1, min.getNumRows() - 1, range.colStart, range.colEnd); - }); - // per-block aggregation (parallel map) mapOOC(qIn, qLocal, tmp -> { MatrixIndexes midx = aggun.isRowAggregate() ? diff --git a/src/main/java/org/apache/sysds/runtime/instructions/ooc/BinaryOOCInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/ooc/BinaryOOCInstruction.java index 1352e7ff9c7..8a5f7cf49a7 100644 --- a/src/main/java/org/apache/sysds/runtime/instructions/ooc/BinaryOOCInstruction.java +++ b/src/main/java/org/apache/sysds/runtime/instructions/ooc/BinaryOOCInstruction.java @@ -67,12 +67,6 @@ protected void processMatrixMatrixInstruction(ExecutionContext ec) { OOCStream qIn2 = m2.getStreamHandle(); OOCStream qOut = new SubscribableTaskQueue<>(); ec.getMatrixObject(output).setStreamHandle(qOut); - qIn1.setDownstreamMessageRelay(qOut::messageDownstream); - qIn2.setDownstreamMessageRelay(qOut::messageDownstream); - qOut.setUpstreamMessageRelay(msg -> { - qIn1.messageUpstream(msg.split()); - qIn2.messageUpstream(msg.split()); - }); final boolean known1 = (m1.getNumRows() >= 0 && m1.getNumColumns() >= 0); final boolean known2 = (m2.getNumRows() >= 0 && m2.getNumColumns() >= 0); @@ -152,8 +146,6 @@ protected void processScalarMatrixInstruction(ExecutionContext ec) { OOCStream qIn = min.getStreamHandle(); OOCStream qOut = createWritableStream(); ec.getMatrixObject(output).setStreamHandle(qOut); - qIn.setDownstreamMessageRelay(qOut::messageDownstream); - qOut.setUpstreamMessageRelay(qIn::messageUpstream); mapOOC(qIn, qOut, tmp -> { IndexedMatrixValue tmpOut = new IndexedMatrixValue(); diff --git a/src/main/java/org/apache/sysds/runtime/instructions/ooc/CachingStream.java b/src/main/java/org/apache/sysds/runtime/instructions/ooc/CachingStream.java index abf1efde9c2..ff4362fe285 100644 --- a/src/main/java/org/apache/sysds/runtime/instructions/ooc/CachingStream.java +++ b/src/main/java/org/apache/sysds/runtime/instructions/ooc/CachingStream.java @@ -31,10 +31,7 @@ import org.apache.sysds.runtime.ooc.cache.io.OOCIOHandler; import org.apache.sysds.runtime.ooc.cache.OOCCacheManager; import org.apache.sysds.runtime.ooc.stream.SourceOOCStream; -import org.apache.sysds.runtime.ooc.stream.message.OOCGetStreamTypeMessage; -import org.apache.sysds.runtime.ooc.stream.message.OOCStreamMessage; import org.apache.sysds.runtime.ooc.util.OOCUtils; -import org.apache.sysds.runtime.util.IndexRange; import shaded.parquet.it.unimi.dsi.fastutil.ints.IntArrayList; import java.util.ArrayList; @@ -43,9 +40,7 @@ import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ExecutionException; -import java.util.function.BiFunction; import java.util.function.Consumer; /** @@ -74,7 +69,6 @@ public class CachingStream implements OOCStreamable { private int _numBlocks = 0; private Consumer>[] _subscribers; - private CopyOnWriteArrayList> _downstreamRelays; // state flags private boolean _cacheInProgress = true; // caching in progress, in the first pass. @@ -92,7 +86,6 @@ public CachingStream(OOCStream source) { public CachingStream(OOCStream source, long streamId) { _source = source; - _source.setDownstreamMessageRelay(this::messageDownstream); _streamId = streamId; if(OOCWatchdog.WATCH) { _watchdogId = "CS-" + hashCode(); @@ -100,7 +93,6 @@ public CachingStream(OOCStream source, long streamId) { OOCWatchdog.registerOpen(_watchdogId, toString(), getCtxMsg(), this); } activateIndexing(); - _downstreamRelays = null; source.setSubscriber(tmp -> { try(tmp) { int blk; @@ -623,76 +615,6 @@ public void setData(CacheableData data) { _source.setData(data); } - @Override - public void messageUpstream(OOCStreamMessage msg) { - if (msg.isCancelled()) - return; - if(msg instanceof OOCGetStreamTypeMessage) { - ((OOCGetStreamTypeMessage) msg).setCachedType(); - activateIndexing(); - return; - } - - _source.messageUpstream(msg); - } - - @Override - public void messageDownstream(OOCStreamMessage msg) { - CopyOnWriteArrayList> relays = _downstreamRelays; - if (relays != null) { - for (Consumer relay : relays) { - if (msg.isCancelled()) - break; - relay.accept(msg); - } - } - } - - @Override - public void setUpstreamMessageRelay(Consumer relay) { - throw new UnsupportedOperationException(); - } - - @Override - public void setDownstreamMessageRelay(Consumer relay) { - addDownstreamMessageRelay(relay); - } - - @Override - public void addUpstreamMessageRelay(Consumer relay) { - throw new UnsupportedOperationException(); - } - - @Override - public void addDownstreamMessageRelay(Consumer relay) { - if (relay == null) - throw new IllegalArgumentException("Cannot set downstream relay to null"); - CopyOnWriteArrayList> relays = _downstreamRelays; - if (relays == null) { - synchronized(this) { - if (_downstreamRelays == null) - _downstreamRelays = new CopyOnWriteArrayList<>(); - relays = _downstreamRelays; - } - } - relays.add(0, relay); - } - - @Override - public void clearUpstreamMessageRelays() { - // No upstream relays supported - } - - @Override - public void clearDownstreamMessageRelays() { - _downstreamRelays = null; - } - - @Override - public void setIXTransform(BiFunction transform) { - throw new UnsupportedOperationException(); - } - @SuppressWarnings("unchecked") public void setSubscriber(Consumer> subscriber, boolean incrConsumers) { if(_deletable) diff --git a/src/main/java/org/apache/sysds/runtime/instructions/ooc/MatrixIndexingOOCInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/ooc/MatrixIndexingOOCInstruction.java index 83d4ba58fd2..90e79d2373f 100644 --- a/src/main/java/org/apache/sysds/runtime/instructions/ooc/MatrixIndexingOOCInstruction.java +++ b/src/main/java/org/apache/sysds/runtime/instructions/ooc/MatrixIndexingOOCInstruction.java @@ -123,30 +123,6 @@ public void processInstruction(ExecutionContext ec) { addOutStream(qOut); mOut.setStreamHandle(qOut); - qIn.setDownstreamMessageRelay(qOut::messageDownstream); - qOut.setUpstreamMessageRelay(qIn::messageUpstream); - qOut.setIXTransform((downstream, range) -> { - if(downstream) { - long rs = range.rowStart - ix.rowStart + 1; - long re = range.rowEnd - ix.rowStart + 1; - long cs = range.colStart - ix.colStart + 1; - long ce = range.colEnd - ix.colStart + 1; - // TODO What happens if range is out of bounds? - rs = Math.max(1, rs); - cs = Math.max(1, cs); - re = Math.min(ix.rowSpan(), re); - ce = Math.min(ix.colSpan(), ce); - return new IndexRange(rs, re, cs, ce); - } - else { - long rs = range.rowStart + ix.rowStart; - long re = range.rowEnd + ix.rowStart; - long cs = range.colStart + ix.colStart; - long ce = range.colEnd + ix.colStart; - return new IndexRange(rs, re, cs, ce); - } - }); - if(firstBlockRow == lastBlockRow && firstBlockCol == lastBlockCol) { MatrixIndexes srcBlock = new MatrixIndexes(firstBlockRow + 1, firstBlockCol + 1); OOCStream filteredStream = new FilteredOOCStream<>(qIn, diff --git a/src/main/java/org/apache/sysds/runtime/instructions/ooc/OOCStreamable.java b/src/main/java/org/apache/sysds/runtime/instructions/ooc/OOCStreamable.java index 4f212f544b2..75ccdee3ee1 100644 --- a/src/main/java/org/apache/sysds/runtime/instructions/ooc/OOCStreamable.java +++ b/src/main/java/org/apache/sysds/runtime/instructions/ooc/OOCStreamable.java @@ -21,11 +21,6 @@ import org.apache.sysds.runtime.controlprogram.caching.CacheableData; import org.apache.sysds.runtime.meta.DataCharacteristics; -import org.apache.sysds.runtime.ooc.stream.message.OOCStreamMessage; -import org.apache.sysds.runtime.util.IndexRange; - -import java.util.function.BiFunction; -import java.util.function.Consumer; public interface OOCStreamable { OOCStream getReadStream(); @@ -43,22 +38,4 @@ public interface OOCStreamable { CacheableData getData(); void setData(CacheableData data); - - void messageUpstream(OOCStreamMessage msg); - - void messageDownstream(OOCStreamMessage msg); - - void setUpstreamMessageRelay(Consumer relay); - - void setDownstreamMessageRelay(Consumer relay); - - void addUpstreamMessageRelay(Consumer relay); - - void addDownstreamMessageRelay(Consumer relay); - - void clearUpstreamMessageRelays(); - - void clearDownstreamMessageRelays(); - - void setIXTransform(BiFunction transform); } diff --git a/src/main/java/org/apache/sysds/runtime/instructions/ooc/ParameterizedBuiltinOOCInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/ooc/ParameterizedBuiltinOOCInstruction.java index 87b17f77192..2f71d0e4538 100644 --- a/src/main/java/org/apache/sysds/runtime/instructions/ooc/ParameterizedBuiltinOOCInstruction.java +++ b/src/main/java/org/apache/sysds/runtime/instructions/ooc/ParameterizedBuiltinOOCInstruction.java @@ -98,9 +98,6 @@ public void processInstruction(ExecutionContext ec) { double pattern = Double.parseDouble(params.get("pattern")); double replacement = Double.parseDouble(params.get("replacement")); - qIn.setDownstreamMessageRelay(qOut::messageDownstream); - qOut.setUpstreamMessageRelay(qIn::messageUpstream); - mapOOC(qIn, qOut, tmp -> new IndexedMatrixValue(tmp.getIndexes(), tmp.getValue().replaceOperations(new MatrixBlock(), pattern, replacement))); ec.getMatrixObject(output).setStreamHandle(qOut); @@ -156,9 +153,6 @@ else if(instOpcode.equalsIgnoreCase(Opcodes.REXPAND.toString())) { boolean ignore = Boolean.parseBoolean(params.get("ignore")); long blen = targetObj.getBlocksize(); - qIn.setDownstreamMessageRelay(qOut::messageDownstream); - qOut.setUpstreamMessageRelay(qIn::messageUpstream); - expandOOC(qIn, qOut, tmp -> { ArrayList out = new ArrayList<>(); LibMatrixReorg.rexpand(tmp, lmaxVal, dirRows, cast, ignore, blen, out); 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 3de438bf17b..7526b09f592 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 @@ -23,14 +23,10 @@ import org.apache.sysds.runtime.controlprogram.caching.CacheableData; import org.apache.sysds.runtime.instructions.spark.data.IndexedMatrixValue; import org.apache.sysds.runtime.meta.DataCharacteristics; -import org.apache.sysds.runtime.ooc.stream.message.OOCStreamMessage; -import org.apache.sysds.runtime.util.IndexRange; -import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ExecutionException; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; -import java.util.function.BiFunction; import java.util.function.Consumer; public class PlaybackStream implements OOCStream { @@ -38,7 +34,6 @@ public class PlaybackStream implements OOCStream { private final AtomicInteger _streamIdx; private final AtomicBoolean _subscriberSet; private QueueCallback _lastDequeue; - private volatile CopyOnWriteArrayList> _downstreamRelays; public PlaybackStream(CachingStream streamCache) { this._streamCache = streamCache; @@ -123,27 +118,6 @@ public void setData(CacheableData data) { _streamCache.setData(data); } - @Override - public void messageUpstream(OOCStreamMessage msg) { - if(msg.isCancelled()) - return; - _streamCache.messageUpstream(msg); - } - - @Override - public void messageDownstream(OOCStreamMessage msg) { - if(msg.isCancelled()) - return; - CopyOnWriteArrayList> relays = _downstreamRelays; - if(relays != null) { - for(Consumer relay : relays) { - if(msg.isCancelled()) - break; - relay.accept(msg); - } - } - } - @Override public void setSubscriber(Consumer> subscriber) { if(!_subscriberSet.compareAndSet(false, true)) @@ -166,51 +140,4 @@ public boolean hasStreamCache() { public CachingStream getStreamCache() { return _streamCache; } - - @Override - public void setUpstreamMessageRelay(Consumer relay) { - throw new UnsupportedOperationException(); - } - - @Override - public void setDownstreamMessageRelay(Consumer relay) { - addDownstreamMessageRelay(relay); - } - - @Override - public void addUpstreamMessageRelay(Consumer relay) { - throw new UnsupportedOperationException(); - } - - @Override - public void addDownstreamMessageRelay(Consumer relay) { - if(relay == null) - throw new IllegalArgumentException("Cannot set downstream relay to null"); - CopyOnWriteArrayList> relays = _downstreamRelays; - if(relays == null) { - synchronized(this) { - if (_downstreamRelays == null) - _downstreamRelays = new CopyOnWriteArrayList<>(); - relays = _downstreamRelays; - } - } - relays.add(0, relay); - _streamCache.addDownstreamMessageRelay(relay); - } - - @Override - public void clearUpstreamMessageRelays() { - // No upstream relays supported - } - - @Override - public void clearDownstreamMessageRelays() { - _downstreamRelays = null; - _streamCache.clearDownstreamMessageRelays(); - } - - @Override - public void setIXTransform(BiFunction transform) { - throw new UnsupportedOperationException(); - } } diff --git a/src/main/java/org/apache/sysds/runtime/instructions/ooc/ReorgOOCInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/ooc/ReorgOOCInstruction.java index 273d33341ab..94d896a3546 100644 --- a/src/main/java/org/apache/sysds/runtime/instructions/ooc/ReorgOOCInstruction.java +++ b/src/main/java/org/apache/sysds/runtime/instructions/ooc/ReorgOOCInstruction.java @@ -34,7 +34,6 @@ import org.apache.sysds.runtime.matrix.operators.Operator; import org.apache.sysds.runtime.matrix.operators.ReorgOperator; import org.apache.sysds.runtime.util.DataConverter; -import org.apache.sysds.runtime.util.IndexRange; public class ReorgOOCInstruction extends ComputationOOCInstruction { // sort-specific attributes (to enable variable attributes) @@ -110,11 +109,6 @@ public void processInstruction( ExecutionContext ec ) { OOCStream qOut = createWritableStream(); ec.getMatrixObject(output).setStreamHandle(qOut); - qIn.setDownstreamMessageRelay(qOut::messageDownstream); - qOut.setUpstreamMessageRelay(qIn::messageUpstream); - qOut.setIXTransform((downstream, range) -> - new IndexRange(range.colStart, range.colEnd, range.rowStart, range.rowEnd)); - // Transpose operation mapOOC(qIn, qOut, tmp -> { MatrixBlock inBlock = (MatrixBlock) tmp.getValue(); 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 67746b64257..04ce65e5725 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 @@ -23,16 +23,11 @@ import org.apache.sysds.runtime.controlprogram.caching.CacheableData; import org.apache.sysds.runtime.controlprogram.parfor.LocalTaskQueue; import org.apache.sysds.runtime.meta.DataCharacteristics; -import org.apache.sysds.runtime.ooc.stream.message.OOCGetStreamTypeMessage; -import org.apache.sysds.runtime.ooc.stream.message.OOCStreamMessage; import org.apache.sysds.runtime.ooc.util.OOCUtils; -import org.apache.sysds.runtime.util.IndexRange; import java.util.LinkedList; -import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; -import java.util.function.BiFunction; import java.util.function.Consumer; public class SubscribableTaskQueue extends LocalTaskQueue> implements OOCStream { @@ -43,9 +38,6 @@ public class SubscribableTaskQueue extends LocalTaskQueue _lastDequeued = null; private CacheableData _cdata; private volatile Consumer> _subscriber = null; - private volatile CopyOnWriteArrayList> _upstreamMsgRelays = null; - private volatile CopyOnWriteArrayList> _downstreamMsgRelays = null; - private volatile BiFunction _ixTransform = null; private String _watchdogId; public SubscribableTaskQueue() { @@ -190,8 +182,6 @@ public synchronized void closeInput() { if(_closed.compareAndSet(false, true)) { super.closeInput(); onDeliveryFinished(); - _upstreamMsgRelays = null; - _downstreamMsgRelays = null; } else { throw new IllegalStateException("Multiple close input calls"); @@ -276,43 +266,6 @@ public OOCStream getWriteStream() { return this; } - @Override - public void messageUpstream(OOCStreamMessage msg) { - if(msg.isCancelled()) - return; - msg.addIXTransform(_ixTransform); - if(msg.isCancelled()) - return; - if(msg instanceof OOCGetStreamTypeMessage) { - if(_cdata != null) - ((OOCGetStreamTypeMessage) msg).setInMemoryType(); - return; - } - CopyOnWriteArrayList> relays = _upstreamMsgRelays; - if(relays != null) { - for(Consumer relay : relays) { - if(msg.isCancelled()) - break; - relay.accept(msg); - } - } - } - - @Override - public void messageDownstream(OOCStreamMessage msg) { - if(!msg.isCancelled()) - return; - msg.addIXTransform(_ixTransform); - CopyOnWriteArrayList> relays = _downstreamMsgRelays; - if(relays != null) { - for(Consumer relay : relays) { - if(msg.isCancelled()) - break; - relay.accept(msg); - } - } - } - @Override public boolean hasStreamCache() { return false; @@ -340,61 +293,6 @@ public void setData(CacheableData data) { _cdata = data; } - @Override - public void setUpstreamMessageRelay(Consumer relay) { - addUpstreamMessageRelay(relay); - } - - @Override - public void setDownstreamMessageRelay(Consumer relay) { - addDownstreamMessageRelay(relay); - } - - @Override - public void addUpstreamMessageRelay(Consumer relay) { - if(relay == null) - throw new IllegalArgumentException("Cannot set upstream relay to null"); - CopyOnWriteArrayList> relays = _upstreamMsgRelays; - if(relays == null) { - synchronized(this) { - if(_upstreamMsgRelays == null) - _upstreamMsgRelays = new CopyOnWriteArrayList<>(); - relays = _upstreamMsgRelays; - } - } - relays.add(0, relay); - } - - @Override - public void addDownstreamMessageRelay(Consumer relay) { - if(relay == null) - throw new IllegalArgumentException("Cannot set downstream relay to null"); - CopyOnWriteArrayList> relays = _downstreamMsgRelays; - if(relays == null) { - synchronized(this) { - if(_downstreamMsgRelays == null) - _downstreamMsgRelays = new CopyOnWriteArrayList<>(); - relays = _downstreamMsgRelays; - } - } - relays.add(0, relay); - } - - @Override - public void clearUpstreamMessageRelays() { - _upstreamMsgRelays = null; - } - - @Override - public void clearDownstreamMessageRelays() { - _downstreamMsgRelays = null; - } - - @Override - public void setIXTransform(BiFunction transform) { - _ixTransform = transform; - } - @Override public synchronized String toString() { return "STQ-" + hashCode(); diff --git a/src/main/java/org/apache/sysds/runtime/instructions/ooc/TernaryOOCInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/ooc/TernaryOOCInstruction.java index 6dfedfc1ff2..6036647cc7f 100644 --- a/src/main/java/org/apache/sysds/runtime/instructions/ooc/TernaryOOCInstruction.java +++ b/src/main/java/org/apache/sysds/runtime/instructions/ooc/TernaryOOCInstruction.java @@ -109,8 +109,6 @@ private void processSingleMatrixInstruction(ExecutionContext ec, int matrixPos) OOCStream qIn = mo.getStreamHandle(); OOCStream qOut = createWritableStream(); ec.getMatrixObject(output).setStreamHandle(qOut); - qIn.setDownstreamMessageRelay(qOut::messageDownstream); - qOut.setUpstreamMessageRelay(qIn::messageUpstream); mapOOC(qIn, qOut, tmp -> { IndexedMatrixValue outVal = new IndexedMatrixValue(); @@ -135,12 +133,6 @@ private void processTwoMatrixInstruction(ExecutionContext ec, int leftPos, int r OOCStream qOut = createWritableStream(); ec.getMatrixObject(output).setStreamHandle(qOut); - qOut.setUpstreamMessageRelay(msg -> { - leftStream.messageUpstream(msg.split()); - rightStream.messageUpstream(msg.split()); - }); - leftStream.setDownstreamMessageRelay(qOut::messageDownstream); - rightStream.setDownstreamMessageRelay(qOut::messageDownstream); joinOOC(leftStream, rightStream, qOut, (l, r) -> { IndexedMatrixValue outVal = new IndexedMatrixValue(); @@ -164,9 +156,6 @@ private void processThreeMatrixInstruction(ExecutionContext ec) { List> streams = List.of( m1.getStreamHandle(), m2.getStreamHandle(), m3.getStreamHandle()); - streams.forEach(s -> s.setDownstreamMessageRelay(qOut::messageDownstream)); - qOut.setUpstreamMessageRelay(msg -> streams.forEach(s -> s.messageUpstream(msg))); - joinOOC(streams, qOut, blocks -> { IndexedMatrixValue b1 = blocks.get(0); IndexedMatrixValue b2 = blocks.get(1); diff --git a/src/main/java/org/apache/sysds/runtime/instructions/ooc/UnaryOOCInstruction.java b/src/main/java/org/apache/sysds/runtime/instructions/ooc/UnaryOOCInstruction.java index a95cf0ec333..df2f50bf573 100644 --- a/src/main/java/org/apache/sysds/runtime/instructions/ooc/UnaryOOCInstruction.java +++ b/src/main/java/org/apache/sysds/runtime/instructions/ooc/UnaryOOCInstruction.java @@ -86,10 +86,6 @@ public void processInstruction( ExecutionContext ec ) { } ec.getMatrixObject(output).setStreamHandle(qOut); - if(!cumulative) { - qIn.setDownstreamMessageRelay(qOut::messageDownstream); - qOut.setUpstreamMessageRelay(qIn::messageUpstream); - } } private OOCStream processCumulativeUnaryInstruction(ExecutionContext ec, UnaryOperator uop, 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 f7f57744390..276f96680a6 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 @@ -24,10 +24,7 @@ import org.apache.sysds.runtime.instructions.ooc.CachingStream; import org.apache.sysds.runtime.instructions.ooc.OOCStream; import org.apache.sysds.runtime.meta.DataCharacteristics; -import org.apache.sysds.runtime.ooc.stream.message.OOCStreamMessage; -import org.apache.sysds.runtime.util.IndexRange; -import java.util.function.BiFunction; import java.util.function.Consumer; import java.util.function.Function; @@ -150,49 +147,4 @@ public CacheableData getData() { public void setData(CacheableData data) { _data = data; } - - @Override - public void messageUpstream(OOCStreamMessage msg) { - _sourceStream.messageUpstream(msg); - } - - @Override - public void messageDownstream(OOCStreamMessage msg) { - _sourceStream.messageDownstream(msg); - } - - @Override - public void setUpstreamMessageRelay(Consumer relay) { - _sourceStream.setUpstreamMessageRelay(relay); - } - - @Override - public void setDownstreamMessageRelay(Consumer relay) { - _sourceStream.setDownstreamMessageRelay(relay); - } - - @Override - public void addUpstreamMessageRelay(Consumer relay) { - _sourceStream.addUpstreamMessageRelay(relay); - } - - @Override - public void addDownstreamMessageRelay(Consumer relay) { - _sourceStream.addDownstreamMessageRelay(relay); - } - - @Override - public void clearUpstreamMessageRelays() { - _sourceStream.clearUpstreamMessageRelays(); - } - - @Override - public void clearDownstreamMessageRelays() { - _sourceStream.clearDownstreamMessageRelays(); - } - - @Override - public void setIXTransform(BiFunction transform) { - _sourceStream.setIXTransform(transform); - } } 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 51e38c8cce9..1c0f4977a27 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 @@ -25,14 +25,11 @@ import org.apache.sysds.runtime.instructions.ooc.OOCStream; import org.apache.sysds.runtime.instructions.ooc.SubscribableTaskQueue; import org.apache.sysds.runtime.meta.DataCharacteristics; -import org.apache.sysds.runtime.ooc.stream.message.OOCStreamMessage; -import org.apache.sysds.runtime.util.IndexRange; import java.util.Arrays; import java.util.List; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; -import java.util.function.BiFunction; import java.util.function.Consumer; public class MergedOOCStream implements OOCStream { @@ -52,11 +49,6 @@ public MergedOOCStream(List> sources) { _failed = new AtomicBoolean(false); _sharedCache = findSharedCache(sources); - _taskQueue.setUpstreamMessageRelay(msg -> { - for(OOCStream source : _sources) - source.messageUpstream(msg); - }); - for(OOCStream source : _sources) { source.setSubscriber(cb -> { try { @@ -229,49 +221,4 @@ public CacheableData getData() { public void setData(CacheableData data) { _taskQueue.setData(data); } - - @Override - public void messageUpstream(OOCStreamMessage msg) { - _taskQueue.messageUpstream(msg); - } - - @Override - public void messageDownstream(OOCStreamMessage msg) { - _taskQueue.messageDownstream(msg); - } - - @Override - public void setUpstreamMessageRelay(Consumer relay) { - _taskQueue.setUpstreamMessageRelay(relay); - } - - @Override - public void setDownstreamMessageRelay(Consumer relay) { - _taskQueue.setDownstreamMessageRelay(relay); - } - - @Override - public void addUpstreamMessageRelay(Consumer relay) { - throw new UnsupportedOperationException(); - } - - @Override - public void addDownstreamMessageRelay(Consumer relay) { - _taskQueue.addDownstreamMessageRelay(relay); - } - - @Override - public void clearUpstreamMessageRelays() { - _taskQueue.clearUpstreamMessageRelays(); - } - - @Override - public void clearDownstreamMessageRelays() { - _taskQueue.clearDownstreamMessageRelays(); - } - - @Override - public void setIXTransform(BiFunction transform) { - _taskQueue.setIXTransform(transform); - } } diff --git a/src/main/java/org/apache/sysds/runtime/ooc/stream/SourceOOCStream.java b/src/main/java/org/apache/sysds/runtime/ooc/stream/SourceOOCStream.java index 0941cf0ea5c..3f1ef94f2a7 100644 --- a/src/main/java/org/apache/sysds/runtime/ooc/stream/SourceOOCStream.java +++ b/src/main/java/org/apache/sysds/runtime/ooc/stream/SourceOOCStream.java @@ -26,7 +26,6 @@ import org.apache.sysds.runtime.instructions.spark.data.IndexedMatrixValue; import org.apache.sysds.runtime.ooc.cache.OOCCacheManager; import org.apache.sysds.runtime.ooc.cache.io.OOCIOHandler; -import org.apache.sysds.runtime.ooc.stream.message.OOCStreamMessage; import org.apache.sysds.runtime.matrix.data.MatrixIndexes; import java.util.concurrent.ConcurrentHashMap; @@ -90,13 +89,6 @@ private void waitForBackpressure() { } } - @Override - public void messageUpstream(OOCStreamMessage msg) { - if(msg.isCancelled()) - return; - super.messageUpstream(msg); - } - public static class SourceGroupCallback implements OOCStream.GroupQueueCallback { private final List _data; private final OOCIOHandler.GroupSourceBlockDescriptor _descriptor; diff --git a/src/main/java/org/apache/sysds/runtime/ooc/stream/SourceOOCStreamable.java b/src/main/java/org/apache/sysds/runtime/ooc/stream/SourceOOCStreamable.java index 4a5f018b2ac..4c9ca3683af 100644 --- a/src/main/java/org/apache/sysds/runtime/ooc/stream/SourceOOCStreamable.java +++ b/src/main/java/org/apache/sysds/runtime/ooc/stream/SourceOOCStreamable.java @@ -25,11 +25,6 @@ import org.apache.sysds.runtime.instructions.ooc.OOCStreamable; import org.apache.sysds.runtime.instructions.spark.data.IndexedMatrixValue; import org.apache.sysds.runtime.meta.DataCharacteristics; -import org.apache.sysds.runtime.ooc.stream.message.OOCStreamMessage; -import org.apache.sysds.runtime.util.IndexRange; - -import java.util.function.BiFunction; -import java.util.function.Consumer; public class SourceOOCStreamable implements OOCStreamable { private final CacheableData _data; @@ -77,49 +72,4 @@ public CacheableData getData() { public void setData(CacheableData data) { throw new UnsupportedOperationException(); } - - @Override - public void messageUpstream(OOCStreamMessage msg) { - - } - - @Override - public void messageDownstream(OOCStreamMessage msg) { - - } - - @Override - public void setUpstreamMessageRelay(Consumer relay) { - - } - - @Override - public void setDownstreamMessageRelay(Consumer relay) { - - } - - @Override - public void addUpstreamMessageRelay(Consumer relay) { - - } - - @Override - public void addDownstreamMessageRelay(Consumer relay) { - - } - - @Override - public void clearUpstreamMessageRelays() { - - } - - @Override - public void clearDownstreamMessageRelays() { - - } - - @Override - public void setIXTransform(BiFunction transform) { - - } } 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 b7fd9a1d1a4..7aef56e96ba 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 @@ -24,10 +24,7 @@ import org.apache.sysds.runtime.instructions.ooc.CachingStream; import org.apache.sysds.runtime.instructions.ooc.OOCStream; import org.apache.sysds.runtime.meta.DataCharacteristics; -import org.apache.sysds.runtime.ooc.stream.message.OOCStreamMessage; -import org.apache.sysds.runtime.util.IndexRange; -import java.util.function.BiFunction; import java.util.function.Consumer; import java.util.function.Function; @@ -181,50 +178,4 @@ public CacheableData getData() { public void setData(CacheableData data) { throw new UnsupportedOperationException(); } - - @Override - public void messageUpstream(OOCStreamMessage msg) { - _sourceStream.messageUpstream(msg); - } - - @Override - public void messageDownstream(OOCStreamMessage msg) { - for(SubOOCStream sub : _subStreams) - sub.messageDownstream(msg); - } - - @Override - public void setUpstreamMessageRelay(Consumer relay) { - throw new UnsupportedOperationException(); - } - - @Override - public void setDownstreamMessageRelay(Consumer relay) { - throw new UnsupportedOperationException(); - } - - @Override - public void addUpstreamMessageRelay(Consumer relay) { - throw new UnsupportedOperationException(); - } - - @Override - public void addDownstreamMessageRelay(Consumer relay) { - throw new UnsupportedOperationException(); - } - - @Override - public void clearUpstreamMessageRelays() { - throw new UnsupportedOperationException(); - } - - @Override - public void clearDownstreamMessageRelays() { - throw new UnsupportedOperationException(); - } - - @Override - public void setIXTransform(BiFunction transform) { - 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 9a231654c26..44f6542a4d3 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 @@ -25,10 +25,7 @@ import org.apache.sysds.runtime.instructions.ooc.OOCStream; import org.apache.sysds.runtime.instructions.ooc.SubscribableTaskQueue; import org.apache.sysds.runtime.meta.DataCharacteristics; -import org.apache.sysds.runtime.ooc.stream.message.OOCStreamMessage; -import org.apache.sysds.runtime.util.IndexRange; -import java.util.function.BiFunction; import java.util.function.Consumer; public class SubOOCStream implements OOCStream { @@ -39,7 +36,6 @@ public class SubOOCStream implements OOCStream { public SubOOCStream(OOCStream sourceStream) { _sourceStream = sourceStream; _taskQueue = new SubscribableTaskQueue<>(); - _taskQueue.setUpstreamMessageRelay(_sourceStream::messageUpstream); } @Override @@ -140,50 +136,4 @@ public CacheableData getData() { public void setData(CacheableData data) { _taskQueue.setData(data); } - - @Override - public void messageUpstream(OOCStreamMessage msg) { - _taskQueue.messageUpstream(msg); - } - - @Override - public void messageDownstream(OOCStreamMessage msg) { - _taskQueue.messageDownstream(msg); - } - - @Override - public void setUpstreamMessageRelay(Consumer relay) { - // Upstream is handled by source stream - throw new UnsupportedOperationException(); - } - - @Override - public void setDownstreamMessageRelay(Consumer relay) { - _taskQueue.setDownstreamMessageRelay(relay); - } - - @Override - public void addUpstreamMessageRelay(Consumer relay) { - throw new UnsupportedOperationException(); - } - - @Override - public void addDownstreamMessageRelay(Consumer relay) { - _taskQueue.addDownstreamMessageRelay(relay); - } - - @Override - public void clearUpstreamMessageRelays() { - _taskQueue.clearUpstreamMessageRelays(); - } - - @Override - public void clearDownstreamMessageRelays() { - _taskQueue.clearDownstreamMessageRelays(); - } - - @Override - public void setIXTransform(BiFunction transform) { - _taskQueue.setIXTransform(transform); - } } diff --git a/src/main/java/org/apache/sysds/runtime/ooc/stream/message/OOCGetStreamTypeMessage.java b/src/main/java/org/apache/sysds/runtime/ooc/stream/message/OOCGetStreamTypeMessage.java deleted file mode 100644 index 0866735d83f..00000000000 --- a/src/main/java/org/apache/sysds/runtime/ooc/stream/message/OOCGetStreamTypeMessage.java +++ /dev/null @@ -1,59 +0,0 @@ -/* - * 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.ooc.stream.message; - -import org.apache.sysds.runtime.util.IndexRange; - -import java.util.function.BiFunction; - -public class OOCGetStreamTypeMessage implements OOCStreamMessage { - public static final byte STREAM_TYPE_UNKNOWN = 0; - public static final byte STREAM_TYPE_CACHED = 1; - public static final byte STREAM_TYPE_IN_MEMORY = 2; - - private byte _streamType; - - public OOCGetStreamTypeMessage() { - _streamType = 0; - } - - @Override - public void addIXTransform(BiFunction transform) {} - - public void setUnknownType() { - _streamType = STREAM_TYPE_UNKNOWN; - } - - public void setCachedType() { - _streamType = STREAM_TYPE_CACHED; - } - - public void setInMemoryType() { - _streamType = STREAM_TYPE_IN_MEMORY; - } - - public byte getStreamType() { - return _streamType; - } - - public boolean isRequestable() { - return _streamType == STREAM_TYPE_CACHED || _streamType == STREAM_TYPE_IN_MEMORY; - } -} diff --git a/src/main/java/org/apache/sysds/runtime/ooc/stream/message/OOCStreamMessage.java b/src/main/java/org/apache/sysds/runtime/ooc/stream/message/OOCStreamMessage.java deleted file mode 100644 index 26459a74ffa..00000000000 --- a/src/main/java/org/apache/sysds/runtime/ooc/stream/message/OOCStreamMessage.java +++ /dev/null @@ -1,38 +0,0 @@ -/* - * 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.ooc.stream.message; - -import org.apache.sysds.runtime.util.IndexRange; - -import java.util.function.BiFunction; - -public interface OOCStreamMessage { - default boolean isCancelled() { - return false; - } - - default void cancel() {} - - default OOCStreamMessage split() { - return this; - } - - void addIXTransform(BiFunction transform); -}