ahhh.. a really solid simplification using the Terminate token as the barrier synchronizer. Lots of other little random clean me ups. Things are looking brighter at close of business.
Project: http://git-wip-us.apache.org/repos/asf/tinkerpop/repo Commit: http://git-wip-us.apache.org/repos/asf/tinkerpop/commit/1d883550 Tree: http://git-wip-us.apache.org/repos/asf/tinkerpop/tree/1d883550 Diff: http://git-wip-us.apache.org/repos/asf/tinkerpop/diff/1d883550 Branch: refs/heads/TINKERPOP-1564 Commit: 1d88355067940e5e8594f1034728bc9dcff007b7 Parents: 22d4bb6 Author: Marko A. Rodriguez <okramma...@gmail.com> Authored: Tue Dec 13 19:05:12 2016 -0700 Committer: Marko A. Rodriguez <okramma...@gmail.com> Committed: Thu Jan 19 13:01:41 2017 -0700 ---------------------------------------------------------------------- .../akka/process/AkkaActorsProvider.java | 2 +- .../actor/traversal/TraversalActorProgram.java | 4 +-- .../actor/traversal/TraversalMasterProgram.java | 7 ++-- .../actor/traversal/TraversalWorkerProgram.java | 37 ++++++++------------ .../traversal/message/VoteToHaltMessage.java | 36 ------------------- .../decoration/ActorProgramStrategy.java | 7 ---- .../verification/ActorVerificationStrategy.java | 5 --- 7 files changed, 19 insertions(+), 79 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/1d883550/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 2cfb046..a0703bd 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 @@ -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(AkkaGraphActors.class, new HashPartitioner(graph.partitioner(), 3))); + return g.withStrategies(new ActorProgramStrategy(AkkaGraphActors.class, new HashPartitioner(graph.partitioner(), new Random().nextInt(10) + 1))); } } } \ No newline at end of file http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/1d883550/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 6528a62..f9f86da 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 @@ -27,7 +27,6 @@ import org.apache.tinkerpop.gremlin.process.actor.traversal.message.SideEffectAd import org.apache.tinkerpop.gremlin.process.actor.traversal.message.SideEffectSetMessage; import org.apache.tinkerpop.gremlin.process.actor.traversal.message.StartMessage; 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.traversal.Traversal; @@ -59,8 +58,7 @@ public final class TraversalActorProgram<R> implements ActorProgram<TraverserSet BarrierAddMessage.class, SideEffectSetMessage.class, BarrierDoneMessage.class, - Terminate.class, - VoteToHaltMessage.class); + Terminate.class); private final Traversal.Admin<?, R> traversal; private final Partitioner partitioner; http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/1d883550/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/TraversalMasterProgram.java ---------------------------------------------------------------------- diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/TraversalMasterProgram.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/TraversalMasterProgram.java index d9153e5..1c44b51 100644 --- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/TraversalMasterProgram.java +++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/TraversalMasterProgram.java @@ -28,7 +28,6 @@ import org.apache.tinkerpop.gremlin.process.actor.traversal.message.SideEffectAd import org.apache.tinkerpop.gremlin.process.actor.traversal.message.SideEffectSetMessage; import org.apache.tinkerpop.gremlin.process.actor.traversal.message.StartMessage; 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.traversal.Step; import org.apache.tinkerpop.gremlin.process.traversal.Traversal; import org.apache.tinkerpop.gremlin.process.traversal.Traverser; @@ -40,9 +39,7 @@ import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalMatrix; import org.apache.tinkerpop.gremlin.structure.Element; import org.apache.tinkerpop.gremlin.structure.Partitioner; -import java.util.Arrays; import java.util.HashMap; -import java.util.List; import java.util.Map; /** @@ -87,7 +84,8 @@ final class TraversalMasterProgram<M> implements ActorProgram.Master<M> { this.barriers.put(step.getId(), barrier); } else if (message instanceof SideEffectAddMessage) { this.traversal.getSideEffects().add(((SideEffectAddMessage) message).getKey(), ((SideEffectAddMessage) message).getValue()); - } else if (message instanceof VoteToHaltMessage) { + } else if (message instanceof Terminate) { + assert Terminate.YES == message; if (!this.barriers.isEmpty()) { for (final Barrier barrier : this.barriers.values()) { final Step<?, ?> step = (Step) barrier; @@ -104,7 +102,6 @@ final class TraversalMasterProgram<M> implements ActorProgram.Master<M> { } } this.barriers.clear(); - this.master.send(this.leaderWorker, StartMessage.instance()); this.master.send(this.leaderWorker, Terminate.MAYBE); } else { while (this.traversal.hasNext()) { http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/1d883550/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/TraversalWorkerProgram.java ---------------------------------------------------------------------- diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/TraversalWorkerProgram.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/TraversalWorkerProgram.java index 8579331..08d2cff 100644 --- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/TraversalWorkerProgram.java +++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/TraversalWorkerProgram.java @@ -27,7 +27,6 @@ import org.apache.tinkerpop.gremlin.process.actor.traversal.message.BarrierDoneM import org.apache.tinkerpop.gremlin.process.actor.traversal.message.SideEffectSetMessage; import org.apache.tinkerpop.gremlin.process.actor.traversal.message.StartMessage; 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.traversal.Step; import org.apache.tinkerpop.gremlin.process.traversal.Traversal; import org.apache.tinkerpop.gremlin.process.traversal.Traverser; @@ -107,12 +106,14 @@ final class TraversalWorkerProgram<M> implements ActorProgram.Worker<M> { this.processTraverser((Traverser.Admin) message); } else if (message instanceof SideEffectSetMessage) { this.matrix.getTraversal().getSideEffects().set(((SideEffectSetMessage) message).getKey(), ((SideEffectSetMessage) message).getValue()); + } else if (message instanceof BarrierDoneMessage) { + final Step<?, ?> step = (Step) this.matrix.getStepById(((BarrierDoneMessage) message).getStepId()); + while (step.hasNext()) { + sendTraverser(step.next()); + } } else if (message instanceof Terminate) { - assert this.isLeader || this.terminate != Terminate.MAYBE; + assert null == this.terminate; this.terminate = (Terminate) message; - this.self.send(this.self.address(), VoteToHaltMessage.instance()); - } else if (message instanceof VoteToHaltMessage) { - // if there is a barrier and thus, halting at barrier, then process barrier if (!this.barriers.isEmpty()) { for (final Barrier barrier : this.barriers.values()) { while (barrier.hasNextBarrier()) { @@ -120,25 +121,17 @@ final class TraversalWorkerProgram<M> implements ActorProgram.Worker<M> { } } this.barriers.clear(); - this.voteToHalt = false; } // use termination token to determine termination condition - if (null != this.terminate) { - if (this.isLeader) { - if (this.voteToHalt && Terminate.YES == this.terminate) - this.self.send(this.self.master(), VoteToHaltMessage.instance()); - else - this.self.send(this.neighborWorker, Terminate.YES); - } else - this.self.send(this.neighborWorker, this.voteToHalt ? this.terminate : Terminate.NO); - this.terminate = null; - this.voteToHalt = true; - } - } else if (message instanceof BarrierDoneMessage) { - final Step<?, ?> step = (Step) this.matrix.getStepById(((BarrierDoneMessage) message).getStepId()); - while (step.hasNext()) { - sendTraverser(step.next()); - } + if (this.isLeader) { + if (this.voteToHalt && Terminate.YES == this.terminate) + this.self.send(this.self.master(), Terminate.YES); + else + this.self.send(this.neighborWorker, Terminate.YES); + } else + this.self.send(this.neighborWorker, this.voteToHalt ? this.terminate : Terminate.NO); + this.terminate = null; + this.voteToHalt = true; } else { throw new IllegalArgumentException("The following message is unknown: " + message); } http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/1d883550/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/message/VoteToHaltMessage.java ---------------------------------------------------------------------- diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/message/VoteToHaltMessage.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/message/VoteToHaltMessage.java deleted file mode 100644 index 4e9a900..0000000 --- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/message/VoteToHaltMessage.java +++ /dev/null @@ -1,36 +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.message; - -/** - * @author Marko A. Rodriguez (http://markorodriguez.com) - */ -public final class VoteToHaltMessage { - - private static final VoteToHaltMessage INSTANCE = new VoteToHaltMessage(); - - private VoteToHaltMessage() { - } - - public static VoteToHaltMessage instance() { - return INSTANCE; - } -} - http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/1d883550/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 358e7db..1ab13fc 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 @@ -21,7 +21,6 @@ package org.apache.tinkerpop.gremlin.process.actor.traversal.strategy.decoration 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; import org.apache.tinkerpop.gremlin.process.traversal.TraversalStrategy; @@ -42,7 +41,6 @@ public final class ActorProgramStrategy extends AbstractTraversalStrategy<Traver private static final Set<Class<? extends DecorationStrategy>> PRIORS = Collections.singleton(RemoteStrategy.class); - private static final Set<Class<? extends DecorationStrategy>> POSTS = Collections.singleton(VertexProgramStrategy.class); private final Partitioner partitioner; private final Class<? extends GraphActors> actors; @@ -70,11 +68,6 @@ public final class ActorProgramStrategy extends AbstractTraversalStrategy<Traver } @Override - public Set<Class<? extends DecorationStrategy>> applyPost() { - return POSTS; - } - - @Override public Set<Class<? extends DecorationStrategy>> applyPrior() { return PRIORS; } http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/1d883550/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/strategy/verification/ActorVerificationStrategy.java ---------------------------------------------------------------------- diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/strategy/verification/ActorVerificationStrategy.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/strategy/verification/ActorVerificationStrategy.java index 5e1913d..30ea2c5 100644 --- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/strategy/verification/ActorVerificationStrategy.java +++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/strategy/verification/ActorVerificationStrategy.java @@ -54,11 +54,6 @@ public final class ActorVerificationStrategy extends AbstractTraversalStrategy<T // only global children are graph computing if (globalChild && step instanceof GraphComputing) ((GraphComputing) step).onGraphComputer(); - - for (String label : step.getLabels()) { - if (Graph.Hidden.isHidden(label)) - step.removeLabel(label); - } } }