Skip to content

[SYSTEMDS-3891] Remove Unused OOC Messaging Features #2552

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 17, 2026
Merged

[SYSTEMDS-3891] Remove Unused OOC Messaging Features #2552

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: 0 additions & 14 deletions src/main/java/org/apache/sysds/runtime/instructions/ooc/AggregateTernaryOOCInstruction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -155,19 +154,6 @@ private void processReduceRow(ExecutionContext ec, AggregateTernaryOperator abOp
if(qIn3 != null)
streams.add(qIn3);

for (OOCStream<IndexedMatrixValue> 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<Void> fut = joinOOC(streams, qMid, blocks -> {
MatrixBlock b1 = (MatrixBlock) blocks.get(0).getValue();
MatrixBlock b2 = (MatrixBlock) blocks.get(1).getValue();
Expand Down
16 changes: 0 additions & 16 deletions src/main/java/org/apache/sysds/runtime/instructions/ooc/AggregateUnaryOOCInstruction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -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() ?
Expand Down
8 changes: 0 additions & 8 deletions src/main/java/org/apache/sysds/runtime/instructions/ooc/BinaryOOCInstruction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,12 +67,6 @@ protected void processMatrixMatrixInstruction(ExecutionContext ec) {
OOCStream<IndexedMatrixValue> qIn2 = m2.getStreamHandle();
OOCStream<IndexedMatrixValue> 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);
Expand Down Expand Up @@ -152,8 +146,6 @@ protected void processScalarMatrixInstruction(ExecutionContext ec) {
OOCStream<IndexedMatrixValue> qIn = min.getStreamHandle();
OOCStream<IndexedMatrixValue> qOut = createWritableStream();
ec.getMatrixObject(output).setStreamHandle(qOut);
qIn.setDownstreamMessageRelay(qOut::messageDownstream);
qOut.setUpstreamMessageRelay(qIn::messageUpstream);

mapOOC(qIn, qOut, tmp -> {
IndexedMatrixValue tmpOut = new IndexedMatrixValue();
Expand Down
78 changes: 0 additions & 78 deletions src/main/java/org/apache/sysds/runtime/instructions/ooc/CachingStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;

/**
Expand Down Expand Up @@ -74,7 +69,6 @@ public class CachingStream implements OOCStreamable<IndexedMatrixValue> {
private int _numBlocks = 0;

private Consumer<OOCStream.QueueCallback<IndexedMatrixValue>>[] _subscribers;
private CopyOnWriteArrayList<Consumer<OOCStreamMessage>> _downstreamRelays;

// state flags
private boolean _cacheInProgress = true; // caching in progress, in the first pass.
Expand All @@ -92,15 +86,13 @@ public CachingStream(OOCStream<IndexedMatrixValue> source) {

public CachingStream(OOCStream<IndexedMatrixValue> source, long streamId) {
_source = source;
_source.setDownstreamMessageRelay(this::messageDownstream);
_streamId = streamId;
if(OOCWatchdog.WATCH) {
_watchdogId = "CS-" + hashCode();
// Capture a short context to help identify origin
OOCWatchdog.registerOpen(_watchdogId, toString(), getCtxMsg(), this);
}
activateIndexing();
_downstreamRelays = null;
source.setSubscriber(tmp -> {
try(tmp) {
int blk;
Expand Down Expand Up @@ -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<Consumer<OOCStreamMessage>> relays = _downstreamRelays;
if (relays != null) {
for (Consumer<OOCStreamMessage> relay : relays) {
if (msg.isCancelled())
break;
relay.accept(msg);
}
}
}

@Override
public void setUpstreamMessageRelay(Consumer<OOCStreamMessage> relay) {
throw new UnsupportedOperationException();
}

@Override
public void setDownstreamMessageRelay(Consumer<OOCStreamMessage> relay) {
addDownstreamMessageRelay(relay);
}

@Override
public void addUpstreamMessageRelay(Consumer<OOCStreamMessage> relay) {
throw new UnsupportedOperationException();
}

@Override
public void addDownstreamMessageRelay(Consumer<OOCStreamMessage> relay) {
if (relay == null)
throw new IllegalArgumentException("Cannot set downstream relay to null");
CopyOnWriteArrayList<Consumer<OOCStreamMessage>> 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<Boolean, IndexRange, IndexRange> transform) {
throw new UnsupportedOperationException();
}

@SuppressWarnings("unchecked")
public void setSubscriber(Consumer<OOCStream.QueueCallback<IndexedMatrixValue>> subscriber, boolean incrConsumers) {
if(_deletable)
Expand Down
24 changes: 0 additions & 24 deletions src/main/java/org/apache/sysds/runtime/instructions/ooc/MatrixIndexingOOCInstruction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<IndexedMatrixValue> filteredStream = new FilteredOOCStream<>(qIn,
Expand Down
23 changes: 0 additions & 23 deletions src/main/java/org/apache/sysds/runtime/instructions/ooc/OOCStreamable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<T> {
OOCStream<T> getReadStream();
Expand All @@ -43,22 +38,4 @@ public interface OOCStreamable<T> {
CacheableData<?> getData();

void setData(CacheableData<?> data);

void messageUpstream(OOCStreamMessage msg);

void messageDownstream(OOCStreamMessage msg);

void setUpstreamMessageRelay(Consumer<OOCStreamMessage> relay);

void setDownstreamMessageRelay(Consumer<OOCStreamMessage> relay);

void addUpstreamMessageRelay(Consumer<OOCStreamMessage> relay);

void addDownstreamMessageRelay(Consumer<OOCStreamMessage> relay);

void clearUpstreamMessageRelays();

void clearDownstreamMessageRelays();

void setIXTransform(BiFunction<Boolean, IndexRange, IndexRange> transform);
}
6 changes: 0 additions & 6 deletions ...in/java/org/apache/sysds/runtime/instructions/ooc/ParameterizedBuiltinOOCInstruction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down Expand Up @@ -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<IndexedMatrixValue> out = new ArrayList<>();
LibMatrixReorg.rexpand(tmp, lmaxVal, dirRows, cast, ignore, blen, out);
Expand Down
73 changes: 0 additions & 73 deletions src/main/java/org/apache/sysds/runtime/instructions/ooc/PlaybackStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,22 +23,17 @@
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<IndexedMatrixValue> {
private final CachingStream _streamCache;
private final AtomicInteger _streamIdx;
private final AtomicBoolean _subscriberSet;
private QueueCallback<IndexedMatrixValue> _lastDequeue;
private volatile CopyOnWriteArrayList<Consumer<OOCStreamMessage>> _downstreamRelays;

public PlaybackStream(CachingStream streamCache) {
this._streamCache = streamCache;
Expand Down Expand Up @@ -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<Consumer<OOCStreamMessage>> relays = _downstreamRelays;
if(relays != null) {
for(Consumer<OOCStreamMessage> relay : relays) {
if(msg.isCancelled())
break;
relay.accept(msg);
}
}
}

@Override
public void setSubscriber(Consumer<QueueCallback<IndexedMatrixValue>> subscriber) {
if(!_subscriberSet.compareAndSet(false, true))
Expand All @@ -166,51 +140,4 @@ public boolean hasStreamCache() {
public CachingStream getStreamCache() {
return _streamCache;
}

@Override
public void setUpstreamMessageRelay(Consumer<OOCStreamMessage> relay) {
throw new UnsupportedOperationException();
}

@Override
public void setDownstreamMessageRelay(Consumer<OOCStreamMessage> relay) {
addDownstreamMessageRelay(relay);
}

@Override
public void addUpstreamMessageRelay(Consumer<OOCStreamMessage> relay) {
throw new UnsupportedOperationException();
}

@Override
public void addDownstreamMessageRelay(Consumer<OOCStreamMessage> relay) {
if(relay == null)
throw new IllegalArgumentException("Cannot set downstream relay to null");
CopyOnWriteArrayList<Consumer<OOCStreamMessage>> 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<Boolean, IndexRange, IndexRange> transform) {
throw new UnsupportedOperationException();
}
}
Loading
Loading