OMG. That was a 3+ hour bug. The bug was not in Actors, but in GroupStep cloning.... So painful. Renamed Actors to GraphActors so its like GraphComputer. Renamed ActorStep to TraversalActorProgramStep so its like TraversalVertexProgramStep. Next up, get rid of GraphComputing and add PushBased as that is the more general categorization. In push-based execution engines, certain steps need to do certain things. What a day -- so much was accomplished and at the same time, what a hellish code-hole nightmare.
Project: http://git-wip-us.apache.org/repos/asf/tinkerpop/repo Commit: http://git-wip-us.apache.org/repos/asf/tinkerpop/commit/22d4bb6c Tree: http://git-wip-us.apache.org/repos/asf/tinkerpop/tree/22d4bb6c Diff: http://git-wip-us.apache.org/repos/asf/tinkerpop/diff/22d4bb6c Branch: refs/heads/TINKERPOP-1564 Commit: 22d4bb6c067311944f8388ad1344929925491aab Parents: aa0ae03 Author: Marko A. Rodriguez <okramma...@gmail.com> Authored: Tue Dec 13 17:59:45 2016 -0700 Committer: Marko A. Rodriguez <okramma...@gmail.com> Committed: Thu Jan 19 13:01:41 2017 -0700 ---------------------------------------------------------------------- .../akka/process/actor/ActorMailbox.java | 140 +++++++++++++++++++ .../gremlin/akka/process/actor/AkkaActors.java | 69 --------- .../akka/process/actor/AkkaGraphActors.java | 69 +++++++++ .../gremlin/akka/process/actor/MasterActor.java | 2 +- .../akka/process/actor/TraverserMailbox.java | 140 ------------------- .../gremlin/akka/process/actor/WorkerActor.java | 2 +- .../src/main/resources/application.conf | 6 +- .../akka/process/AkkaActorsProvider.java | 4 +- .../gremlin/akka/process/AkkaPlayTest.java | 8 +- .../tinkerpop/gremlin/process/actor/Actors.java | 32 ----- .../gremlin/process/actor/GraphActors.java | 32 +++++ .../actor/traversal/TraversalActorProgram.java | 36 +++-- .../actor/traversal/step/map/ActorStep.java | 79 ----------- .../step/map/TraversalActorProgramStep.java | 79 +++++++++++ .../decoration/ActorProgramStrategy.java | 10 +- 15 files changed, 362 insertions(+), 346 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/22d4bb6c/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actor/ActorMailbox.java ---------------------------------------------------------------------- diff --git a/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actor/ActorMailbox.java b/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actor/ActorMailbox.java new file mode 100644 index 0000000..28afb22 --- /dev/null +++ b/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actor/ActorMailbox.java @@ -0,0 +1,140 @@ +/* + * 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.gremlin.akka.process.actor; + +import akka.actor.ActorRef; +import akka.actor.ActorSystem; +import akka.dispatch.Envelope; +import akka.dispatch.MailboxType; +import akka.dispatch.MessageQueue; +import akka.dispatch.ProducesMessageQueue; +import com.typesafe.config.Config; +import org.apache.tinkerpop.gremlin.process.traversal.Traverser; +import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.TraverserSet; +import scala.Option; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.LinkedList; +import java.util.List; +import java.util.Map; +import java.util.Queue; + +/** + * @author Marko A. Rodriguez (http://markorodriguez.com) + */ +public final class ActorMailbox implements MailboxType, ProducesMessageQueue<ActorMailbox.ActorMessageQueue> { + + private final List<Class> messagePriorities = new ArrayList<>(); + + public static class ActorMessageQueue implements MessageQueue, ActorSemantics { + private final List<Queue> messages; + private final Map<Class, Queue> priorities; + private final Object MUTEX = new Object(); + + public ActorMessageQueue(final List<Class> messagePriorities) { + this.messages = new ArrayList<>(messagePriorities.size()); + this.priorities = new HashMap<>(messagePriorities.size()); + for (final Class clazz : messagePriorities) { + final Queue queue; + if (Traverser.class.isAssignableFrom(clazz)) + queue = new TraverserSet<>(); + else + queue = new LinkedList<>(); + this.messages.add(queue); + this.priorities.put(clazz, queue); + } + } + + public void enqueue(final ActorRef receiver, final Envelope handle) { + synchronized (MUTEX) { + final Queue queue = this.priorities.get(handle.message() instanceof Traverser ? Traverser.class : handle.message().getClass()); + if (null == queue) + throw new IllegalArgumentException("The provided message type is not registered: " + handle.message().getClass()); + else + queue.offer(handle.message() instanceof Traverser ? handle.message() : handle); + } + } + + public Envelope dequeue() { + synchronized (MUTEX) { + for (final Queue queue : this.messages) { + if (!queue.isEmpty()) { + final Object m = queue.poll(); + return m instanceof Traverser ? new Envelope(m, ActorRef.noSender()) : (Envelope) m; + } + } + return null; + } + } + + public int numberOfMessages() { + synchronized (MUTEX) { + int counter = 0; + for (final Queue queue : this.messages) { + counter = counter + queue.size(); + } + return counter; + } + } + + public boolean hasMessages() { + synchronized (MUTEX) { + for (final Queue queue : this.messages) { + if (!queue.isEmpty()) + return true; + } + return false; + } + } + + public void cleanUp(final ActorRef owner, final MessageQueue deadLetters) { + synchronized (MUTEX) { + for (final Queue queue : this.messages) { + while (!queue.isEmpty()) { + final Object m = queue.poll(); + deadLetters.enqueue(owner, m instanceof Traverser ? new Envelope(m, ActorRef.noSender()) : (Envelope) m); + } + } + } + } + } + + // This constructor signature must exist, it will be called by Akka + public ActorMailbox(final ActorSystem.Settings settings, final Config config) { + try { + final String[] messages = ((String) settings.config().getAnyRef("message-priorities")).replace("[", "").replace("]", "").split(","); + for (final String clazz : messages) { + this.messagePriorities.add(Class.forName(clazz.trim())); + } + } catch (final ClassNotFoundException e) { + throw new IllegalArgumentException(e.getMessage(), e); + } + } + + // The create method is called to create the MessageQueue + public MessageQueue create(final Option<ActorRef> owner, final Option<ActorSystem> system) { + return new ActorMessageQueue(this.messagePriorities); + } + + public static interface ActorSemantics { + + } +} http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/22d4bb6c/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actor/AkkaActors.java ---------------------------------------------------------------------- diff --git a/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actor/AkkaActors.java b/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actor/AkkaActors.java deleted file mode 100644 index 5faa8d6..0000000 --- a/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actor/AkkaActors.java +++ /dev/null @@ -1,69 +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.tinkerpop.gremlin.akka.process.actor; - -import akka.actor.ActorSystem; -import akka.actor.Props; -import com.typesafe.config.Config; -import com.typesafe.config.ConfigFactory; -import com.typesafe.config.ConfigValueFactory; -import org.apache.tinkerpop.gremlin.process.actor.ActorProgram; -import org.apache.tinkerpop.gremlin.process.actor.Actors; -import org.apache.tinkerpop.gremlin.process.actor.Address; -import org.apache.tinkerpop.gremlin.structure.Partitioner; - -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.Future; -import java.util.stream.Collectors; - -/** - * @author Marko A. Rodriguez (http://markorodriguez.com) - */ -public final class AkkaActors<R> implements Actors<R> { - - private final ActorProgram<R> actorProgram; - private final ActorSystem system; - private final Address.Master master; - - public AkkaActors(final ActorProgram<R> actorProgram, final Partitioner partitioner) { - this.actorProgram = actorProgram; - final Config config = ConfigFactory.defaultApplication(). - withValue("message-priorities", - ConfigValueFactory.fromAnyRef(this.actorProgram.getMessagePriorities().stream().map(Class::getCanonicalName).collect(Collectors.toList()).toString())); - this.system = ActorSystem.create("traversal-" + actorProgram.hashCode(), config); - this.master = new Address.Master(this.system.actorOf(Props.create(MasterActor.class, this.actorProgram, partitioner), "master").path().toString()); - } - - @Override - public Address.Master master() { - return this.master; - } - - @Override - public Future<R> submit() { - return CompletableFuture.supplyAsync(() -> { - while (!this.system.isTerminated()) { - - } - return this.actorProgram.getResult(); - }); - } -} - http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/22d4bb6c/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actor/AkkaGraphActors.java ---------------------------------------------------------------------- diff --git a/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actor/AkkaGraphActors.java b/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actor/AkkaGraphActors.java new file mode 100644 index 0000000..b28f536 --- /dev/null +++ b/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actor/AkkaGraphActors.java @@ -0,0 +1,69 @@ +/* + * 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.gremlin.akka.process.actor; + +import akka.actor.ActorSystem; +import akka.actor.Props; +import com.typesafe.config.Config; +import com.typesafe.config.ConfigFactory; +import com.typesafe.config.ConfigValueFactory; +import org.apache.tinkerpop.gremlin.process.actor.ActorProgram; +import org.apache.tinkerpop.gremlin.process.actor.GraphActors; +import org.apache.tinkerpop.gremlin.process.actor.Address; +import org.apache.tinkerpop.gremlin.structure.Partitioner; + +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Future; +import java.util.stream.Collectors; + +/** + * @author Marko A. Rodriguez (http://markorodriguez.com) + */ +public final class AkkaGraphActors<R> implements GraphActors<R> { + + private final ActorProgram<R> actorProgram; + private final ActorSystem system; + private final Address.Master master; + + public AkkaGraphActors(final ActorProgram<R> actorProgram, final Partitioner partitioner) { + this.actorProgram = actorProgram; + final Config config = ConfigFactory.defaultApplication(). + withValue("message-priorities", + ConfigValueFactory.fromAnyRef(this.actorProgram.getMessagePriorities().stream().map(Class::getCanonicalName).collect(Collectors.toList()).toString())); + this.system = ActorSystem.create("traversal-" + actorProgram.hashCode(), config); + this.master = new Address.Master(this.system.actorOf(Props.create(MasterActor.class, this.actorProgram, partitioner), "master").path().toString()); + } + + @Override + public Address.Master master() { + return this.master; + } + + @Override + public Future<R> submit() { + return CompletableFuture.supplyAsync(() -> { + while (!this.system.isTerminated()) { + + } + return this.actorProgram.getResult(); + }); + } +} + http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/22d4bb6c/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actor/MasterActor.java ---------------------------------------------------------------------- diff --git a/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actor/MasterActor.java b/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actor/MasterActor.java index 8d61492..bb7f28b 100644 --- a/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actor/MasterActor.java +++ b/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actor/MasterActor.java @@ -38,7 +38,7 @@ import java.util.Map; /** * @author Marko A. Rodriguez (http://markorodriguez.com) */ -public final class MasterActor extends AbstractActor implements RequiresMessageQueue<TraverserMailbox.TraverserSetSemantics>, Actor.Master { +public final class MasterActor extends AbstractActor implements RequiresMessageQueue<ActorMailbox.ActorSemantics>, Actor.Master { private final Address.Master master; private final List<Address.Worker> workers; http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/22d4bb6c/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actor/TraverserMailbox.java ---------------------------------------------------------------------- diff --git a/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actor/TraverserMailbox.java b/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actor/TraverserMailbox.java deleted file mode 100644 index cbf6257..0000000 --- a/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actor/TraverserMailbox.java +++ /dev/null @@ -1,140 +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.tinkerpop.gremlin.akka.process.actor; - -import akka.actor.ActorRef; -import akka.actor.ActorSystem; -import akka.dispatch.Envelope; -import akka.dispatch.MailboxType; -import akka.dispatch.MessageQueue; -import akka.dispatch.ProducesMessageQueue; -import com.typesafe.config.Config; -import org.apache.tinkerpop.gremlin.process.traversal.Traverser; -import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.TraverserSet; -import scala.Option; - -import java.util.ArrayList; -import java.util.HashMap; -import java.util.LinkedList; -import java.util.List; -import java.util.Map; -import java.util.Queue; - -/** - * @author Marko A. Rodriguez (http://markorodriguez.com) - */ -public final class TraverserMailbox implements MailboxType, ProducesMessageQueue<TraverserMailbox.TraverserMessageQueue> { - - private final List<Class> messagePriorities = new ArrayList<>(); - - public static class TraverserMessageQueue implements MessageQueue, TraverserSetSemantics { - private final List<Queue> messages; - private final Map<Class, Queue> priorities; - private final Object MUTEX = new Object(); - - public TraverserMessageQueue(final List<Class> messagePriorities) { - this.messages = new ArrayList<>(messagePriorities.size()); - this.priorities = new HashMap<>(messagePriorities.size()); - for (final Class clazz : messagePriorities) { - final Queue queue; - if (Traverser.class.isAssignableFrom(clazz)) - queue = new TraverserSet<>(); - else - queue = new LinkedList<>(); - this.messages.add(queue); - this.priorities.put(clazz, queue); - } - } - - public void enqueue(final ActorRef receiver, final Envelope handle) { - synchronized (MUTEX) { - final Queue queue = this.priorities.get(handle.message() instanceof Traverser ? Traverser.class : handle.message().getClass()); - if (null == queue) - throw new IllegalArgumentException("The provided message type is not registered: " + handle.message().getClass()); - else - queue.offer(handle.message() instanceof Traverser ? handle.message() : handle); - } - } - - public Envelope dequeue() { - synchronized (MUTEX) { - for (final Queue queue : this.messages) { - if (!queue.isEmpty()) { - final Object m = queue.poll(); - return m instanceof Traverser ? new Envelope(m, ActorRef.noSender()) : (Envelope) m; - } - } - return null; - } - } - - public int numberOfMessages() { - synchronized (MUTEX) { - int counter = 0; - for (final Queue queue : this.messages) { - counter = counter + queue.size(); - } - return counter; - } - } - - public boolean hasMessages() { - synchronized (MUTEX) { - for (final Queue queue : this.messages) { - if (!queue.isEmpty()) - return true; - } - return false; - } - } - - public void cleanUp(final ActorRef owner, final MessageQueue deadLetters) { - synchronized (MUTEX) { - for (final Queue queue : this.messages) { - while (!queue.isEmpty()) { - final Object m = queue.poll(); - deadLetters.enqueue(owner, m instanceof Traverser ? new Envelope(m, ActorRef.noSender()) : (Envelope) m); - } - } - } - } - } - - // This constructor signature must exist, it will be called by Akka - public TraverserMailbox(final ActorSystem.Settings settings, final Config config) { - try { - final String[] messages = ((String) settings.config().getAnyRef("message-priorities")).replace("[", "").replace("]", "").split(","); - for (final String clazz : messages) { - this.messagePriorities.add(Class.forName(clazz.trim())); - } - } catch (final ClassNotFoundException e) { - throw new IllegalArgumentException(e.getMessage(), e); - } - } - - // The create method is called to create the MessageQueue - public MessageQueue create(final Option<ActorRef> owner, final Option<ActorSystem> system) { - return new TraverserMessageQueue(this.messagePriorities); - } - - public static interface TraverserSetSemantics { - - } -} http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/22d4bb6c/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actor/WorkerActor.java ---------------------------------------------------------------------- diff --git a/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actor/WorkerActor.java b/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actor/WorkerActor.java index c023312..6aab38b 100644 --- a/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actor/WorkerActor.java +++ b/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actor/WorkerActor.java @@ -37,7 +37,7 @@ import java.util.Map; /** * @author Marko A. Rodriguez (http://markorodriguez.com) */ -public final class WorkerActor extends AbstractActor implements RequiresMessageQueue<TraverserMailbox.TraverserSetSemantics>, Actor.Worker { +public final class WorkerActor extends AbstractActor implements RequiresMessageQueue<ActorMailbox.ActorSemantics>, Actor.Worker { private final Partition localPartition; private final Address.Worker self; http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/22d4bb6c/akka-gremlin/src/main/resources/application.conf ---------------------------------------------------------------------- diff --git a/akka-gremlin/src/main/resources/application.conf b/akka-gremlin/src/main/resources/application.conf index a9bda12..7ced92c 100644 --- a/akka-gremlin/src/main/resources/application.conf +++ b/akka-gremlin/src/main/resources/application.conf @@ -1,9 +1,9 @@ custom-dispatcher { - mailbox-requirement = "org.apache.tinkerpop.gremlin.akka.process.actor.TraverserMailbox$TraverserSetSemantics" + mailbox-requirement = "org.apache.tinkerpop.gremlin.akka.process.actor.ActorMailbox$ActorSemantics" } akka.actor.mailbox.requirements { - "org.apache.tinkerpop.gremlin.akka.process.actor.TraverserMailbox$TraverserSetSemantics" = custom-dispatcher-mailbox + "org.apache.tinkerpop.gremlin.akka.process.actor.ActorMailbox$ActorSemantics" = custom-dispatcher-mailbox } akka { @@ -11,5 +11,5 @@ akka { } custom-dispatcher-mailbox { - mailbox-type = "org.apache.tinkerpop.gremlin.akka.process.actor.TraverserMailbox" + mailbox-type = "org.apache.tinkerpop.gremlin.akka.process.actor.ActorMailbox" } \ No newline at end of file http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/22d4bb6c/akka-gremlin/src/test/java/org/apache/tinkerpop/gremlin/akka/process/AkkaActorsProvider.java ---------------------------------------------------------------------- diff --git a/akka-gremlin/src/test/java/org/apache/tinkerpop/gremlin/akka/process/AkkaActorsProvider.java b/akka-gremlin/src/test/java/org/apache/tinkerpop/gremlin/akka/process/AkkaActorsProvider.java index 7c8c1a1..2cfb046 100644 --- a/akka-gremlin/src/test/java/org/apache/tinkerpop/gremlin/akka/process/AkkaActorsProvider.java +++ b/akka-gremlin/src/test/java/org/apache/tinkerpop/gremlin/akka/process/AkkaActorsProvider.java @@ -22,7 +22,7 @@ package org.apache.tinkerpop.gremlin.akka.process; import org.apache.commons.configuration.Configuration; import org.apache.tinkerpop.gremlin.AbstractGraphProvider; import org.apache.tinkerpop.gremlin.LoadGraphWith; -import org.apache.tinkerpop.gremlin.akka.process.actor.AkkaActors; +import org.apache.tinkerpop.gremlin.akka.process.actor.AkkaGraphActors; import org.apache.tinkerpop.gremlin.process.actor.traversal.strategy.decoration.ActorProgramStrategy; import org.apache.tinkerpop.gremlin.process.traversal.TraversalInterruptionTest; import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource; @@ -153,7 +153,7 @@ public class AkkaActorsProvider extends AbstractGraphProvider { //throw new VerificationException("This test current does not work with Gremlin-Python", EmptyTraversal.instance()); else { final GraphTraversalSource g = graph.traversal(); - return g.withStrategies(new ActorProgramStrategy(AkkaActors.class, new HashPartitioner(graph.partitioner(), 3))); + return g.withStrategies(new ActorProgramStrategy(AkkaGraphActors.class, new HashPartitioner(graph.partitioner(), 3))); } } } \ No newline at end of file http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/22d4bb6c/akka-gremlin/src/test/java/org/apache/tinkerpop/gremlin/akka/process/AkkaPlayTest.java ---------------------------------------------------------------------- diff --git a/akka-gremlin/src/test/java/org/apache/tinkerpop/gremlin/akka/process/AkkaPlayTest.java b/akka-gremlin/src/test/java/org/apache/tinkerpop/gremlin/akka/process/AkkaPlayTest.java index 5be2a67..3874d06 100644 --- a/akka-gremlin/src/test/java/org/apache/tinkerpop/gremlin/akka/process/AkkaPlayTest.java +++ b/akka-gremlin/src/test/java/org/apache/tinkerpop/gremlin/akka/process/AkkaPlayTest.java @@ -19,7 +19,7 @@ package org.apache.tinkerpop.gremlin.akka.process; -import org.apache.tinkerpop.gremlin.akka.process.actor.AkkaActors; +import org.apache.tinkerpop.gremlin.akka.process.actor.AkkaGraphActors; import org.apache.tinkerpop.gremlin.process.actor.traversal.strategy.decoration.ActorProgramStrategy; import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource; import org.apache.tinkerpop.gremlin.structure.Graph; @@ -28,6 +28,8 @@ import org.apache.tinkerpop.gremlin.structure.util.partitioner.HashPartitioner; import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerGraph; import org.junit.Test; +import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.outE; + /** * @author Marko A. Rodriguez (http://markorodriguez.com) */ @@ -37,8 +39,8 @@ public class AkkaPlayTest { public void testPlay1() throws Exception { final Graph graph = TinkerGraph.open(); graph.io(GryoIo.build()).readGraph("../data/tinkerpop-modern.kryo"); - GraphTraversalSource g = graph.traversal().withStrategies(new ActorProgramStrategy(AkkaActors.class, new HashPartitioner(graph.partitioner(), 3))); - System.out.println(g.V().values("name").toList()); + GraphTraversalSource g = graph.traversal().withStrategies(new ActorProgramStrategy(AkkaGraphActors.class, new HashPartitioner(graph.partitioner(), 3))); + System.out.println(g.V().group().by("name").by(outE().values("weight").fold()).toList()); //3, 1.9, 1 /*for (int i = 0; i < 10000; i++) { final Graph graph = TinkerGraph.open(); http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/22d4bb6c/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/Actors.java ---------------------------------------------------------------------- diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/Actors.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/Actors.java deleted file mode 100644 index 7b0c4a4..0000000 --- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/Actors.java +++ /dev/null @@ -1,32 +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.tinkerpop.gremlin.process.actor; - -import java.util.concurrent.Future; - -/** - * @author Marko A. Rodriguez (http://markorodriguez.com) - */ -public interface Actors<R> { - - public Address.Master master(); - - public Future<R> submit(); -} http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/22d4bb6c/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/GraphActors.java ---------------------------------------------------------------------- diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/GraphActors.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/GraphActors.java new file mode 100644 index 0000000..c0cdc9e --- /dev/null +++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/GraphActors.java @@ -0,0 +1,32 @@ +/* + * 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.gremlin.process.actor; + +import java.util.concurrent.Future; + +/** + * @author Marko A. Rodriguez (http://markorodriguez.com) + */ +public interface GraphActors<R> { + + public Address.Master master(); + + public Future<R> submit(); +} http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/22d4bb6c/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/TraversalActorProgram.java ---------------------------------------------------------------------- diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/TraversalActorProgram.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/TraversalActorProgram.java index dbf932d..6528a62 100644 --- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/TraversalActorProgram.java +++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/TraversalActorProgram.java @@ -30,17 +30,20 @@ import org.apache.tinkerpop.gremlin.process.actor.traversal.message.Terminate; import org.apache.tinkerpop.gremlin.process.actor.traversal.message.VoteToHaltMessage; import org.apache.tinkerpop.gremlin.process.actor.traversal.strategy.decoration.ActorProgramStrategy; import org.apache.tinkerpop.gremlin.process.actor.traversal.strategy.verification.ActorVerificationStrategy; -import org.apache.tinkerpop.gremlin.process.computer.traversal.step.map.TraversalVertexProgramStep; -import org.apache.tinkerpop.gremlin.process.computer.traversal.strategy.decoration.VertexProgramStrategy; import org.apache.tinkerpop.gremlin.process.traversal.Traversal; import org.apache.tinkerpop.gremlin.process.traversal.TraversalStrategies; +import org.apache.tinkerpop.gremlin.process.traversal.TraversalStrategy; import org.apache.tinkerpop.gremlin.process.traversal.Traverser; -import org.apache.tinkerpop.gremlin.process.traversal.strategy.verification.ComputerVerificationStrategy; +import org.apache.tinkerpop.gremlin.process.traversal.strategy.optimization.InlineFilterStrategy; +import org.apache.tinkerpop.gremlin.process.traversal.strategy.optimization.LazyBarrierStrategy; +import org.apache.tinkerpop.gremlin.process.traversal.strategy.optimization.MatchPredicateStrategy; +import org.apache.tinkerpop.gremlin.process.traversal.strategy.optimization.PathRetractionStrategy; +import org.apache.tinkerpop.gremlin.process.traversal.strategy.optimization.RepeatUnrollStrategy; import org.apache.tinkerpop.gremlin.process.traversal.strategy.verification.ReadOnlyStrategy; -import org.apache.tinkerpop.gremlin.process.traversal.strategy.verification.StandardVerificationStrategy; import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.TraverserSet; import org.apache.tinkerpop.gremlin.structure.Partitioner; +import java.util.ArrayList; import java.util.Arrays; import java.util.List; @@ -50,8 +53,8 @@ import java.util.List; public final class TraversalActorProgram<R> implements ActorProgram<TraverserSet<R>> { private static final List<Class> MESSAGE_PRIORITIES = Arrays.asList( - Traverser.class, StartMessage.class, + Traverser.class, SideEffectAddMessage.class, BarrierAddMessage.class, SideEffectSetMessage.class, @@ -65,12 +68,23 @@ public final class TraversalActorProgram<R> implements ActorProgram<TraverserSet public TraversalActorProgram(final Traversal.Admin<?, R> traversal, final Partitioner partitioner) { this.partitioner = partitioner; - final TraversalStrategies strategies = traversal.getStrategies().clone(); - strategies.addStrategies(ActorVerificationStrategy.instance(), VertexProgramStrategy.instance(), ReadOnlyStrategy.instance()); - strategies.removeStrategies(ActorProgramStrategy.class, ComputerVerificationStrategy.class, StandardVerificationStrategy.class); - traversal.setStrategies(strategies); - traversal.applyStrategies(); - this.traversal = (Traversal.Admin) ((TraversalVertexProgramStep) traversal.getStartStep()).computerTraversal.get(); + this.traversal = traversal; + final TraversalStrategies strategies = this.traversal.getStrategies().clone(); + strategies.addStrategies(ActorVerificationStrategy.instance(), ReadOnlyStrategy.instance()); + // TODO: make TinkerGraph/etc. strategies smart about actors + new ArrayList<>(strategies.toList()).stream(). + filter(s -> s instanceof TraversalStrategy.ProviderOptimizationStrategy). + map(TraversalStrategy::getClass). + forEach(strategies::removeStrategies); + strategies.removeStrategies( + ActorProgramStrategy.class, + LazyBarrierStrategy.class, + RepeatUnrollStrategy.class, + MatchPredicateStrategy.class, + InlineFilterStrategy.class, + PathRetractionStrategy.class); + this.traversal.setStrategies(strategies); + this.traversal.applyStrategies(); } @Override http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/22d4bb6c/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/step/map/ActorStep.java ---------------------------------------------------------------------- diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/step/map/ActorStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/step/map/ActorStep.java deleted file mode 100644 index 7da9a12..0000000 --- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/step/map/ActorStep.java +++ /dev/null @@ -1,79 +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.tinkerpop.gremlin.process.actor.traversal.step.map; - -/** - * @author Marko A. Rodriguez (http://markorodriguez.com) - */ - -import org.apache.tinkerpop.gremlin.process.actor.ActorProgram; -import org.apache.tinkerpop.gremlin.process.actor.Actors; -import org.apache.tinkerpop.gremlin.process.actor.traversal.TraversalActorProgram; -import org.apache.tinkerpop.gremlin.process.actor.traversal.strategy.decoration.ActorProgramStrategy; -import org.apache.tinkerpop.gremlin.process.computer.traversal.strategy.decoration.VertexProgramStrategy; -import org.apache.tinkerpop.gremlin.process.traversal.Traversal; -import org.apache.tinkerpop.gremlin.process.traversal.TraversalStrategies; -import org.apache.tinkerpop.gremlin.process.traversal.Traverser; -import org.apache.tinkerpop.gremlin.process.traversal.step.util.AbstractStep; -import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.TraverserSet; -import org.apache.tinkerpop.gremlin.structure.Partitioner; -import org.apache.tinkerpop.gremlin.structure.util.StringFactory; - -import java.util.NoSuchElementException; - -/** - * @author Marko A. Rodriguez (http://markorodriguez.com) - */ -public final class ActorStep<S, E> extends AbstractStep<E, E> { - - private final Class<? extends Actors> actorsClass; - private final Traversal.Admin<S, E> partitionTraversal; - private final Partitioner partitioner; - - private boolean first = true; - - - public ActorStep(final Traversal.Admin<?, ?> traversal, final Class<? extends Actors> actorsClass, final Partitioner partitioner) { - super(traversal); - this.actorsClass = actorsClass; - this.partitionTraversal = (Traversal.Admin) traversal.clone(); - this.partitioner = partitioner; - } - - @Override - public String toString() { - return StringFactory.stepString(this, this.partitionTraversal); - } - - @Override - protected Traverser.Admin<E> processNextStart() throws NoSuchElementException { - if (this.first) { - this.first = false; - try { - final Actors<TraverserSet<E>> actors = this.actorsClass.getConstructor(ActorProgram.class, Partitioner.class).newInstance(new TraversalActorProgram<E>(this.partitionTraversal, partitioner), this.partitioner); - actors.submit().get().forEach(this.starts::add); - } catch (final Exception e) { - throw new IllegalStateException(e.getMessage(), e); - } - } - return this.starts.next(); - } -} - http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/22d4bb6c/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/step/map/TraversalActorProgramStep.java ---------------------------------------------------------------------- diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/step/map/TraversalActorProgramStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/step/map/TraversalActorProgramStep.java new file mode 100644 index 0000000..de82599 --- /dev/null +++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/step/map/TraversalActorProgramStep.java @@ -0,0 +1,79 @@ +/* + * 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.gremlin.process.actor.traversal.step.map; + +/** + * @author Marko A. Rodriguez (http://markorodriguez.com) + */ + +import org.apache.tinkerpop.gremlin.process.actor.ActorProgram; +import org.apache.tinkerpop.gremlin.process.actor.GraphActors; +import org.apache.tinkerpop.gremlin.process.actor.traversal.TraversalActorProgram; +import org.apache.tinkerpop.gremlin.process.traversal.Traversal; +import org.apache.tinkerpop.gremlin.process.traversal.Traverser; +import org.apache.tinkerpop.gremlin.process.traversal.step.util.AbstractStep; +import org.apache.tinkerpop.gremlin.process.traversal.step.util.EmptyStep; +import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.TraverserSet; +import org.apache.tinkerpop.gremlin.structure.Partitioner; +import org.apache.tinkerpop.gremlin.structure.util.StringFactory; + +import java.util.NoSuchElementException; + +/** + * @author Marko A. Rodriguez (http://markorodriguez.com) + */ +public final class TraversalActorProgramStep<S, E> extends AbstractStep<E, E> { + + private final Class<? extends GraphActors> actorsClass; + private final Traversal.Admin<S, E> actorsTraversal; + private final Partitioner partitioner; + + private boolean first = true; + + + public TraversalActorProgramStep(final Traversal.Admin<?, ?> traversal, final Class<? extends GraphActors> actorsClass, final Partitioner partitioner) { + super(traversal); + this.actorsClass = actorsClass; + this.actorsTraversal = (Traversal.Admin) traversal.clone(); + this.actorsTraversal.setParent(EmptyStep.instance()); + this.partitioner = partitioner; + } + + @Override + public String toString() { + return StringFactory.stepString(this, this.actorsTraversal); + } + + @Override + protected Traverser.Admin<E> processNextStart() throws NoSuchElementException { + if (this.first) { + this.first = false; + try { + final GraphActors<TraverserSet<E>> graphActors = this.actorsClass.getConstructor(ActorProgram.class, Partitioner.class). + newInstance(new TraversalActorProgram<E>(this.actorsTraversal, this.partitioner), this.partitioner); + graphActors.submit().get().forEach(this.starts::add); + } catch (final Exception e) { + throw new IllegalStateException(e.getMessage(), e); + } + } + return this.starts.next(); + } +} + http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/22d4bb6c/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/strategy/decoration/ActorProgramStrategy.java ---------------------------------------------------------------------- diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/strategy/decoration/ActorProgramStrategy.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/strategy/decoration/ActorProgramStrategy.java index 1994de3..358e7db 100644 --- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/strategy/decoration/ActorProgramStrategy.java +++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/strategy/decoration/ActorProgramStrategy.java @@ -19,8 +19,8 @@ package org.apache.tinkerpop.gremlin.process.actor.traversal.strategy.decoration; -import org.apache.tinkerpop.gremlin.process.actor.Actors; -import org.apache.tinkerpop.gremlin.process.actor.traversal.step.map.ActorStep; +import org.apache.tinkerpop.gremlin.process.actor.GraphActors; +import org.apache.tinkerpop.gremlin.process.actor.traversal.step.map.TraversalActorProgramStep; import org.apache.tinkerpop.gremlin.process.computer.traversal.strategy.decoration.VertexProgramStrategy; import org.apache.tinkerpop.gremlin.process.remote.traversal.strategy.decoration.RemoteStrategy; import org.apache.tinkerpop.gremlin.process.traversal.Traversal; @@ -45,9 +45,9 @@ public final class ActorProgramStrategy extends AbstractTraversalStrategy<Traver private static final Set<Class<? extends DecorationStrategy>> POSTS = Collections.singleton(VertexProgramStrategy.class); private final Partitioner partitioner; - private final Class<? extends Actors> actors; + private final Class<? extends GraphActors> actors; - public ActorProgramStrategy(final Class<? extends Actors> actors, final Partitioner partitioner) { + public ActorProgramStrategy(final Class<? extends GraphActors> actors, final Partitioner partitioner) { this.actors = actors; this.partitioner = partitioner; } @@ -59,7 +59,7 @@ public final class ActorProgramStrategy extends AbstractTraversalStrategy<Traver if (!(traversal.getParent() instanceof EmptyStep)) return; - final ActorStep<?, ?> actorStep = new ActorStep<>(traversal, this.actors, this.partitioner); + final TraversalActorProgramStep<?, ?> actorStep = new TraversalActorProgramStep<>(traversal, this.actors, this.partitioner); TraversalHelper.removeAllSteps(traversal); traversal.addStep(actorStep);
