You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tinkerpop.apache.org by ok...@apache.org on 2019/04/18 17:12:33 UTC
[tinkerpop] branch tp4 updated: I have a working implementation of
recurssive-based repeat() in RxJava. This was an intense slog. Learned
alot. I have written some pretty nasty nested branching test cases and the
model is holding up.
This is an automated email from the ASF dual-hosted git repository.
okram pushed a commit to branch tp4
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
The following commit(s) were added to refs/heads/tp4 by this push:
new 6b39526 I have a working implementation of recurssive-based repeat() in RxJava. This was an intense slog. Learned alot. I have written some pretty nasty nested branching test cases and the model is holding up.
6b39526 is described below
commit 6b39526f4bd0109d6789af5c1d300bc18463c6d3
Author: Marko A. Rodriguez <ok...@gmail.com>
AuthorDate: Thu Apr 18 11:12:23 2019 -0600
I have a working implementation of recurssive-based repeat() in RxJava. This was an intense slog. Learned alot. I have written some pretty nasty nested branching test cases and the model is holding up.
---
.../apache/tinkerpop/machine/SimpleTestSuite.java | 14 +++
.../machine/processor/rxjava/RepeatHead.java | 135 +++++++++++++++++++++
.../machine/processor/rxjava/RepeatTail.java | 119 ++++++++++++++++++
.../machine/processor/rxjava/SerialRxJava.java | 41 ++++---
4 files changed, 290 insertions(+), 19 deletions(-)
diff --git a/java/machine/machine-test/src/main/java/org/apache/tinkerpop/machine/SimpleTestSuite.java b/java/machine/machine-test/src/main/java/org/apache/tinkerpop/machine/SimpleTestSuite.java
index 862ff71..7b1b6e8 100644
--- a/java/machine/machine-test/src/main/java/org/apache/tinkerpop/machine/SimpleTestSuite.java
+++ b/java/machine/machine-test/src/main/java/org/apache/tinkerpop/machine/SimpleTestSuite.java
@@ -24,6 +24,8 @@ import org.apache.tinkerpop.machine.bytecode.compiler.Order;
import org.junit.jupiter.api.Test;
import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
import static org.apache.tinkerpop.language.gremlin.P.gt;
import static org.apache.tinkerpop.language.gremlin.P.lt;
@@ -225,6 +227,18 @@ public class SimpleTestSuite extends AbstractTestSuite<Long> {
g.inject(1L).repeat(union(incr(), __.<Long>incr().incr())).times(2));
}
+ // @Test (Parallel RxJava finds this too compute intensive)
+ void g_injectX1X_repeatXunionXincr__incrX_unionXincr__incrXX_timesX3X() {
+ verify(Stream.generate(() -> 7L).limit(64).collect(Collectors.toList()),
+ g.inject(1L).repeat(__.<Long,Long,Long>union(incr(), incr()).union(incr(),incr())).times(3));
+ }
+
+ // @Test (Parallel RxJava finds this too compute intensive)
+ void g_injectX1X_repeatXunionXincr__unionXincr__incrXX_unionXincr__incrXX_timesX3X() {
+ verify(Stream.generate(() -> 7L).limit(216).collect(Collectors.toList()),
+ g.inject(1L).repeat(__.<Long,Long,Long>union(incr(), __.union(incr(),incr())).union(incr(),incr())).times(3));
+ }
+
@Test
void g_injectX1X_repeatXunionXincr__incr_incrXX_timesX3X() {
verify(List.of(4L, 5L, 5L, 6L, 5L, 6L, 6L, 7L),
diff --git a/java/machine/processor/rxjava/src/main/java/org/apache/tinkerpop/machine/processor/rxjava/RepeatHead.java b/java/machine/processor/rxjava/src/main/java/org/apache/tinkerpop/machine/processor/rxjava/RepeatHead.java
new file mode 100644
index 0000000..ef94b8b
--- /dev/null
+++ b/java/machine/processor/rxjava/src/main/java/org/apache/tinkerpop/machine/processor/rxjava/RepeatHead.java
@@ -0,0 +1,135 @@
+/*
+ * 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.tinkerpop.machine.processor.rxjava;
+
+import io.reactivex.Flowable;
+import io.reactivex.FlowableTransformer;
+import org.apache.tinkerpop.machine.function.branch.RepeatBranch;
+import org.apache.tinkerpop.machine.traverser.Traverser;
+import org.apache.tinkerpop.machine.traverser.TraverserSet;
+import org.reactivestreams.Processor;
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+final class RepeatHead<C, S> implements Processor<Traverser<C, S>, Traverser<C, S>>, FlowableTransformer<Traverser<C, S>, Traverser<C, S>> {
+
+ private final TraverserSet<C, S> starts = new TraverserSet<>();
+ private final RepeatBranch<C, S> repeatBranch;
+ private Publisher<Traverser<C, S>> source;
+ private Subscription upstream;
+ private Subscriber<? super Traverser<C, S>> downstream;
+ private RepeatTail<C, S> repeatTail;
+
+ RepeatHead(final RepeatBranch<C, S> repeatBranch) {
+ this.repeatBranch = repeatBranch;
+ }
+
+ void setRepeatTail(final RepeatTail<C, S> repeatTail) {
+ this.repeatTail = repeatTail;
+ }
+
+
+ @Override
+ public void subscribe(final Subscriber<? super Traverser<C, S>> subscriber) {
+ this.downstream = subscriber;
+ this.source.subscribe(this);
+ Subscription subscription = new Subscription() {
+ @Override
+ public void request(long l) {
+ upstream.request(l);
+ drain();
+ }
+
+ @Override
+ public void cancel() {
+ upstream.cancel();
+ }
+ };
+ this.downstream.onSubscribe(subscription);
+ }
+
+ void tailComplete() {
+ drain();
+ downstream.onComplete();
+ }
+
+ @Override
+ public void onSubscribe(final Subscription subscription) {
+ this.upstream = subscription;
+
+ }
+
+ void addTraverser(final Traverser<C, S> traverser) {
+ this.starts.add(traverser);
+ }
+
+ @Override
+ public void onNext(final Traverser<C, S> traverser) {
+ this.processTraverser(traverser);
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+ this.downstream.onError(throwable);
+ }
+
+ @Override
+ public void onComplete() {
+ this.upstream.cancel();
+ }
+
+ private void drain() {
+ while (!this.starts.isEmpty()) {
+ final Traverser<C, S> traverser = this.starts.remove();
+ this.processTraverser(traverser);
+ }
+ }
+
+ private void processTraverser(final Traverser<C, S> traverser) {
+ if (this.repeatBranch.hasStartPredicates()) {
+ if (1 == this.repeatBranch.getUntilLocation()) {
+ if (this.repeatBranch.getUntil().filterTraverser(traverser)) {
+ this.repeatTail.onNextSkip(traverser.repeatDone(this.repeatBranch));
+ } else if (2 == this.repeatBranch.getEmitLocation() && this.repeatBranch.getEmit().filterTraverser(traverser)) {
+ this.downstream.onNext(traverser);
+ this.repeatTail.onNextSkip(traverser.repeatDone(this.repeatBranch));
+ } else
+ this.downstream.onNext(traverser);
+ } else if (1 == this.repeatBranch.getEmitLocation()) {
+ if (this.repeatBranch.getEmit().filterTraverser(traverser))
+ this.repeatTail.onNextSkip(traverser.repeatDone(this.repeatBranch));
+ if (2 == this.repeatBranch.getUntilLocation() && this.repeatBranch.getUntil().filterTraverser(traverser))
+ this.repeatTail.onNextSkip(traverser.repeatDone(this.repeatBranch));
+ else
+ this.downstream.onNext(traverser);
+ }
+ } else
+ this.downstream.onNext(traverser);
+ }
+
+ @Override
+ public Publisher<Traverser<C, S>> apply(final Flowable<Traverser<C, S>> flowable) {
+ this.source = flowable;
+ return this;
+ }
+}
diff --git a/java/machine/processor/rxjava/src/main/java/org/apache/tinkerpop/machine/processor/rxjava/RepeatTail.java b/java/machine/processor/rxjava/src/main/java/org/apache/tinkerpop/machine/processor/rxjava/RepeatTail.java
new file mode 100644
index 0000000..a6ea696
--- /dev/null
+++ b/java/machine/processor/rxjava/src/main/java/org/apache/tinkerpop/machine/processor/rxjava/RepeatTail.java
@@ -0,0 +1,119 @@
+/*
+ * 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.tinkerpop.machine.processor.rxjava;
+
+import io.reactivex.Flowable;
+import io.reactivex.FlowableTransformer;
+import org.apache.tinkerpop.machine.function.branch.RepeatBranch;
+import org.apache.tinkerpop.machine.traverser.Traverser;
+import org.reactivestreams.Processor;
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+final class RepeatTail<C, S> implements Processor<Traverser<C, S>, Traverser<C, S>>, FlowableTransformer<Traverser<C, S>, Traverser<C, S>> {
+
+ private Publisher<Traverser<C, S>> source;
+ private final RepeatBranch<C, S> repeatBranch;
+ private Subscription upstream;
+ private RepeatHead<C, S> repeatHead;
+ private Subscriber<? super Traverser<C, S>> downstream;
+
+ RepeatTail(final RepeatBranch<C, S> repeatBranch) {
+ this.repeatBranch = repeatBranch;
+ }
+
+ void setRepeatHead(final RepeatHead<C, S> repeatHead) {
+ this.repeatHead = repeatHead;
+ }
+
+ @Override
+ public Publisher<Traverser<C, S>> apply(final Flowable<Traverser<C, S>> flowable) {
+ this.source = flowable;
+ return this;
+ }
+
+
+ @Override
+ public void subscribe(final Subscriber<? super Traverser<C, S>> subscriber) {
+ this.downstream = subscriber;
+ this.source.subscribe(this);
+ this.downstream.onSubscribe(new Subscription() {
+ @Override
+ public void request(long l) {
+ upstream.request(l);
+ repeatHead.tailComplete();
+ }
+
+ @Override
+ public void cancel() {
+ upstream.cancel();
+ }
+ });
+ }
+
+ @Override
+ public void onSubscribe(final Subscription subscription) {
+ this.upstream = subscription;
+
+ }
+
+ void onNextSkip(final Traverser<C,S> traverser) {
+ this.downstream.onNext(traverser);
+ }
+
+ @Override
+ public void onNext(final Traverser<C, S> traverser) {
+ final Traverser<C, S> t = traverser.repeatLoop(this.repeatBranch);
+ if (this.repeatBranch.hasEndPredicates()) {
+ if (3 == this.repeatBranch.getUntilLocation()) {
+ if (this.repeatBranch.getUntil().filterTraverser(t)) {
+ this.downstream.onNext(t.repeatDone(this.repeatBranch));
+ } else if (4 == this.repeatBranch.getEmitLocation() && this.repeatBranch.getEmit().filterTraverser(t)) {
+ this.repeatHead.addTraverser(t);
+ this.downstream.onNext(t.repeatDone(this.repeatBranch));
+ } else
+ this.repeatHead.addTraverser(t);
+ } else if (3 == this.repeatBranch.getEmitLocation()) {
+ if (this.repeatBranch.getEmit().filterTraverser(t))
+ this.downstream.onNext(t.repeatDone(this.repeatBranch));
+ if (4 == this.repeatBranch.getUntilLocation() && this.repeatBranch.getUntil().filterTraverser(t))
+ this.downstream.onNext(t.repeatDone(this.repeatBranch));
+ else
+ this.repeatHead.addTraverser(t);
+ }
+ } else {
+ this.repeatHead.addTraverser(t);
+ }
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+ this.downstream.onError(throwable);
+ }
+
+ @Override
+ public void onComplete() {
+ this.downstream.onComplete();
+ }
+
+}
diff --git a/java/machine/processor/rxjava/src/main/java/org/apache/tinkerpop/machine/processor/rxjava/SerialRxJava.java b/java/machine/processor/rxjava/src/main/java/org/apache/tinkerpop/machine/processor/rxjava/SerialRxJava.java
index 230467e..ee04fb3 100644
--- a/java/machine/processor/rxjava/src/main/java/org/apache/tinkerpop/machine/processor/rxjava/SerialRxJava.java
+++ b/java/machine/processor/rxjava/src/main/java/org/apache/tinkerpop/machine/processor/rxjava/SerialRxJava.java
@@ -91,7 +91,7 @@ public final class SerialRxJava<C, S, E> extends AbstractRxJava<C, S, E> {
final BarrierFunction<C, S, E, B> barrierFunction = (BarrierFunction<C, S, E, B>) function;
return flow.reduce(barrierFunction.getInitialValue(), new Barrier<>(barrierFunction)).toFlowable().flatMapIterable(new BarrierFlow<>(barrierFunction, traverserFactory));
} else if (function instanceof BranchFunction) {
- final Flowable<List> selectorFlow = flow.map(new BranchFlow<>((BranchFunction<C, S, B>) function)).share();
+ final Flowable<List> selectorFlow = flow.map(new BranchFlow<>((BranchFunction<C, S, B>) function));
final List<Publisher<Traverser<C, E>>> branchFlows = new ArrayList<>();
int branchCounter = 0;
for (final Map.Entry<Compilation<C, S, ?>, List<Compilation<C, S, E>>> branches : ((BranchFunction<C, S, E>) function).getBranches().entrySet()) {
@@ -107,25 +107,28 @@ public final class SerialRxJava<C, S, E> extends AbstractRxJava<C, S, E> {
return PublishProcessor.merge(branchFlows);
} else if (function instanceof RepeatBranch) {
final RepeatBranch<C, S> repeatBranch = (RepeatBranch<C, S>) function;
- final List<Publisher<Traverser<C, S>>> outputs = new ArrayList<>();
- Flowable<List> selectorFlow;
- for (int i = 0; i < MAX_REPETITIONS; i++) {
- if (repeatBranch.hasStartPredicates()) {
- selectorFlow = flow.flatMapIterable(new RepeatStart<>(repeatBranch)).share();
- outputs.add(selectorFlow.filter(list -> list.get(0).equals(0)).map(list -> (Traverser<C, S>) list.get(1)));
- flow = compile(selectorFlow.filter(list -> list.get(0).equals(1)).map(list -> (Traverser<C, S>) list.get(1)), repeatBranch.getRepeat());
- } else
- flow = compile(flow, repeatBranch.getRepeat());
- ///
- if (repeatBranch.hasEndPredicates()) {
- selectorFlow = flow.flatMapIterable(new RepeatEnd<>(repeatBranch)).share();
- outputs.add(selectorFlow.filter(list -> list.get(0).equals(0)).map(list -> (Traverser<C, S>) list.get(1)));
- flow = selectorFlow.filter(list -> list.get(0).equals(1)).map(list -> (Traverser<C, S>) list.get(1));
- } else
- flow = flow.map(t -> t.repeatLoop(repeatBranch));
- }
- return (Flowable) PublishProcessor.merge(outputs);
+ final RepeatHead<C, S> repeatHead = new RepeatHead<>(repeatBranch);
+ final RepeatTail<C, S> repeatTail = new RepeatTail<>(repeatBranch);
+ repeatHead.setRepeatTail(repeatTail);
+ repeatTail.setRepeatHead(repeatHead);
+ int branches = countBranches(repeatBranch.getRepeat());
+ return (Flowable) compile(flow.compose(repeatHead).publish().refCount(0 == branches ? 1 : branches), repeatBranch.getRepeat()).compose(repeatTail);
}
throw new RuntimeException("Need a new execution plan step: " + function);
}
+
+ private static <C, S, E> int countBranches(final Compilation<C, S, E> compilation) {
+ int counter = 0;
+ for (final CFunction<C> function : compilation.getFunctions()) {
+ if (function instanceof BranchFunction) {
+ for (final List<Compilation<C, S, E>> branches : ((BranchFunction<C, S, E>) function).getBranches().values()) {
+ counter = counter + branches.size();
+ for (final Compilation<C, S, E> branch : branches) {
+ counter = counter + countBranches(branch);
+ }
+ }
+ }
+ }
+ return counter;
+ }
}