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 b4ffbbbaedb..49a41ce2365 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 @@ static QueueCallback eos(DMLRuntimeException e) { */ void setSubscriber(Consumer> subscriber); + default void start() { + OOCPrimitive primitive = getPrimitive(); + if(primitive != null) + primitive.tryStartExecution(); + } + interface QueueCallback 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 75ccdee3ee1..0f087a7f55b 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 @@ 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 { OOCStream getReadStream(); @@ -38,4 +39,12 @@ public interface OOCStreamable { 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 04ce65e5725..5400b6ba98f 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.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 extends LocalTaskQueue _lastDequeued = null; private CacheableData _cdata; + private volatile OOCPrimitive _primitive; private volatile Consumer> _subscriber = null; private String _watchdogId; @@ -276,6 +278,18 @@ public CachingStream getStreamCache() { 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 00000000000..72a67fedfdf --- /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 00000000000..67c3072de50 --- /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 _children; + private final List _parents; + private final AtomicBoolean _executionStarted; + protected OOCAccessPattern _pattern; + + protected OOCPrimitive(List children) { + _parents = new ArrayList<>(); + List 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 getChildren() { + return _children; + } + + public final List 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 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 6d3a7c5936f..c940bdd47a6 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.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 extends SubscribableTaskQueue { private final OOCStream _source; @@ -49,6 +50,11 @@ public static ReservationBudget detachBudget(OOCStream.QueueCallback callback return callback instanceof BudgetedQueueCallback budgeted ? budgeted.detachBudget() : null; } + @Override + public OOCPrimitive getPrimitive() { + return _source.getPrimitive(); + } + private void admit(OOCStream.QueueCallback 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 276f96680a6..96eae6d46e0 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.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 CachingStream getStreamCache() { return _sourceStream.getStreamCache(); } + @Override + public OOCPrimitive getPrimitive() { + return _sourceStream.getPrimitive(); + } + @Override public void setSubscriber(Consumer> 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 1c0f4977a27..dda930990be 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 OOCStream getReadStream() { return this; } + @Override + public void start() { + for(OOCStream source : _sources) + source.start(); + } + @Override public OOCStream 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 7aef56e96ba..989cf82f009 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.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 CachingStream getStreamCache() { return _sourceStream.getStreamCache(); } + @Override + public OOCPrimitive getPrimitive() { + return _sourceStream.getPrimitive(); + } + @Override public void setSubscriber(Consumer> 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 44f6542a4d3..653eb3ba223 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.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 CachingStream getStreamCache() { return _sourceStream.getStreamCache(); } + @Override + public OOCPrimitive getPrimitive() { + return _sourceStream.getPrimitive(); + } + @Override public void setSubscriber(Consumer> 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 00000000000..9b6e435ca60 --- /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 stream = new SubscribableTaskQueue<>(); + stream.assignPrimitive(sink); + FilteredOOCStream 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 children) { + super(children); + } + + @Override + protected void startExecution() { + _executions++; + } + + @Override + public void inferPatterns() { + _pattern = OOCAccessPattern.ANY; + inferParentPatterns(); + } + + @Override + public void requestPattern(OOCAccessPattern accessPattern) { + _pattern = accessPattern; + } + } +}