You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@systemds.apache.org by ja...@apache.org on 2026/07/20 09:28:09 UTC
(systemds) branch main updated: [SYSTEMDS-3891] Add OOC Primitives
This is an automated email from the ASF dual-hosted git repository.
janniklinde pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/systemds.git
The following commit(s) were added to refs/heads/main by this push:
new 07cdffeab5 [SYSTEMDS-3891] Add OOC Primitives
07cdffeab5 is described below
commit 07cdffeab5d9634ac890a4e3943d509b49548373
Author: Jannik Lindemann <52...@users.noreply.github.com>
AuthorDate: Mon Jul 20 11:28:02 2026 +0200
[SYSTEMDS-3891] Add OOC Primitives
---
.../sysds/runtime/instructions/ooc/OOCStream.java | 7 ++
.../runtime/instructions/ooc/OOCStreamable.java | 9 +++
.../instructions/ooc/SubscribableTaskQueue.java | 14 ++++
.../runtime/ooc/planning/OOCAccessPattern.java | 53 +++++++++++++
.../sysds/runtime/ooc/primitives/OOCPrimitive.java | 92 ++++++++++++++++++++++
.../runtime/ooc/stream/AllocatedOOCStream.java | 6 ++
.../runtime/ooc/stream/FilteredOOCStream.java | 6 ++
.../sysds/runtime/ooc/stream/MergedOOCStream.java | 6 ++
.../runtime/ooc/stream/SplittingOOCStream.java | 6 ++
.../sysds/runtime/ooc/stream/SubOOCStream.java | 6 ++
.../sysds/test/component/ooc/OOCPrimitiveTest.java | 78 ++++++++++++++++++
11 files changed, 283 insertions(+)
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 b4ffbbbaed..49a41ce236 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
@@ -20,6 +20,7 @@
package org.apache.sysds.runtime.instructions.ooc;
import org.apache.sysds.runtime.DMLRuntimeException;
+import org.apache.sysds.runtime.ooc.primitives.OOCPrimitive;
import java.util.function.Consumer;
@@ -48,6 +49,12 @@ public interface OOCStream<T> extends OOCStreamable<T> {
*/
void setSubscriber(Consumer<QueueCallback<T>> subscriber);
+ default void start() {
+ OOCPrimitive primitive = getPrimitive();
+ if(primitive != null)
+ primitive.tryStartExecution();
+ }
+
interface QueueCallback<T> extends AutoCloseable {
T get();
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 75ccdee3ee..0f087a7f55 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,6 +21,7 @@ package org.apache.sysds.runtime.instructions.ooc;
import org.apache.sysds.runtime.controlprogram.caching.CacheableData;
import org.apache.sysds.runtime.meta.DataCharacteristics;
+import org.apache.sysds.runtime.ooc.primitives.OOCPrimitive;
public interface OOCStreamable<T> {
OOCStream<T> getReadStream();
@@ -38,4 +39,12 @@ public interface OOCStreamable<T> {
CacheableData<?> getData();
void setData(CacheableData<?> data);
+
+ default OOCPrimitive getPrimitive() {
+ return null;
+ }
+
+ default void assignPrimitive(OOCPrimitive primitive) {
+ throw new UnsupportedOperationException("Stream does not support primitive assignment");
+ }
}
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 04ce65e572..5400b6ba98 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,6 +23,7 @@ import org.apache.sysds.runtime.DMLRuntimeException;
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.primitives.OOCPrimitive;
import org.apache.sysds.runtime.ooc.util.OOCUtils;
import java.util.LinkedList;
@@ -37,6 +38,7 @@ public class SubscribableTaskQueue<T> extends LocalTaskQueue<OOCStream.QueueCall
private final AtomicInteger _blockCount = new AtomicInteger(0);
private QueueCallback<T> _lastDequeued = null;
private CacheableData<?> _cdata;
+ private volatile OOCPrimitive _primitive;
private volatile Consumer<QueueCallback<T>> _subscriber = null;
private String _watchdogId;
@@ -276,6 +278,18 @@ public class SubscribableTaskQueue<T> extends LocalTaskQueue<OOCStream.QueueCall
return null;
}
+ @Override
+ public OOCPrimitive getPrimitive() {
+ return _primitive;
+ }
+
+ @Override
+ public void assignPrimitive(OOCPrimitive primitive) {
+ if(_primitive != null)
+ throw new IllegalStateException("Primitive already assigned");
+ _primitive = primitive;
+ }
+
@Override
public DataCharacteristics getDataCharacteristics() {
return _cdata == null ? null : _cdata.getDataCharacteristics();
diff --git a/src/main/java/org/apache/sysds/runtime/ooc/planning/OOCAccessPattern.java b/src/main/java/org/apache/sysds/runtime/ooc/planning/OOCAccessPattern.java
new file mode 100644
index 0000000000..72a67fedfd
--- /dev/null
+++ b/src/main/java/org/apache/sysds/runtime/ooc/planning/OOCAccessPattern.java
@@ -0,0 +1,53 @@
+/*
+ * 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.planning;
+
+public enum OOCAccessPattern {
+ ROW_MAJOR, COL_MAJOR, ANY, UNKNOWN, UNSET;
+
+ public OOCAccessPattern fused(OOCAccessPattern other) {
+ return switch(this) {
+ case ANY -> other;
+ case UNKNOWN, UNSET -> this;
+ case ROW_MAJOR -> other == ROW_MAJOR || other == ANY ? this : UNKNOWN;
+ case COL_MAJOR -> other == COL_MAJOR || other == ANY ? this : UNKNOWN;
+ };
+ }
+
+ public OOCAccessPattern transposed() {
+ return switch(this) {
+ case ROW_MAJOR -> COL_MAJOR;
+ case COL_MAJOR -> ROW_MAJOR;
+ default -> this;
+ };
+ }
+
+ public OOCAccessPattern preferred(OOCAccessPattern preferred) {
+ return this == ANY || this == UNSET ? preferred : this;
+ }
+
+ public boolean isPlannable() {
+ return this == ROW_MAJOR || this == COL_MAJOR || this == ANY;
+ }
+
+ public boolean isUnset() {
+ return this == UNSET;
+ }
+}
diff --git a/src/main/java/org/apache/sysds/runtime/ooc/primitives/OOCPrimitive.java b/src/main/java/org/apache/sysds/runtime/ooc/primitives/OOCPrimitive.java
new file mode 100644
index 0000000000..67c3072de5
--- /dev/null
+++ b/src/main/java/org/apache/sysds/runtime/ooc/primitives/OOCPrimitive.java
@@ -0,0 +1,92 @@
+/*
+ * 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.primitives;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.sysds.runtime.ooc.planning.OOCAccessPattern;
+
+public abstract class OOCPrimitive {
+ private final List<OOCPrimitive> _children;
+ private final List<OOCPrimitive> _parents;
+ private final AtomicBoolean _executionStarted;
+ protected OOCAccessPattern _pattern;
+
+ protected OOCPrimitive(List<OOCPrimitive> children) {
+ _parents = new ArrayList<>();
+ List<OOCPrimitive> uniqueChildren = new ArrayList<>(children.size());
+ for(OOCPrimitive child : children) {
+ if(containsIdentity(uniqueChildren, child))
+ continue;
+ uniqueChildren.add(child);
+ child.addParent(this);
+ }
+ _children = List.copyOf(uniqueChildren);
+ _executionStarted = new AtomicBoolean();
+ _pattern = OOCAccessPattern.UNSET;
+ }
+
+ public final List<OOCPrimitive> getChildren() {
+ return _children;
+ }
+
+ public final List<OOCPrimitive> getParents() {
+ return List.copyOf(_parents);
+ }
+
+ private void addParent(OOCPrimitive parent) {
+ if(!containsIdentity(_parents, parent))
+ _parents.add(parent);
+ }
+
+ protected final void inferParentPatterns() {
+ for(OOCPrimitive parent : _parents)
+ if(parent._pattern.isUnset())
+ parent.inferPatterns();
+ }
+
+ public final OOCAccessPattern getAccessPattern() {
+ return _pattern;
+ }
+
+ public final boolean hasStartedExecution() {
+ return _executionStarted.get();
+ }
+
+ public final void tryStartExecution() {
+ if(_executionStarted.compareAndSet(false, true))
+ startExecution();
+ }
+
+ private static boolean containsIdentity(List<OOCPrimitive> primitives, OOCPrimitive primitive) {
+ for(OOCPrimitive current : primitives)
+ if(current == primitive)
+ return true;
+ return false;
+ }
+
+ protected abstract void startExecution();
+
+ public abstract void inferPatterns();
+
+ public abstract void requestPattern(OOCAccessPattern accessPattern);
+}
diff --git a/src/main/java/org/apache/sysds/runtime/ooc/stream/AllocatedOOCStream.java b/src/main/java/org/apache/sysds/runtime/ooc/stream/AllocatedOOCStream.java
index 6d3a7c5936..c940bdd47a 100644
--- a/src/main/java/org/apache/sysds/runtime/ooc/stream/AllocatedOOCStream.java
+++ b/src/main/java/org/apache/sysds/runtime/ooc/stream/AllocatedOOCStream.java
@@ -27,6 +27,7 @@ import org.apache.sysds.runtime.instructions.ooc.SubscribableTaskQueue;
import org.apache.sysds.runtime.ooc.cache.OOCFuture;
import org.apache.sysds.runtime.ooc.memory.MemoryAllowance;
import org.apache.sysds.runtime.ooc.memory.ReservationBudget;
+import org.apache.sysds.runtime.ooc.primitives.OOCPrimitive;
public final class AllocatedOOCStream<T> extends SubscribableTaskQueue<T> {
private final OOCStream<T> _source;
@@ -49,6 +50,11 @@ public final class AllocatedOOCStream<T> extends SubscribableTaskQueue<T> {
return callback instanceof BudgetedQueueCallback<?> budgeted ? budgeted.detachBudget() : null;
}
+ @Override
+ public OOCPrimitive getPrimitive() {
+ return _source.getPrimitive();
+ }
+
private void admit(OOCStream.QueueCallback<T> callback) {
if(callback.isFailure()) {
try(callback) {
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 276f96680a..96eae6d46e 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,6 +24,7 @@ import org.apache.sysds.runtime.controlprogram.caching.CacheableData;
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.primitives.OOCPrimitive;
import java.util.function.Consumer;
import java.util.function.Function;
@@ -88,6 +89,11 @@ public class FilteredOOCStream<T> implements OOCStream<T> {
return _sourceStream.getStreamCache();
}
+ @Override
+ public OOCPrimitive getPrimitive() {
+ return _sourceStream.getPrimitive();
+ }
+
@Override
public void setSubscriber(Consumer<QueueCallback<T>> subscriber) {
_sourceStream.setSubscriber(cb -> {
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 1c0f4977a2..dda930990b 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
@@ -197,6 +197,12 @@ public class MergedOOCStream<T> implements OOCStream<T> {
return this;
}
+ @Override
+ public void start() {
+ for(OOCStream<T> source : _sources)
+ source.start();
+ }
+
@Override
public OOCStream<T> getWriteStream() {
throw new UnsupportedOperationException();
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 7aef56e96b..989cf82f00 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,6 +24,7 @@ import org.apache.sysds.runtime.controlprogram.caching.CacheableData;
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.primitives.OOCPrimitive;
import java.util.function.Consumer;
import java.util.function.Function;
@@ -144,6 +145,11 @@ public class SplittingOOCStream<T> implements OOCStream<T> {
return _sourceStream.getStreamCache();
}
+ @Override
+ public OOCPrimitive getPrimitive() {
+ return _sourceStream.getPrimitive();
+ }
+
@Override
public void setSubscriber(Consumer<QueueCallback<T>> subscriber) {
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 44f6542a4d..653eb3ba22 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,6 +25,7 @@ import org.apache.sysds.runtime.instructions.ooc.CachingStream;
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.primitives.OOCPrimitive;
import java.util.function.Consumer;
@@ -86,6 +87,11 @@ public class SubOOCStream<T> implements OOCStream<T> {
return _sourceStream.getStreamCache();
}
+ @Override
+ public OOCPrimitive getPrimitive() {
+ return _sourceStream.getPrimitive();
+ }
+
@Override
public void setSubscriber(Consumer<QueueCallback<T>> subscriber) {
_taskQueue.setSubscriber(cb -> {
diff --git a/src/test/java/org/apache/sysds/test/component/ooc/OOCPrimitiveTest.java b/src/test/java/org/apache/sysds/test/component/ooc/OOCPrimitiveTest.java
new file mode 100644
index 0000000000..9b6e435ca6
--- /dev/null
+++ b/src/test/java/org/apache/sysds/test/component/ooc/OOCPrimitiveTest.java
@@ -0,0 +1,78 @@
+/*
+ * 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.test.component.ooc;
+
+import java.util.List;
+
+import org.apache.sysds.runtime.instructions.ooc.SubscribableTaskQueue;
+import org.apache.sysds.runtime.ooc.planning.OOCAccessPattern;
+import org.apache.sysds.runtime.ooc.primitives.OOCPrimitive;
+import org.apache.sysds.runtime.ooc.stream.FilteredOOCStream;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class OOCPrimitiveTest {
+ @Test
+ public void testGraphPatternsAndExecution() {
+ TestPrimitive source = new TestPrimitive(List.of());
+ TestPrimitive sink = new TestPrimitive(List.of(source, source));
+
+ Assert.assertEquals(List.of(source), sink.getChildren());
+ Assert.assertEquals(List.of(sink), source.getParents());
+ source.inferPatterns();
+ Assert.assertEquals(OOCAccessPattern.ANY, source.getAccessPattern());
+ Assert.assertEquals(OOCAccessPattern.ANY, sink.getAccessPattern());
+ Assert.assertEquals(OOCAccessPattern.COL_MAJOR, OOCAccessPattern.ROW_MAJOR.transposed());
+ Assert.assertEquals(OOCAccessPattern.UNKNOWN, OOCAccessPattern.ROW_MAJOR.fused(OOCAccessPattern.COL_MAJOR));
+
+ SubscribableTaskQueue<Integer> stream = new SubscribableTaskQueue<>();
+ stream.assignPrimitive(sink);
+ FilteredOOCStream<Integer> filtered = new FilteredOOCStream<>(stream, ignored -> true);
+ Assert.assertSame(sink, filtered.getPrimitive());
+ stream.start();
+ filtered.start();
+ Assert.assertTrue(sink.hasStartedExecution());
+ Assert.assertEquals(1, sink._executions);
+ }
+
+ private static final class TestPrimitive extends OOCPrimitive {
+ private int _executions;
+
+ private TestPrimitive(List<OOCPrimitive> children) {
+ super(children);
+ }
+
+ @Override
+ protected void startExecution() {
+ _executions++;
+ }
+
+ @Override
+ public void inferPatterns() {
+ _pattern = OOCAccessPattern.ANY;
+ inferParentPatterns();
+ }
+
+ @Override
+ public void requestPattern(OOCAccessPattern accessPattern) {
+ _pattern = accessPattern;
+ }
+ }
+}