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 2017/01/04 15:01:26 UTC
[1/3] tinkerpop git commit: refactored GraphActors packaging -- its
not actor/, but actors/. JavaDoc and various cleanups. Also,
about to NOT serialize a traversal but instead use Bytecode. Next push will do
this with TraveraslVertexProgram.
Repository: tinkerpop
Updated Branches:
refs/heads/TINKERPOP-1564 0aa3076e9 -> 7ee2d3726
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/GraphActors.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/GraphActors.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/GraphActors.java
new file mode 100644
index 0000000..9b8fe38
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/GraphActors.java
@@ -0,0 +1,107 @@
+/*
+ * 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.actors;
+
+import org.apache.commons.configuration.BaseConfiguration;
+import org.apache.commons.configuration.Configuration;
+import org.apache.tinkerpop.gremlin.process.Processor;
+import org.apache.tinkerpop.gremlin.process.actors.traversal.strategy.decoration.ActorProgramStrategy;
+import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
+import org.apache.tinkerpop.gremlin.process.traversal.strategy.ProcessorTraversalStrategy;
+import org.apache.tinkerpop.gremlin.structure.Graph;
+
+import java.util.concurrent.Future;
+
+/**
+ * GraphActors is a message-passing based graph {@link Processor} that is:
+ * asynchronous, distributed, and partition-centric.
+ *
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public interface GraphActors<R> extends Processor {
+
+ public static final String GRAPH_ACTORS = "gremlin.graphActors";
+ public static final String GRAPH_ACTORS_WORKERS = "gremlin.graphActors.workers";
+
+ /**
+ * Provide the {@link ActorProgram} that the GraphActors will execute.
+ *
+ * @param program the program to execute
+ * @return the updated GraphActors with newly defined program
+ */
+ public GraphActors<R> program(final ActorProgram program);
+
+ /**
+ * Specify the number of workers per {@link Graph} {@link org.apache.tinkerpop.gremlin.structure.Partition}.
+ *
+ * @param workers the number of workers per partition
+ * @return the updated GraphActors with newly defined workers
+ */
+ public GraphActors<R> workers(final int workers);
+
+ /**
+ * Add an arbitrary configuration to the GraphActors system.
+ * Typically, these configurations are provider-specific and do not generalize across all GraphActor implementations.
+ *
+ * @param key the key of the configuration
+ * @param value the value of the configuration
+ * @return the updated GraphActors with newly defined configuration
+ */
+ public GraphActors<R> configure(final String key, final Object value);
+
+ /**
+ * Execute the {@link ActorProgram} on the {@link GraphActors} system against the specified {@link Graph}.
+ *
+ * @return a {@link Future} denoting a reference to the asynchronous computation's result
+ */
+ @Override
+ public Future<R> submit(final Graph graph);
+
+ /**
+ * Returns an {@link ActorProgramStrategy} which enables a {@link Traversal} to execute on {@link GraphActors}.
+ *
+ * @return a {@link org.apache.tinkerpop.gremlin.process.traversal.TraversalStrategy} capable of executing traversals on a GraphActors system
+ */
+ @Override
+ public default ProcessorTraversalStrategy<GraphActors> getProcessorTraversalStrategy() {
+ return new ActorProgramStrategy(this);
+ }
+
+ /**
+ * Create an arbitrary GraphActors system given the information contained in the provided {@link Configuration}.
+ *
+ * @param configuration the {@link Configuration} containing, at minimum, {@link GraphActors#GRAPH_ACTORS} system class name
+ * @param <A> the particular type of GraphActors
+ * @return a constructed GraphActors system
+ */
+ public static <A extends GraphActors> A open(final Configuration configuration) {
+ try {
+ return (A) Class.forName(configuration.getString(GRAPH_ACTORS)).getMethod("open", Configuration.class).invoke(null, configuration);
+ } catch (final Exception e) {
+ throw new IllegalArgumentException(e.getMessage(), e);
+ }
+ }
+
+ public static <A extends GraphActors> A open(final Class<A> graphActorsClass) {
+ final BaseConfiguration configuration = new BaseConfiguration();
+ configuration.setProperty(GRAPH_ACTORS, graphActorsClass.getCanonicalName());
+ return GraphActors.open(configuration);
+ }
+}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/TraversalActorProgram.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/TraversalActorProgram.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/TraversalActorProgram.java
new file mode 100644
index 0000000..484b904
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/TraversalActorProgram.java
@@ -0,0 +1,129 @@
+/*
+ * 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.actors.traversal;
+
+import org.apache.commons.configuration.Configuration;
+import org.apache.tinkerpop.gremlin.jsr223.JavaTranslator;
+import org.apache.tinkerpop.gremlin.process.actors.Actor;
+import org.apache.tinkerpop.gremlin.process.actors.ActorProgram;
+import org.apache.tinkerpop.gremlin.process.actors.traversal.message.BarrierAddMessage;
+import org.apache.tinkerpop.gremlin.process.actors.traversal.message.BarrierDoneMessage;
+import org.apache.tinkerpop.gremlin.process.actors.traversal.message.SideEffectAddMessage;
+import org.apache.tinkerpop.gremlin.process.actors.traversal.message.SideEffectSetMessage;
+import org.apache.tinkerpop.gremlin.process.actors.traversal.message.StartMessage;
+import org.apache.tinkerpop.gremlin.process.actors.traversal.message.Terminate;
+import org.apache.tinkerpop.gremlin.process.actors.traversal.strategy.decoration.ActorProgramStrategy;
+import org.apache.tinkerpop.gremlin.process.actors.traversal.strategy.verification.ActorVerificationStrategy;
+import org.apache.tinkerpop.gremlin.process.traversal.Bytecode;
+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.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.traverser.util.TraverserSet;
+import org.apache.tinkerpop.gremlin.structure.Graph;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public final class TraversalActorProgram<R> implements ActorProgram {
+
+ public static final String TRAVERSAL_ACTOR_PROGRAM_BYTECODE = "gremlin.traversalActorProgram.bytecode";
+
+ private static final List<Class> MESSAGE_PRIORITIES = Arrays.asList(
+ StartMessage.class,
+ Traverser.class,
+ SideEffectAddMessage.class,
+ BarrierAddMessage.class,
+ SideEffectSetMessage.class,
+ BarrierDoneMessage.class,
+ Terminate.class);
+
+ private Traversal.Admin<?, R> traversal;
+ public TraverserSet<R> result = new TraverserSet<>();
+
+ public TraversalActorProgram(final Traversal.Admin<?, R> traversal) {
+ 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
+ public void storeState(final Configuration configuration) {
+ configuration.setProperty(ACTOR_PROGRAM, TraversalActorProgram.class.getCanonicalName());
+ configuration.setProperty(TRAVERSAL_ACTOR_PROGRAM_BYTECODE, this.traversal.getBytecode());
+ }
+
+ @Override
+ public void loadState(final Graph graph, final Configuration configuration) {
+ final Bytecode bytecode = (Bytecode) configuration.getProperty(TRAVERSAL_ACTOR_PROGRAM_BYTECODE);
+ this.traversal = (Traversal.Admin<?, R>) JavaTranslator.of(graph.traversal()).translate(bytecode);
+ }
+
+ @Override
+ public TraversalActorProgram.Worker createWorkerProgram(final Actor.Worker worker) {
+ return new TraversalWorkerProgram(worker, this.traversal.clone());
+ }
+
+ @Override
+ public TraversalActorProgram.Master createMasterProgram(final Actor.Master master) {
+ return new TraversalMasterProgram(master, this.traversal.clone(), this.result);
+ }
+
+ @Override
+ public Optional<List<Class>> getMessagePriorities() {
+ return Optional.of(MESSAGE_PRIORITIES);
+ }
+
+ @Override
+ public TraversalActorProgram<R> clone() {
+ try {
+ final TraversalActorProgram<R> clone = (TraversalActorProgram<R>) super.clone();
+ clone.traversal = this.traversal.clone();
+ return clone;
+ } catch (final CloneNotSupportedException e) {
+ throw new IllegalStateException(e.getMessage(), e);
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/TraversalMasterProgram.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/TraversalMasterProgram.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/TraversalMasterProgram.java
new file mode 100644
index 0000000..e447cdb
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/TraversalMasterProgram.java
@@ -0,0 +1,179 @@
+/*
+ * 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.actors.traversal;
+
+import org.apache.tinkerpop.gremlin.process.actors.Actor;
+import org.apache.tinkerpop.gremlin.process.actors.ActorProgram;
+import org.apache.tinkerpop.gremlin.process.actors.Address;
+import org.apache.tinkerpop.gremlin.process.actors.traversal.message.BarrierAddMessage;
+import org.apache.tinkerpop.gremlin.process.actors.traversal.message.BarrierDoneMessage;
+import org.apache.tinkerpop.gremlin.process.actors.traversal.message.SideEffectAddMessage;
+import org.apache.tinkerpop.gremlin.process.actors.traversal.message.SideEffectSetMessage;
+import org.apache.tinkerpop.gremlin.process.actors.traversal.message.StartMessage;
+import org.apache.tinkerpop.gremlin.process.actors.traversal.message.Terminate;
+import org.apache.tinkerpop.gremlin.process.traversal.Step;
+import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
+import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
+import org.apache.tinkerpop.gremlin.process.traversal.step.Barrier;
+import org.apache.tinkerpop.gremlin.process.traversal.step.Distributing;
+import org.apache.tinkerpop.gremlin.process.traversal.step.LocalBarrier;
+import org.apache.tinkerpop.gremlin.process.traversal.step.Pushing;
+import org.apache.tinkerpop.gremlin.process.traversal.step.SideEffectCapable;
+import org.apache.tinkerpop.gremlin.process.traversal.step.filter.RangeGlobalStep;
+import org.apache.tinkerpop.gremlin.process.traversal.step.filter.TailGlobalStep;
+import org.apache.tinkerpop.gremlin.process.traversal.step.map.OrderGlobalStep;
+import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.OrderedTraverser;
+import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.TraverserSet;
+import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalMatrix;
+import org.apache.tinkerpop.gremlin.structure.Element;
+import org.apache.tinkerpop.gremlin.structure.Partition;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+final class TraversalMasterProgram implements ActorProgram.Master<Object> {
+
+ private final Actor.Master master;
+ private final Traversal.Admin<?, ?> traversal;
+ private final TraversalMatrix<?, ?> matrix;
+ private Map<String, Barrier> barriers = new HashMap<>();
+ private final TraverserSet<?> results;
+ private Address.Worker leaderWorker;
+ private int orderCounter = -1;
+ private final Map<Partition, Address.Worker> partitionToWorkerMap = new HashMap<>();
+
+ public TraversalMasterProgram(final Actor.Master master, final Traversal.Admin<?, ?> traversal, final TraverserSet<?> results) {
+ this.traversal = traversal;
+ // System.out.println("master[created]: " + master.address().getId());
+ // System.out.println(this.traversal);
+ this.matrix = new TraversalMatrix<>(this.traversal);
+ this.results = results;
+ this.master = master;
+ Distributing.configure(this.traversal, true, true);
+ Pushing.configure(this.traversal, true, false);
+ }
+
+ @Override
+ public void setup() {
+ this.leaderWorker = this.master.workers().get(0);
+ for (int i = 0; i < this.master.partitioner().getPartitions().size(); i++) {
+ this.partitionToWorkerMap.put(this.master.partitioner().getPartitions().get(i), this.master.workers().get(i));
+ }
+ this.broadcast(StartMessage.instance());
+ this.master.send(this.leaderWorker, Terminate.MAYBE);
+ }
+
+ @Override
+ public void execute(final Object message) {
+ if (message instanceof Traverser.Admin) {
+ this.processTraverser((Traverser.Admin) message);
+ } else if (message instanceof BarrierAddMessage) {
+ final Barrier barrier = (Barrier) this.matrix.getStepById(((BarrierAddMessage) message).getStepId());
+ final Step<?, ?> step = (Step) barrier;
+ barrier.addBarrier(((BarrierAddMessage) message).getBarrier());
+ 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 Terminate) {
+ assert Terminate.YES == message;
+ if (!this.barriers.isEmpty()) {
+ for (final Barrier barrier : this.barriers.values()) {
+ final Step<?, ?> step = (Step) barrier;
+ if (!(barrier instanceof LocalBarrier)) {
+ this.orderBarrier(step);
+ if (step instanceof OrderGlobalStep) this.orderCounter = 0;
+ while (step.hasNext()) {
+ this.sendTraverser(-1 == this.orderCounter ?
+ step.next() :
+ new OrderedTraverser<>(step.next(), this.orderCounter++));
+ }
+ } else {
+ if (step instanceof SideEffectCapable) {
+ final String key = ((SideEffectCapable) step).getSideEffectKey();
+ this.broadcast(new SideEffectSetMessage(key, this.traversal.getSideEffects().get(key)));
+ }
+ this.broadcast(new BarrierDoneMessage(barrier));
+ barrier.done();
+ }
+ }
+ this.barriers.clear();
+ this.master.send(this.leaderWorker, Terminate.MAYBE);
+ } else {
+ while (this.traversal.hasNext()) {
+ this.results.add((Traverser.Admin) this.traversal.nextTraverser());
+ }
+ if (this.orderCounter != -1)
+ this.results.sort((a, b) -> Integer.compare(((OrderedTraverser<?>) a).order(), ((OrderedTraverser<?>) b).order()));
+
+ this.master.close();
+ }
+ } else {
+ throw new IllegalStateException("Unknown message:" + message);
+ }
+ }
+
+ @Override
+ public void terminate() {
+ this.master.result().setResult(this.results);
+ }
+
+ private void broadcast(final Object message) {
+ for (final Address.Worker worker : this.master.workers()) {
+ this.master.send(worker, message);
+ }
+ }
+
+ private void processTraverser(final Traverser.Admin traverser) {
+ if (traverser.isHalted() || traverser.get() instanceof Element) {
+ this.sendTraverser(traverser);
+ } else {
+ final Step<?, ?> step = this.matrix.<Object, Object, Step<Object, Object>>getStepById(traverser.getStepId());
+ step.addStart(traverser);
+ if (step instanceof Barrier) {
+ this.barriers.put(step.getId(), (Barrier) step);
+ } else {
+ while (step.hasNext()) {
+ this.processTraverser(step.next());
+ }
+ }
+ }
+ }
+
+ private void sendTraverser(final Traverser.Admin traverser) {
+ if (traverser.isHalted())
+ this.results.add(traverser);
+ else if (traverser.get() instanceof Element)
+ this.master.send(this.partitionToWorkerMap.get(this.master.partitioner().getPartition((Element) traverser.get())), traverser);
+ else
+ this.master.send(this.master.address(), traverser);
+ }
+
+ private void orderBarrier(final Step step) {
+ if (this.orderCounter != -1 && step instanceof Barrier && (step instanceof RangeGlobalStep || step instanceof TailGlobalStep)) {
+ final Barrier barrier = (Barrier) step;
+ final TraverserSet<?> rangingBarrier = (TraverserSet<?>) barrier.nextBarrier();
+ rangingBarrier.sort((a, b) -> Integer.compare(((OrderedTraverser<?>) a).order(), ((OrderedTraverser<?>) b).order()));
+ barrier.addBarrier(rangingBarrier);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/TraversalWorkerProgram.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/TraversalWorkerProgram.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/TraversalWorkerProgram.java
new file mode 100644
index 0000000..2aaa7b5
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/TraversalWorkerProgram.java
@@ -0,0 +1,170 @@
+/*
+ * 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.actors.traversal;
+
+import org.apache.tinkerpop.gremlin.process.actors.Actor;
+import org.apache.tinkerpop.gremlin.process.actors.ActorProgram;
+import org.apache.tinkerpop.gremlin.process.actors.Address;
+import org.apache.tinkerpop.gremlin.process.actors.traversal.message.BarrierAddMessage;
+import org.apache.tinkerpop.gremlin.process.actors.traversal.message.BarrierDoneMessage;
+import org.apache.tinkerpop.gremlin.process.actors.traversal.message.SideEffectSetMessage;
+import org.apache.tinkerpop.gremlin.process.actors.traversal.message.StartMessage;
+import org.apache.tinkerpop.gremlin.process.actors.traversal.message.Terminate;
+import org.apache.tinkerpop.gremlin.process.traversal.Step;
+import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
+import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
+import org.apache.tinkerpop.gremlin.process.traversal.step.Barrier;
+import org.apache.tinkerpop.gremlin.process.traversal.step.Distributing;
+import org.apache.tinkerpop.gremlin.process.traversal.step.Pushing;
+import org.apache.tinkerpop.gremlin.process.traversal.step.map.GraphStep;
+import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalHelper;
+import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalMatrix;
+import org.apache.tinkerpop.gremlin.structure.Edge;
+import org.apache.tinkerpop.gremlin.structure.Element;
+import org.apache.tinkerpop.gremlin.structure.Partition;
+import org.apache.tinkerpop.gremlin.structure.Vertex;
+import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+final class TraversalWorkerProgram implements ActorProgram.Worker<Object> {
+
+ private final Actor.Worker self;
+ private final TraversalMatrix<?, ?> matrix;
+ private final Map<Partition, Address.Worker> partitionToWorkerMap = new HashMap<>();
+ //
+ private Address.Worker neighborWorker;
+ private boolean isLeader;
+ private Terminate terminate = null;
+ private boolean voteToHalt = false;
+ private Map<String, Barrier> barriers = new HashMap<>();
+
+ public TraversalWorkerProgram(final Actor.Worker self, final Traversal.Admin<?, ?> traversal) {
+ this.self = self;
+ // System.out.println("worker[created]: " + this.self.address().getId());
+ // set up partition and traversal information
+ final WorkerTraversalSideEffects sideEffects = new WorkerTraversalSideEffects(traversal.getSideEffects(), this.self);
+ TraversalHelper.applyTraversalRecursively(t -> t.setSideEffects(sideEffects), traversal);
+ this.matrix = new TraversalMatrix<>(traversal);
+ Distributing.configure(traversal, false, true);
+ Pushing.configure(traversal, true, false);
+ //////
+ final GraphStep graphStep = (GraphStep) traversal.getStartStep();
+ if (0 == graphStep.getIds().length)
+ ((GraphStep) traversal.getStartStep()).setIteratorSupplier(graphStep.returnsVertex() ? this.self.partition()::vertices : this.self.partition()::edges);
+ else {
+ if (graphStep.returnsVertex())
+ ((GraphStep<Vertex, Vertex>) traversal.getStartStep()).setIteratorSupplier(
+ () -> IteratorUtils.filter(self.partition().vertices(graphStep.getIds()), this.self.partition()::contains));
+ else
+ ((GraphStep<Edge, Edge>) traversal.getStartStep()).setIteratorSupplier(
+ () -> IteratorUtils.filter(self.partition().edges(graphStep.getIds()), this.self.partition()::contains));
+ }
+ }
+
+ @Override
+ public void setup() {
+ // create termination ring topology
+ final int i = this.self.workers().indexOf(this.self.address());
+ this.neighborWorker = this.self.workers().get(i == this.self.workers().size() - 1 ? 0 : i + 1);
+ this.isLeader = i == 0;
+ for (int j = 0; j < this.self.partitioner().getPartitions().size(); j++) {
+ this.partitionToWorkerMap.put(this.self.partitioner().getPartitions().get(j), this.self.workers().get(j));
+ }
+ }
+
+ @Override
+ public void execute(final Object message) {
+ //System.out.println(message + "::" + this.isLeader);
+ if (message instanceof StartMessage) {
+ // initial message from master that says: "start processing"
+ final GraphStep step = (GraphStep) this.matrix.getTraversal().getStartStep();
+ while (step.hasNext()) {
+ this.sendTraverser(step.next());
+ }
+ } else if (message instanceof Traverser.Admin) {
+ 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 null == this.terminate;
+ this.terminate = (Terminate) message;
+ if (!this.barriers.isEmpty()) {
+ for (final Barrier barrier : this.barriers.values()) {
+ while (barrier.hasNextBarrier()) {
+ this.self.send(this.self.master(), new BarrierAddMessage(barrier));
+ }
+ }
+ this.barriers.clear();
+ }
+ // use termination token to determine termination condition
+ 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);
+ }
+ }
+
+ @Override
+ public void terminate() {
+
+ }
+
+ //////////////
+
+ private void processTraverser(final Traverser.Admin traverser) {
+ assert !(traverser.get() instanceof Element) || !traverser.isHalted() || this.self.partition().contains((Element) traverser.get());
+ final Step<?, ?> step = this.matrix.<Object, Object, Step<Object, Object>>getStepById(traverser.getStepId());
+ step.addStart(traverser);
+ if (step instanceof Barrier) {
+ this.barriers.put(step.getId(), (Barrier) step);
+ } else {
+ while (step.hasNext()) {
+ this.sendTraverser(step.next());
+ }
+ }
+ }
+
+ private void sendTraverser(final Traverser.Admin traverser) {
+ this.voteToHalt = false;
+ if (traverser.isHalted())
+ this.self.send(this.self.master(), traverser);
+ else if (traverser.get() instanceof Element && !this.self.partition().contains((Element) traverser.get()))
+ this.self.send(this.partitionToWorkerMap.get(this.self.partitioner().getPartition((Element) traverser.get())), traverser);
+ else
+ this.self.send(this.self.address(), traverser);
+ }
+}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/WorkerTraversalSideEffects.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/WorkerTraversalSideEffects.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/WorkerTraversalSideEffects.java
new file mode 100644
index 0000000..b660eda
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/WorkerTraversalSideEffects.java
@@ -0,0 +1,148 @@
+/*
+ * 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.actors.traversal;
+
+import org.apache.tinkerpop.gremlin.process.actors.Actor;
+import org.apache.tinkerpop.gremlin.process.actors.traversal.message.SideEffectAddMessage;
+import org.apache.tinkerpop.gremlin.process.traversal.TraversalSideEffects;
+
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.BinaryOperator;
+import java.util.function.Supplier;
+import java.util.function.UnaryOperator;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public final class WorkerTraversalSideEffects implements TraversalSideEffects {
+
+ private TraversalSideEffects sideEffects;
+ private Actor.Worker worker;
+
+
+ private WorkerTraversalSideEffects() {
+ // for serialization
+ }
+
+ public WorkerTraversalSideEffects(final TraversalSideEffects sideEffects, final Actor.Worker worker) {
+ this.sideEffects = sideEffects;
+ this.worker = worker;
+ }
+
+ public TraversalSideEffects getSideEffects() {
+ return this.sideEffects;
+ }
+
+ @Override
+ public void set(final String key, final Object value) {
+ this.sideEffects.set(key, value);
+ }
+
+ @Override
+ public <V> V get(final String key) throws IllegalArgumentException {
+ return this.sideEffects.get(key);
+ }
+
+ @Override
+ public void remove(final String key) {
+ this.sideEffects.remove(key);
+ }
+
+ @Override
+ public Set<String> keys() {
+ return this.sideEffects.keys();
+ }
+
+ @Override
+ public void add(final String key, final Object value) {
+ this.sideEffects.add(key, value);
+ this.worker.send(this.worker.master(), new SideEffectAddMessage(key, value));
+ }
+
+ @Override
+ public <V> void register(final String key, final Supplier<V> initialValue, final BinaryOperator<V> reducer) {
+ this.sideEffects.register(key, initialValue, reducer);
+ }
+
+ @Override
+ public <V> void registerIfAbsent(final String key, final Supplier<V> initialValue, final BinaryOperator<V> reducer) {
+ this.sideEffects.registerIfAbsent(key, initialValue, reducer);
+ }
+
+ @Override
+ public <V> BinaryOperator<V> getReducer(final String key) {
+ return this.sideEffects.getReducer(key);
+ }
+
+ @Override
+ public <V> Supplier<V> getSupplier(final String key) {
+ return this.sideEffects.getSupplier(key);
+ }
+
+ @Override
+ @Deprecated
+ public void registerSupplier(final String key, final Supplier supplier) {
+ this.sideEffects.registerSupplier(key, supplier);
+ }
+
+ @Override
+ @Deprecated
+ public <V> Optional<Supplier<V>> getRegisteredSupplier(final String key) {
+ return this.sideEffects.getRegisteredSupplier(key);
+ }
+
+ @Override
+ public <S> void setSack(final Supplier<S> initialValue, final UnaryOperator<S> splitOperator, final BinaryOperator<S> mergeOperator) {
+ this.sideEffects.setSack(initialValue, splitOperator, mergeOperator);
+ }
+
+ @Override
+ public <S> Supplier<S> getSackInitialValue() {
+ return this.sideEffects.getSackInitialValue();
+ }
+
+ @Override
+ public <S> UnaryOperator<S> getSackSplitter() {
+ return this.sideEffects.getSackSplitter();
+ }
+
+ @Override
+ public <S> BinaryOperator<S> getSackMerger() {
+ return this.sideEffects.getSackMerger();
+ }
+
+ @Override
+ public TraversalSideEffects clone() {
+ try {
+ final WorkerTraversalSideEffects clone = (WorkerTraversalSideEffects) super.clone();
+ clone.sideEffects = this.sideEffects.clone();
+ return clone;
+ } catch (final CloneNotSupportedException e) {
+ throw new IllegalStateException(e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public void mergeInto(final TraversalSideEffects sideEffects) {
+ this.sideEffects.mergeInto(sideEffects);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/message/BarrierAddMessage.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/message/BarrierAddMessage.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/message/BarrierAddMessage.java
new file mode 100644
index 0000000..ac4c61d
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/message/BarrierAddMessage.java
@@ -0,0 +1,47 @@
+/*
+ * 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.actors.traversal.message;
+
+import org.apache.tinkerpop.gremlin.process.traversal.Step;
+import org.apache.tinkerpop.gremlin.process.traversal.step.Barrier;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public final class BarrierAddMessage {
+
+ private final Object barrier;
+ private final String stepId;
+
+ public BarrierAddMessage(final Barrier barrier) {
+ this.barrier = barrier.nextBarrier();
+ this.stepId = ((Step) barrier).getId();
+ }
+
+ public Object getBarrier() {
+ return this.barrier;
+ }
+
+ public String getStepId() {
+ return this.stepId;
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/message/BarrierDoneMessage.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/message/BarrierDoneMessage.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/message/BarrierDoneMessage.java
new file mode 100644
index 0000000..7979c33
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/message/BarrierDoneMessage.java
@@ -0,0 +1,41 @@
+/*
+ * 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.actors.traversal.message;
+
+import org.apache.tinkerpop.gremlin.process.traversal.Step;
+import org.apache.tinkerpop.gremlin.process.traversal.step.Barrier;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public final class BarrierDoneMessage {
+
+ private final String stepId;
+
+ public BarrierDoneMessage(final Barrier barrier) {
+ this.stepId = ((Step) barrier).getId();
+
+ }
+
+ public String getStepId() {
+ return this.stepId;
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/message/SideEffectAddMessage.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/message/SideEffectAddMessage.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/message/SideEffectAddMessage.java
new file mode 100644
index 0000000..1c0a9de
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/message/SideEffectAddMessage.java
@@ -0,0 +1,43 @@
+/*
+ * 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.actors.traversal.message;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public final class SideEffectAddMessage {
+
+ private final String key;
+ private final Object value;
+
+ public SideEffectAddMessage(final String key, final Object value) {
+ this.value = value;
+ this.key = key;
+ }
+
+ public String getKey() {
+ return this.key;
+ }
+
+ public Object getValue() {
+ return this.value;
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/message/SideEffectSetMessage.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/message/SideEffectSetMessage.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/message/SideEffectSetMessage.java
new file mode 100644
index 0000000..84788f9
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/message/SideEffectSetMessage.java
@@ -0,0 +1,42 @@
+/*
+ * 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.actors.traversal.message;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public final class SideEffectSetMessage {
+
+ private final String key;
+ private final Object value;
+
+ public SideEffectSetMessage(final String key, final Object value) {
+ this.key = key;
+ this.value = value;
+ }
+
+ public String getKey() {
+ return this.key;
+ }
+
+ public Object getValue() {
+ return this.value;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/message/StartMessage.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/message/StartMessage.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/message/StartMessage.java
new file mode 100644
index 0000000..e704033
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/message/StartMessage.java
@@ -0,0 +1,35 @@
+/*
+ * 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.actors.traversal.message;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public final class StartMessage {
+
+ private static final StartMessage INSTANCE = new StartMessage();
+
+ private StartMessage() {
+ }
+
+ public static StartMessage instance() {
+ return INSTANCE;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/message/Terminate.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/message/Terminate.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/message/Terminate.java
new file mode 100644
index 0000000..5621528
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/message/Terminate.java
@@ -0,0 +1,28 @@
+/*
+ * 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.actors.traversal.message;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public enum Terminate {
+
+ MAYBE, YES, NO
+}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/step/map/TraversalActorProgramStep.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/step/map/TraversalActorProgramStep.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/step/map/TraversalActorProgramStep.java
new file mode 100644
index 0000000..d6b8858
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/step/map/TraversalActorProgramStep.java
@@ -0,0 +1,73 @@
+/*
+ * 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.actors.traversal.step.map;
+
+import org.apache.commons.configuration.Configuration;
+import org.apache.tinkerpop.gremlin.process.actors.ActorProgram;
+import org.apache.tinkerpop.gremlin.process.actors.GraphActors;
+import org.apache.tinkerpop.gremlin.process.actors.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.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 Traversal.Admin<S, E> actorsTraversal;
+ private final Configuration graphActorsConfiguration;
+ private boolean first = true;
+
+ public TraversalActorProgramStep(final Traversal.Admin<?, ?> traversal, final Configuration graphActorsConfiguration) {
+ super(traversal);
+ this.graphActorsConfiguration = graphActorsConfiguration;
+ this.actorsTraversal = (Traversal.Admin) traversal.clone();
+ this.actorsTraversal.setParent(EmptyStep.instance());
+ }
+
+ @Override
+ protected Traverser.Admin<E> processNextStart() throws NoSuchElementException {
+ if (this.first) {
+ this.first = false;
+ try {
+ final GraphActors<TraverserSet<E>> graphActors = GraphActors.open(this.graphActorsConfiguration);
+ final ActorProgram actorProgram = new TraversalActorProgram<>(this.actorsTraversal);
+ graphActors.program(actorProgram).submit(this.getTraversal().getGraph().get()).get().forEach(this.starts::add);
+ } catch (final Exception e) {
+ throw new IllegalStateException(e.getMessage(), e);
+ }
+ }
+ return this.starts.next();
+ }
+
+ @Override
+ public String toString() {
+ return StringFactory.stepString(this, this.actorsTraversal);
+ }
+
+}
+
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/strategy/decoration/ActorProgramStrategy.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/strategy/decoration/ActorProgramStrategy.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/strategy/decoration/ActorProgramStrategy.java
new file mode 100644
index 0000000..6e4365e
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/strategy/decoration/ActorProgramStrategy.java
@@ -0,0 +1,94 @@
+/*
+ * 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.actors.traversal.strategy.decoration;
+
+import org.apache.commons.configuration.Configuration;
+import org.apache.tinkerpop.gremlin.process.actors.GraphActors;
+import org.apache.tinkerpop.gremlin.process.actors.traversal.step.map.TraversalActorProgramStep;
+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;
+import org.apache.tinkerpop.gremlin.process.traversal.step.util.EmptyStep;
+import org.apache.tinkerpop.gremlin.process.traversal.strategy.AbstractTraversalStrategy;
+import org.apache.tinkerpop.gremlin.process.traversal.strategy.ProcessorTraversalStrategy;
+import org.apache.tinkerpop.gremlin.process.traversal.strategy.verification.ReadOnlyStrategy;
+import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalHelper;
+
+import java.util.Collections;
+import java.util.Set;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public final class ActorProgramStrategy extends AbstractTraversalStrategy<TraversalStrategy.DecorationStrategy>
+ implements TraversalStrategy.DecorationStrategy, ProcessorTraversalStrategy<GraphActors> {
+
+ private static final Set<Class<? extends DecorationStrategy>> PRIORS = Collections.singleton(RemoteStrategy.class);
+
+ private final Configuration graphActorsConfiguration;
+
+ public ActorProgramStrategy(final GraphActors graphActors) {
+ this.graphActorsConfiguration = graphActors.configuration();
+ }
+
+ @Override
+ public void apply(final Traversal.Admin<?, ?> traversal) {
+ ReadOnlyStrategy.instance().apply(traversal);
+
+ if (!(traversal.getParent() instanceof EmptyStep))
+ return;
+
+ final TraversalActorProgramStep<?, ?> actorStep = new TraversalActorProgramStep<>(traversal, this.graphActorsConfiguration);
+ TraversalHelper.removeAllSteps(traversal);
+ traversal.addStep(actorStep);
+
+ // validations
+ assert traversal.getStartStep().equals(actorStep);
+ assert traversal.getSteps().size() == 1;
+ assert traversal.getEndStep() == actorStep;
+ }
+
+ @Override
+ public Set<Class<? extends DecorationStrategy>> applyPrior() {
+ return PRIORS;
+ }
+
+ ////////////////////////////////////////////////////////////
+
+ @Override
+ public Configuration getConfiguration() {
+ return this.graphActorsConfiguration;
+ }
+
+ public static ActorProgramStrategy create(final Configuration configuration) {
+ try {
+ return new ActorProgramStrategy(GraphActors.open(configuration));
+ } catch (final Exception e) {
+ throw new IllegalArgumentException(e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public GraphActors getProcessor() {
+ return GraphActors.open(this.graphActorsConfiguration);
+ }
+
+}
+
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/strategy/verification/ActorVerificationStrategy.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/strategy/verification/ActorVerificationStrategy.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/strategy/verification/ActorVerificationStrategy.java
new file mode 100644
index 0000000..cdf5465
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/traversal/strategy/verification/ActorVerificationStrategy.java
@@ -0,0 +1,48 @@
+/*
+ * 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.actors.traversal.strategy.verification;
+
+import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
+import org.apache.tinkerpop.gremlin.process.traversal.TraversalStrategy;
+import org.apache.tinkerpop.gremlin.process.traversal.step.sideEffect.InjectStep;
+import org.apache.tinkerpop.gremlin.process.traversal.strategy.AbstractTraversalStrategy;
+import org.apache.tinkerpop.gremlin.process.traversal.strategy.verification.VerificationException;
+import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalHelper;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public final class ActorVerificationStrategy extends AbstractTraversalStrategy<TraversalStrategy.VerificationStrategy> implements TraversalStrategy.VerificationStrategy {
+
+ private static final ActorVerificationStrategy INSTANCE = new ActorVerificationStrategy();
+
+ private ActorVerificationStrategy() {
+ }
+
+ @Override
+ public void apply(final Traversal.Admin<?, ?> traversal) {
+ if (!TraversalHelper.getStepsOfAssignableClass(InjectStep.class, traversal).isEmpty())
+ throw new VerificationException("Inject traversal currently not supported", traversal);
+ }
+
+ public static ActorVerificationStrategy instance() {
+ return INSTANCE;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/util/DefaultActorsResult.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/util/DefaultActorsResult.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/util/DefaultActorsResult.java
new file mode 100644
index 0000000..208a9a1
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/util/DefaultActorsResult.java
@@ -0,0 +1,42 @@
+/*
+ * 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.actors.util;
+
+import org.apache.tinkerpop.gremlin.process.actors.ActorsResult;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public final class DefaultActorsResult<R> implements ActorsResult<R> {
+
+ private R result;
+
+ public DefaultActorsResult() {
+
+ }
+
+ public R getResult() {
+ return this.result;
+ }
+
+ public void setResult(final R result) {
+ this.result = result;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/util/GraphActorsHelper.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/util/GraphActorsHelper.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/util/GraphActorsHelper.java
new file mode 100644
index 0000000..2af1ecd
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/util/GraphActorsHelper.java
@@ -0,0 +1,48 @@
+/*
+ * 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.actors.util;
+
+import org.apache.commons.configuration.Configuration;
+import org.apache.tinkerpop.gremlin.process.actors.GraphActors;
+import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
+
+import java.util.Iterator;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public final class GraphActorsHelper {
+
+ private GraphActorsHelper() {
+
+ }
+
+ public static GraphActors configure(GraphActors actors, final Configuration configuration) {
+ final Iterator<String> keys = IteratorUtils.asList(configuration.getKeys()).iterator();
+ while (keys.hasNext()) {
+ final String key = keys.next();
+ if (key.equals(GraphActors.GRAPH_ACTORS_WORKERS))
+ actors = actors.workers(configuration.getInt(GraphActors.GRAPH_ACTORS_WORKERS));
+ else if (!key.equals(GraphActors.GRAPH_ACTORS))
+ actors = actors.configure(key, configuration.getProperty(key));
+ }
+ return actors;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/TraversalStrategies.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/TraversalStrategies.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/TraversalStrategies.java
index 96dae61..42bd864 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/TraversalStrategies.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/traversal/TraversalStrategies.java
@@ -18,7 +18,7 @@
*/
package org.apache.tinkerpop.gremlin.process.traversal;
-import org.apache.tinkerpop.gremlin.process.actor.GraphActors;
+import org.apache.tinkerpop.gremlin.process.actors.GraphActors;
import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
import org.apache.tinkerpop.gremlin.process.computer.traversal.strategy.optimization.GraphFilterStrategy;
import org.apache.tinkerpop.gremlin.process.traversal.strategy.decoration.ConnectiveStrategy;
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/util/StringFactory.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/util/StringFactory.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/util/StringFactory.java
index f7c350d..ee7a196 100644
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/util/StringFactory.java
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/structure/util/StringFactory.java
@@ -19,7 +19,7 @@
package org.apache.tinkerpop.gremlin.structure.util;
import org.apache.tinkerpop.gremlin.process.Processor;
-import org.apache.tinkerpop.gremlin.process.actor.GraphActors;
+import org.apache.tinkerpop.gremlin.process.actors.GraphActors;
import org.apache.tinkerpop.gremlin.process.computer.ComputerResult;
import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
import org.apache.tinkerpop.gremlin.process.computer.MapReduce;
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/GraphManager.java
----------------------------------------------------------------------
diff --git a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/GraphManager.java b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/GraphManager.java
index 43b3608..482577f 100644
--- a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/GraphManager.java
+++ b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/GraphManager.java
@@ -19,7 +19,7 @@
package org.apache.tinkerpop.gremlin;
import org.apache.commons.configuration.Configuration;
-import org.apache.tinkerpop.gremlin.process.actor.GraphActors;
+import org.apache.tinkerpop.gremlin.process.actors.GraphActors;
import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
import org.apache.tinkerpop.gremlin.process.traversal.TraversalEngine;
import org.apache.tinkerpop.gremlin.process.traversal.TraversalStrategy;
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/GraphProvider.java
----------------------------------------------------------------------
diff --git a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/GraphProvider.java b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/GraphProvider.java
index 9d63b3c..d0d877d 100644
--- a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/GraphProvider.java
+++ b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/GraphProvider.java
@@ -19,7 +19,7 @@
package org.apache.tinkerpop.gremlin;
import org.apache.commons.configuration.Configuration;
-import org.apache.tinkerpop.gremlin.process.actor.GraphActors;
+import org.apache.tinkerpop.gremlin.process.actors.GraphActors;
import org.apache.tinkerpop.gremlin.process.computer.GraphComputer;
import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
import org.apache.tinkerpop.gremlin.process.traversal.TraversalStrategy;
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/actors/GraphActorsTest.java
----------------------------------------------------------------------
diff --git a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/actors/GraphActorsTest.java b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/actors/GraphActorsTest.java
index ec3ece2..cc9d995 100644
--- a/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/actors/GraphActorsTest.java
+++ b/gremlin-test/src/main/java/org/apache/tinkerpop/gremlin/process/actors/GraphActorsTest.java
@@ -21,7 +21,6 @@ package org.apache.tinkerpop.gremlin.process.actors;
import org.apache.tinkerpop.gremlin.LoadGraphWith;
import org.apache.tinkerpop.gremlin.process.AbstractGremlinProcessTest;
-import org.apache.tinkerpop.gremlin.process.actor.GraphActors;
import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
import org.junit.Test;
[2/3] tinkerpop git commit: refactored GraphActors packaging -- its
not actor/, but actors/. JavaDoc and various cleanups. Also,
about to NOT serialize a traversal but instead use Bytecode. Next push will do
this with TraveraslVertexProgram.
Posted by ok...@apache.org.
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/ActorProgram.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/ActorProgram.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/ActorProgram.java
deleted file mode 100644
index 89002fe..0000000
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/ActorProgram.java
+++ /dev/null
@@ -1,145 +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 org.apache.commons.configuration.Configuration;
-import org.apache.tinkerpop.gremlin.structure.Graph;
-
-import java.util.List;
-import java.util.Optional;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-public interface ActorProgram<M> extends Cloneable {
-
- public static final String ACTOR_PROGRAM = "gremlin.actorProgram";
-
- /**
- * When it is necessary to store the state of the ActorProgram, this method is called.
- * This is typically required when the ActorProgram needs to be serialized to another machine.
- * Note that what is stored is simply the instance/configuration state, not any processed data.
- * The default implementation provided simply stores the ActorProgarm class name for reflective reconstruction.
- * It is typically a good idea to ActorProgram.super.storeState().
- *
- * @param configuration the configuration to store the state of the ActorProgram in.
- */
- public default void storeState(final Configuration configuration) {
- configuration.setProperty(ACTOR_PROGRAM, this.getClass().getName());
- }
-
- /**
- * When it is necessary to load the state of the ActorProgram, this method is called.
- * This is typically required when the ActorProgram needs to be serialized to another machine.
- * Note that what is loaded is simply the instance state, not any processed data.
- *
- * @param graph the graph that the ActorProgram will run against
- * @param configuration the configuration to load the state of the ActorProgram from.
- */
- public default void loadState(final Graph graph, final Configuration configuration) {
-
- }
-
- /**
- * Create the {@link org.apache.tinkerpop.gremlin.process.actor.Actor.Worker} program.
- * This is typically used by {@link Worker} to spawn its program.
- *
- * @param worker the worker actor creating the worker program
- * @return the worker program
- */
- public ActorProgram.Worker createWorkerProgram(final Actor.Worker worker);
-
- /**
- * Create the {@link org.apache.tinkerpop.gremlin.process.actor.Actor.Master} program.
- * This is typically used by {@link Master} to spawn its program.
- *
- * @param master the master actor creating the master program
- * @return the master program
- */
- public ActorProgram.Master createMasterProgram(final Actor.Master master);
-
- /**
- * Get the ordered list of message classes where order determines the priority
- * of message reception by the respective {@link Actor}. For instance,
- * if an {@link Actor} has a message of type {@code X} and a message of type {@code Y}
- * in its message buffer, and {@code X} has a higher priority, it will be fetched
- * first from the buffer. If no list is provided then its FIFO.
- * The default implementation returns an {@link Optional#empty()}.
- *
- * @return the optional ordered list of message priorities.
- */
- public default Optional<List<Class>> getMessagePriorities() {
- return Optional.empty();
- }
-
- /**
- * When multiple workers on a single machine need ActorProgram instances, it is possible to use clone.
- * This will provide a speedier way of generating instances, over the {@link ActorProgram#storeState} and {@link ActorProgram#loadState} model.
- * The default implementation simply returns the object as it assumes that the ActorProgram instance is a stateless singleton.
- *
- * @return A clone of the VertexProgram object
- */
- @SuppressWarnings("CloneDoesntDeclareCloneNotSupportedException")
- public ActorProgram<M> clone();
-
- /**
- * The Worker program is executed by a worker process in the {@link GraphActors} system.
- * There are many workers and a single master.
- * All workers execute the same program.
- *
- * @param <M> The message type accepted by the worker
- */
- public static interface Worker<M> {
-
- /**
- * This method is evaluated when the worker process is spawned.
- */
- public void setup();
-
- /**
- * This method is evaluated when the worker receives a new message.
- *
- * @param message the received message
- */
- public void execute(final M message);
-
- /**
- * This method is evaluated when the worker process is destroyed.
- */
- public void terminate();
-
- }
-
- /**
- * The Master program is executed by the master process in the {@link GraphActors} system.
- * There are many workers and a single master.
- *
- * @param <M> The message type accepted by the master
- */
- public static interface Master<M> {
- public void setup();
-
- public void execute(final M message);
-
- public void terminate();
-
- }
-
-}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/ActorsResult.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/ActorsResult.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/ActorsResult.java
deleted file mode 100644
index c9db36a..0000000
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/ActorsResult.java
+++ /dev/null
@@ -1,30 +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;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-public interface ActorsResult<R> {
-
- public R getResult();
-
- public void setResult(final R result);
-}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/Address.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/Address.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/Address.java
deleted file mode 100644
index 9f59e5e..0000000
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/Address.java
+++ /dev/null
@@ -1,76 +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.io.Serializable;
-import java.net.InetAddress;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-public abstract class Address implements Serializable {
-
- private final String id;
- private final InetAddress location;
-
- public Address(final String id, final InetAddress location) {
- this.id = id;
- this.location = location;
- }
-
- public InetAddress getLocation() {
- return this.location;
- }
-
- public String getId() {
- return this.id;
- }
-
- @Override
- public boolean equals(final Object other) {
- return other instanceof Address && ((Address) other).id.equals(this.id);
- }
-
- @Override
- public int hashCode() {
- return this.id.hashCode();
- }
-
- @Override
- public String toString() {
- return this.id;
- }
-
- public static final class Master extends Address {
-
- public Master(final String id, final InetAddress location) {
- super(id, location);
- }
-
- }
-
- public static final class Worker extends Address {
-
- public Worker(final String id, final InetAddress location) {
- super(id, location);
- }
-
- }
-}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/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
deleted file mode 100644
index 51f4c4a..0000000
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/GraphActors.java
+++ /dev/null
@@ -1,98 +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 org.apache.commons.configuration.BaseConfiguration;
-import org.apache.commons.configuration.Configuration;
-import org.apache.tinkerpop.gremlin.process.Processor;
-import org.apache.tinkerpop.gremlin.process.actor.traversal.strategy.decoration.ActorProgramStrategy;
-import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
-import org.apache.tinkerpop.gremlin.process.traversal.strategy.ProcessorTraversalStrategy;
-import org.apache.tinkerpop.gremlin.structure.Graph;
-
-import java.util.concurrent.Future;
-
-/**
- * GraphActors is a message-passing based graph {@link Processor} that is:
- * asynchronous, distributed, and partition centric.
- *
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-public interface GraphActors<R> extends Processor {
-
- public static final String GRAPH_ACTORS = "gremlin.graphActors";
- public static final String GRAPH_ACTORS_WORKERS = "gremlin.graphActors.workers";
-
- /**
- * Provide the {@link ActorProgram} that the GraphActors will execute.
- *
- * @param program the program to execute
- * @return the updated GraphActors with newly defined program
- */
- public GraphActors<R> program(final ActorProgram<R> program);
-
- /**
- * Specify the number of workers per {@link Graph} {@link org.apache.tinkerpop.gremlin.structure.Partition}.
- *
- * @param workers the number of workers per partition
- * @return the updated GraphActors with newly defined workers
- */
- public GraphActors<R> workers(final int workers);
-
- /**
- * Add an arbitrary configuration to the GraphActors system.
- * Typically, these configurations are provider-specific and do not generalize across all GraphActor implementations.
- *
- * @param key the key of the configuration
- * @param value the value of the configuration
- * @return the updated GraphActors with newly defined configuration
- */
- public GraphActors<R> configure(final String key, final Object value);
-
- /**
- * Submit the {@link ActorProgram} for execution by the {@link GraphActors}.
- *
- * @return a {@link Future} denoting a reference to the asynchronous computation's result
- */
- public Future<R> submit(final Graph graph);
-
- /**
- * Returns an {@link ActorProgramStrategy} which enables a {@link Traversal} to execute on {@link GraphActors}.
- *
- * @return a traversal strategy capable of executing traversals on a GraphActors
- */
- public default ProcessorTraversalStrategy<GraphActors> getProcessorTraversalStrategy() {
- return new ActorProgramStrategy(this);
- }
-
- public static <A extends GraphActors> A open(final Configuration configuration) {
- try {
- return (A) Class.forName(configuration.getString(GRAPH_ACTORS)).getMethod("open", Configuration.class).invoke(null, configuration);
- } catch (final Exception e) {
- throw new IllegalArgumentException(e.getMessage(), e);
- }
- }
-
- public static <A extends GraphActors> A open(final Class<A> graphActorsClass) {
- final BaseConfiguration configuration = new BaseConfiguration();
- configuration.setProperty(GRAPH_ACTORS, graphActorsClass.getCanonicalName());
- return GraphActors.open(configuration);
- }
-}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/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
deleted file mode 100644
index b584322..0000000
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/TraversalActorProgram.java
+++ /dev/null
@@ -1,111 +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;
-
-import org.apache.tinkerpop.gremlin.process.actor.Actor;
-import org.apache.tinkerpop.gremlin.process.actor.ActorProgram;
-import org.apache.tinkerpop.gremlin.process.actor.traversal.message.BarrierAddMessage;
-import org.apache.tinkerpop.gremlin.process.actor.traversal.message.BarrierDoneMessage;
-import org.apache.tinkerpop.gremlin.process.actor.traversal.message.SideEffectAddMessage;
-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.strategy.decoration.ActorProgramStrategy;
-import org.apache.tinkerpop.gremlin.process.actor.traversal.strategy.verification.ActorVerificationStrategy;
-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.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.traverser.util.TraverserSet;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Optional;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-public final class TraversalActorProgram<R> implements ActorProgram<TraverserSet<R>> {
-
- private static final List<Class> MESSAGE_PRIORITIES = Arrays.asList(
- StartMessage.class,
- Traverser.class,
- SideEffectAddMessage.class,
- BarrierAddMessage.class,
- SideEffectSetMessage.class,
- BarrierDoneMessage.class,
- Terminate.class);
-
- private Traversal.Admin<?, R> traversal;
- public TraverserSet<R> result = new TraverserSet<>();
-
- public TraversalActorProgram(final Traversal.Admin<?, R> traversal) {
- 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
- public Worker createWorkerProgram(final Actor.Worker worker) {
- return new TraversalWorkerProgram<>(worker, this.traversal.clone());
- }
-
- @Override
- public Master createMasterProgram(final Actor.Master master) {
- return new TraversalMasterProgram<>(master, this.traversal.clone(), this.result);
- }
-
- @Override
- public Optional<List<Class>> getMessagePriorities() {
- return Optional.of(MESSAGE_PRIORITIES);
- }
-
- @Override
- public ActorProgram<TraverserSet<R>> clone() {
- try {
- final TraversalActorProgram<R> clone = (TraversalActorProgram<R>) super.clone();
- clone.traversal = this.traversal.clone();
- return clone;
- } catch (final CloneNotSupportedException e) {
- throw new IllegalStateException(e.getMessage(), e);
- }
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/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
deleted file mode 100644
index 2aaf686..0000000
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/TraversalMasterProgram.java
+++ /dev/null
@@ -1,179 +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;
-
-import org.apache.tinkerpop.gremlin.process.actor.Actor;
-import org.apache.tinkerpop.gremlin.process.actor.ActorProgram;
-import org.apache.tinkerpop.gremlin.process.actor.Address;
-import org.apache.tinkerpop.gremlin.process.actor.traversal.message.BarrierAddMessage;
-import org.apache.tinkerpop.gremlin.process.actor.traversal.message.BarrierDoneMessage;
-import org.apache.tinkerpop.gremlin.process.actor.traversal.message.SideEffectAddMessage;
-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.traversal.Step;
-import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
-import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
-import org.apache.tinkerpop.gremlin.process.traversal.step.Barrier;
-import org.apache.tinkerpop.gremlin.process.traversal.step.Distributing;
-import org.apache.tinkerpop.gremlin.process.traversal.step.LocalBarrier;
-import org.apache.tinkerpop.gremlin.process.traversal.step.Pushing;
-import org.apache.tinkerpop.gremlin.process.traversal.step.SideEffectCapable;
-import org.apache.tinkerpop.gremlin.process.traversal.step.filter.RangeGlobalStep;
-import org.apache.tinkerpop.gremlin.process.traversal.step.filter.TailGlobalStep;
-import org.apache.tinkerpop.gremlin.process.traversal.step.map.OrderGlobalStep;
-import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.OrderedTraverser;
-import org.apache.tinkerpop.gremlin.process.traversal.traverser.util.TraverserSet;
-import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalMatrix;
-import org.apache.tinkerpop.gremlin.structure.Element;
-import org.apache.tinkerpop.gremlin.structure.Partition;
-
-import java.util.HashMap;
-import java.util.Map;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-final class TraversalMasterProgram<M> implements ActorProgram.Master<M> {
-
- private final Actor.Master master;
- private final Traversal.Admin<?, ?> traversal;
- private final TraversalMatrix<?, ?> matrix;
- private Map<String, Barrier> barriers = new HashMap<>();
- private final TraverserSet<?> results;
- private Address.Worker leaderWorker;
- private int orderCounter = -1;
- private final Map<Partition, Address.Worker> partitionToWorkerMap = new HashMap<>();
-
- public TraversalMasterProgram(final Actor.Master master, final Traversal.Admin<?, ?> traversal, final TraverserSet<?> results) {
- this.traversal = traversal;
- // System.out.println("master[created]: " + master.address().getId());
- // System.out.println(this.traversal);
- this.matrix = new TraversalMatrix<>(this.traversal);
- this.results = results;
- this.master = master;
- Distributing.configure(this.traversal, true, true);
- Pushing.configure(this.traversal, true, false);
- }
-
- @Override
- public void setup() {
- this.leaderWorker = this.master.workers().get(0);
- for (int i = 0; i < this.master.partitioner().getPartitions().size(); i++) {
- this.partitionToWorkerMap.put(this.master.partitioner().getPartitions().get(i), this.master.workers().get(i));
- }
- this.broadcast(StartMessage.instance());
- this.master.send(this.leaderWorker, Terminate.MAYBE);
- }
-
- @Override
- public void execute(final M message) {
- if (message instanceof Traverser.Admin) {
- this.processTraverser((Traverser.Admin) message);
- } else if (message instanceof BarrierAddMessage) {
- final Barrier barrier = (Barrier) this.matrix.getStepById(((BarrierAddMessage) message).getStepId());
- final Step<?, ?> step = (Step) barrier;
- barrier.addBarrier(((BarrierAddMessage) message).getBarrier());
- 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 Terminate) {
- assert Terminate.YES == message;
- if (!this.barriers.isEmpty()) {
- for (final Barrier barrier : this.barriers.values()) {
- final Step<?, ?> step = (Step) barrier;
- if (!(barrier instanceof LocalBarrier)) {
- this.orderBarrier(step);
- if (step instanceof OrderGlobalStep) this.orderCounter = 0;
- while (step.hasNext()) {
- this.sendTraverser(-1 == this.orderCounter ?
- step.next() :
- new OrderedTraverser<>(step.next(), this.orderCounter++));
- }
- } else {
- if (step instanceof SideEffectCapable) {
- final String key = ((SideEffectCapable) step).getSideEffectKey();
- this.broadcast(new SideEffectSetMessage(key, this.traversal.getSideEffects().get(key)));
- }
- this.broadcast(new BarrierDoneMessage(barrier));
- barrier.done();
- }
- }
- this.barriers.clear();
- this.master.send(this.leaderWorker, Terminate.MAYBE);
- } else {
- while (this.traversal.hasNext()) {
- this.results.add((Traverser.Admin) this.traversal.nextTraverser());
- }
- if (this.orderCounter != -1)
- this.results.sort((a, b) -> Integer.compare(((OrderedTraverser<?>) a).order(), ((OrderedTraverser<?>) b).order()));
-
- this.master.close();
- }
- } else {
- throw new IllegalStateException("Unknown message:" + message);
- }
- }
-
- @Override
- public void terminate() {
- this.master.result().setResult(this.results);
- }
-
- private void broadcast(final Object message) {
- for (final Address.Worker worker : this.master.workers()) {
- this.master.send(worker, message);
- }
- }
-
- private void processTraverser(final Traverser.Admin traverser) {
- if (traverser.isHalted() || traverser.get() instanceof Element) {
- this.sendTraverser(traverser);
- } else {
- final Step<?, ?> step = this.matrix.<Object, Object, Step<Object, Object>>getStepById(traverser.getStepId());
- step.addStart(traverser);
- if (step instanceof Barrier) {
- this.barriers.put(step.getId(), (Barrier) step);
- } else {
- while (step.hasNext()) {
- this.processTraverser(step.next());
- }
- }
- }
- }
-
- private void sendTraverser(final Traverser.Admin traverser) {
- if (traverser.isHalted())
- this.results.add(traverser);
- else if (traverser.get() instanceof Element)
- this.master.send(this.partitionToWorkerMap.get(this.master.partitioner().getPartition((Element) traverser.get())), traverser);
- else
- this.master.send(this.master.address(), traverser);
- }
-
- private void orderBarrier(final Step step) {
- if (this.orderCounter != -1 && step instanceof Barrier && (step instanceof RangeGlobalStep || step instanceof TailGlobalStep)) {
- final Barrier barrier = (Barrier) step;
- final TraverserSet<?> rangingBarrier = (TraverserSet<?>) barrier.nextBarrier();
- rangingBarrier.sort((a, b) -> Integer.compare(((OrderedTraverser<?>) a).order(), ((OrderedTraverser<?>) b).order()));
- barrier.addBarrier(rangingBarrier);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/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
deleted file mode 100644
index 001219a..0000000
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/TraversalWorkerProgram.java
+++ /dev/null
@@ -1,170 +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;
-
-import org.apache.tinkerpop.gremlin.process.actor.Actor;
-import org.apache.tinkerpop.gremlin.process.actor.ActorProgram;
-import org.apache.tinkerpop.gremlin.process.actor.Address;
-import org.apache.tinkerpop.gremlin.process.actor.traversal.message.BarrierAddMessage;
-import org.apache.tinkerpop.gremlin.process.actor.traversal.message.BarrierDoneMessage;
-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.traversal.Step;
-import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
-import org.apache.tinkerpop.gremlin.process.traversal.Traverser;
-import org.apache.tinkerpop.gremlin.process.traversal.step.Barrier;
-import org.apache.tinkerpop.gremlin.process.traversal.step.Distributing;
-import org.apache.tinkerpop.gremlin.process.traversal.step.Pushing;
-import org.apache.tinkerpop.gremlin.process.traversal.step.map.GraphStep;
-import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalHelper;
-import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalMatrix;
-import org.apache.tinkerpop.gremlin.structure.Edge;
-import org.apache.tinkerpop.gremlin.structure.Element;
-import org.apache.tinkerpop.gremlin.structure.Partition;
-import org.apache.tinkerpop.gremlin.structure.Vertex;
-import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
-
-import java.util.HashMap;
-import java.util.Map;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-final class TraversalWorkerProgram<M> implements ActorProgram.Worker<M> {
-
- private final Actor.Worker self;
- private final TraversalMatrix<?, ?> matrix;
- private final Map<Partition, Address.Worker> partitionToWorkerMap = new HashMap<>();
- //
- private Address.Worker neighborWorker;
- private boolean isLeader;
- private Terminate terminate = null;
- private boolean voteToHalt = false;
- private Map<String, Barrier> barriers = new HashMap<>();
-
- public TraversalWorkerProgram(final Actor.Worker self, final Traversal.Admin<?, ?> traversal) {
- this.self = self;
- // System.out.println("worker[created]: " + this.self.address().getId());
- // set up partition and traversal information
- final WorkerTraversalSideEffects sideEffects = new WorkerTraversalSideEffects(traversal.getSideEffects(), this.self);
- TraversalHelper.applyTraversalRecursively(t -> t.setSideEffects(sideEffects), traversal);
- this.matrix = new TraversalMatrix<>(traversal);
- Distributing.configure(traversal, false, true);
- Pushing.configure(traversal, true, false);
- //////
- final GraphStep graphStep = (GraphStep) traversal.getStartStep();
- if (0 == graphStep.getIds().length)
- ((GraphStep) traversal.getStartStep()).setIteratorSupplier(graphStep.returnsVertex() ? this.self.partition()::vertices : this.self.partition()::edges);
- else {
- if (graphStep.returnsVertex())
- ((GraphStep<Vertex, Vertex>) traversal.getStartStep()).setIteratorSupplier(
- () -> IteratorUtils.filter(self.partition().vertices(graphStep.getIds()), this.self.partition()::contains));
- else
- ((GraphStep<Edge, Edge>) traversal.getStartStep()).setIteratorSupplier(
- () -> IteratorUtils.filter(self.partition().edges(graphStep.getIds()), this.self.partition()::contains));
- }
- }
-
- @Override
- public void setup() {
- // create termination ring topology
- final int i = this.self.workers().indexOf(this.self.address());
- this.neighborWorker = this.self.workers().get(i == this.self.workers().size() - 1 ? 0 : i + 1);
- this.isLeader = i == 0;
- for (int j = 0; j < this.self.partitioner().getPartitions().size(); j++) {
- this.partitionToWorkerMap.put(this.self.partitioner().getPartitions().get(j), this.self.workers().get(j));
- }
- }
-
- @Override
- public void execute(final M message) {
- //System.out.println(message + "::" + this.isLeader);
- if (message instanceof StartMessage) {
- // initial message from master that says: "start processing"
- final GraphStep step = (GraphStep) this.matrix.getTraversal().getStartStep();
- while (step.hasNext()) {
- this.sendTraverser(step.next());
- }
- } else if (message instanceof Traverser.Admin) {
- 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 null == this.terminate;
- this.terminate = (Terminate) message;
- if (!this.barriers.isEmpty()) {
- for (final Barrier barrier : this.barriers.values()) {
- while (barrier.hasNextBarrier()) {
- this.self.send(this.self.master(), new BarrierAddMessage(barrier));
- }
- }
- this.barriers.clear();
- }
- // use termination token to determine termination condition
- 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);
- }
- }
-
- @Override
- public void terminate() {
-
- }
-
- //////////////
-
- private void processTraverser(final Traverser.Admin traverser) {
- assert !(traverser.get() instanceof Element) || !traverser.isHalted() || this.self.partition().contains((Element) traverser.get());
- final Step<?, ?> step = this.matrix.<Object, Object, Step<Object, Object>>getStepById(traverser.getStepId());
- step.addStart(traverser);
- if (step instanceof Barrier) {
- this.barriers.put(step.getId(), (Barrier) step);
- } else {
- while (step.hasNext()) {
- this.sendTraverser(step.next());
- }
- }
- }
-
- private void sendTraverser(final Traverser.Admin traverser) {
- this.voteToHalt = false;
- if (traverser.isHalted())
- this.self.send(this.self.master(), traverser);
- else if (traverser.get() instanceof Element && !this.self.partition().contains((Element) traverser.get()))
- this.self.send(this.partitionToWorkerMap.get(this.self.partitioner().getPartition((Element) traverser.get())), traverser);
- else
- this.self.send(this.self.address(), traverser);
- }
-}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/WorkerTraversalSideEffects.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/WorkerTraversalSideEffects.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/WorkerTraversalSideEffects.java
deleted file mode 100644
index 0950435..0000000
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/WorkerTraversalSideEffects.java
+++ /dev/null
@@ -1,148 +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;
-
-import org.apache.tinkerpop.gremlin.process.actor.Actor;
-import org.apache.tinkerpop.gremlin.process.actor.traversal.message.SideEffectAddMessage;
-import org.apache.tinkerpop.gremlin.process.traversal.TraversalSideEffects;
-
-import java.util.Optional;
-import java.util.Set;
-import java.util.function.BinaryOperator;
-import java.util.function.Supplier;
-import java.util.function.UnaryOperator;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-public final class WorkerTraversalSideEffects implements TraversalSideEffects {
-
- private TraversalSideEffects sideEffects;
- private Actor.Worker worker;
-
-
- private WorkerTraversalSideEffects() {
- // for serialization
- }
-
- public WorkerTraversalSideEffects(final TraversalSideEffects sideEffects, final Actor.Worker worker) {
- this.sideEffects = sideEffects;
- this.worker = worker;
- }
-
- public TraversalSideEffects getSideEffects() {
- return this.sideEffects;
- }
-
- @Override
- public void set(final String key, final Object value) {
- this.sideEffects.set(key, value);
- }
-
- @Override
- public <V> V get(final String key) throws IllegalArgumentException {
- return this.sideEffects.get(key);
- }
-
- @Override
- public void remove(final String key) {
- this.sideEffects.remove(key);
- }
-
- @Override
- public Set<String> keys() {
- return this.sideEffects.keys();
- }
-
- @Override
- public void add(final String key, final Object value) {
- this.sideEffects.add(key, value);
- this.worker.send(this.worker.master(), new SideEffectAddMessage(key, value));
- }
-
- @Override
- public <V> void register(final String key, final Supplier<V> initialValue, final BinaryOperator<V> reducer) {
- this.sideEffects.register(key, initialValue, reducer);
- }
-
- @Override
- public <V> void registerIfAbsent(final String key, final Supplier<V> initialValue, final BinaryOperator<V> reducer) {
- this.sideEffects.registerIfAbsent(key, initialValue, reducer);
- }
-
- @Override
- public <V> BinaryOperator<V> getReducer(final String key) {
- return this.sideEffects.getReducer(key);
- }
-
- @Override
- public <V> Supplier<V> getSupplier(final String key) {
- return this.sideEffects.getSupplier(key);
- }
-
- @Override
- @Deprecated
- public void registerSupplier(final String key, final Supplier supplier) {
- this.sideEffects.registerSupplier(key, supplier);
- }
-
- @Override
- @Deprecated
- public <V> Optional<Supplier<V>> getRegisteredSupplier(final String key) {
- return this.sideEffects.getRegisteredSupplier(key);
- }
-
- @Override
- public <S> void setSack(final Supplier<S> initialValue, final UnaryOperator<S> splitOperator, final BinaryOperator<S> mergeOperator) {
- this.sideEffects.setSack(initialValue, splitOperator, mergeOperator);
- }
-
- @Override
- public <S> Supplier<S> getSackInitialValue() {
- return this.sideEffects.getSackInitialValue();
- }
-
- @Override
- public <S> UnaryOperator<S> getSackSplitter() {
- return this.sideEffects.getSackSplitter();
- }
-
- @Override
- public <S> BinaryOperator<S> getSackMerger() {
- return this.sideEffects.getSackMerger();
- }
-
- @Override
- public TraversalSideEffects clone() {
- try {
- final WorkerTraversalSideEffects clone = (WorkerTraversalSideEffects) super.clone();
- clone.sideEffects = this.sideEffects.clone();
- return clone;
- } catch (final CloneNotSupportedException e) {
- throw new IllegalStateException(e.getMessage(), e);
- }
- }
-
- @Override
- public void mergeInto(final TraversalSideEffects sideEffects) {
- this.sideEffects.mergeInto(sideEffects);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/message/BarrierAddMessage.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/message/BarrierAddMessage.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/message/BarrierAddMessage.java
deleted file mode 100644
index dba9b86..0000000
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/message/BarrierAddMessage.java
+++ /dev/null
@@ -1,47 +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;
-
-import org.apache.tinkerpop.gremlin.process.traversal.Step;
-import org.apache.tinkerpop.gremlin.process.traversal.step.Barrier;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-public final class BarrierAddMessage {
-
- private final Object barrier;
- private final String stepId;
-
- public BarrierAddMessage(final Barrier barrier) {
- this.barrier = barrier.nextBarrier();
- this.stepId = ((Step) barrier).getId();
- }
-
- public Object getBarrier() {
- return this.barrier;
- }
-
- public String getStepId() {
- return this.stepId;
- }
-
-
-}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/message/BarrierDoneMessage.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/message/BarrierDoneMessage.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/message/BarrierDoneMessage.java
deleted file mode 100644
index 837a55f..0000000
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/message/BarrierDoneMessage.java
+++ /dev/null
@@ -1,41 +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;
-
-import org.apache.tinkerpop.gremlin.process.traversal.Step;
-import org.apache.tinkerpop.gremlin.process.traversal.step.Barrier;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-public final class BarrierDoneMessage {
-
- private final String stepId;
-
- public BarrierDoneMessage(final Barrier barrier) {
- this.stepId = ((Step) barrier).getId();
-
- }
-
- public String getStepId() {
- return this.stepId;
- }
-}
-
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/message/SideEffectAddMessage.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/message/SideEffectAddMessage.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/message/SideEffectAddMessage.java
deleted file mode 100644
index 511c125..0000000
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/message/SideEffectAddMessage.java
+++ /dev/null
@@ -1,43 +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 SideEffectAddMessage {
-
- private final String key;
- private final Object value;
-
- public SideEffectAddMessage(final String key, final Object value) {
- this.value = value;
- this.key = key;
- }
-
- public String getKey() {
- return this.key;
- }
-
- public Object getValue() {
- return this.value;
- }
-}
-
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/message/SideEffectSetMessage.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/message/SideEffectSetMessage.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/message/SideEffectSetMessage.java
deleted file mode 100644
index 31f83c2..0000000
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/message/SideEffectSetMessage.java
+++ /dev/null
@@ -1,42 +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 SideEffectSetMessage {
-
- private final String key;
- private final Object value;
-
- public SideEffectSetMessage(final String key, final Object value) {
- this.key = key;
- this.value = value;
- }
-
- public String getKey() {
- return this.key;
- }
-
- public Object getValue() {
- return this.value;
- }
-}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/message/StartMessage.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/message/StartMessage.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/message/StartMessage.java
deleted file mode 100644
index 1b4292e..0000000
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/message/StartMessage.java
+++ /dev/null
@@ -1,35 +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 StartMessage {
-
- private static final StartMessage INSTANCE = new StartMessage();
-
- private StartMessage() {
- }
-
- public static StartMessage instance() {
- return INSTANCE;
- }
-}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/message/Terminate.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/message/Terminate.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/message/Terminate.java
deleted file mode 100644
index 4ab789d..0000000
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/message/Terminate.java
+++ /dev/null
@@ -1,28 +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 enum Terminate {
-
- MAYBE, YES, NO
-}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/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
deleted file mode 100644
index e4520aa..0000000
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/step/map/TraversalActorProgramStep.java
+++ /dev/null
@@ -1,73 +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;
-
-import org.apache.commons.configuration.Configuration;
-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.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 Traversal.Admin<S, E> actorsTraversal;
- private final Configuration graphActorsConfiguration;
- private boolean first = true;
-
- public TraversalActorProgramStep(final Traversal.Admin<?, ?> traversal, final Configuration graphActorsConfiguration) {
- super(traversal);
- this.graphActorsConfiguration = graphActorsConfiguration;
- this.actorsTraversal = (Traversal.Admin) traversal.clone();
- this.actorsTraversal.setParent(EmptyStep.instance());
- }
-
- @Override
- protected Traverser.Admin<E> processNextStart() throws NoSuchElementException {
- if (this.first) {
- this.first = false;
- try {
- final GraphActors<TraverserSet<E>> graphActors = GraphActors.open(this.graphActorsConfiguration);
- final ActorProgram<TraverserSet<E>> actorProgram = new TraversalActorProgram<>(this.actorsTraversal);
- graphActors.program(actorProgram).submit(this.getTraversal().getGraph().get()).get().forEach(this.starts::add);
- } catch (final Exception e) {
- throw new IllegalStateException(e.getMessage(), e);
- }
- }
- return this.starts.next();
- }
-
- @Override
- public String toString() {
- return StringFactory.stepString(this, this.actorsTraversal);
- }
-
-}
-
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/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
deleted file mode 100644
index 7e713de..0000000
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/strategy/decoration/ActorProgramStrategy.java
+++ /dev/null
@@ -1,94 +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.strategy.decoration;
-
-import org.apache.commons.configuration.Configuration;
-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.remote.traversal.strategy.decoration.RemoteStrategy;
-import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
-import org.apache.tinkerpop.gremlin.process.traversal.TraversalStrategy;
-import org.apache.tinkerpop.gremlin.process.traversal.step.util.EmptyStep;
-import org.apache.tinkerpop.gremlin.process.traversal.strategy.AbstractTraversalStrategy;
-import org.apache.tinkerpop.gremlin.process.traversal.strategy.ProcessorTraversalStrategy;
-import org.apache.tinkerpop.gremlin.process.traversal.strategy.verification.ReadOnlyStrategy;
-import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalHelper;
-
-import java.util.Collections;
-import java.util.Set;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-public final class ActorProgramStrategy extends AbstractTraversalStrategy<TraversalStrategy.DecorationStrategy>
- implements TraversalStrategy.DecorationStrategy, ProcessorTraversalStrategy<GraphActors> {
-
- private static final Set<Class<? extends DecorationStrategy>> PRIORS = Collections.singleton(RemoteStrategy.class);
-
- private final Configuration graphActorsConfiguration;
-
- public ActorProgramStrategy(final GraphActors graphActors) {
- this.graphActorsConfiguration = graphActors.configuration();
- }
-
- @Override
- public void apply(final Traversal.Admin<?, ?> traversal) {
- ReadOnlyStrategy.instance().apply(traversal);
-
- if (!(traversal.getParent() instanceof EmptyStep))
- return;
-
- final TraversalActorProgramStep<?, ?> actorStep = new TraversalActorProgramStep<>(traversal, this.graphActorsConfiguration);
- TraversalHelper.removeAllSteps(traversal);
- traversal.addStep(actorStep);
-
- // validations
- assert traversal.getStartStep().equals(actorStep);
- assert traversal.getSteps().size() == 1;
- assert traversal.getEndStep() == actorStep;
- }
-
- @Override
- public Set<Class<? extends DecorationStrategy>> applyPrior() {
- return PRIORS;
- }
-
- ////////////////////////////////////////////////////////////
-
- @Override
- public Configuration getConfiguration() {
- return this.graphActorsConfiguration;
- }
-
- public static ActorProgramStrategy create(final Configuration configuration) {
- try {
- return new ActorProgramStrategy(GraphActors.open(configuration));
- } catch (final Exception e) {
- throw new IllegalArgumentException(e.getMessage(), e);
- }
- }
-
- @Override
- public GraphActors getProcessor() {
- return GraphActors.open(this.graphActorsConfiguration);
- }
-
-}
-
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/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
deleted file mode 100644
index f6e93ef..0000000
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/traversal/strategy/verification/ActorVerificationStrategy.java
+++ /dev/null
@@ -1,55 +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.strategy.verification;
-
-import org.apache.tinkerpop.gremlin.process.traversal.Step;
-import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
-import org.apache.tinkerpop.gremlin.process.traversal.TraversalStrategy;
-import org.apache.tinkerpop.gremlin.process.traversal.step.GraphComputing;
-import org.apache.tinkerpop.gremlin.process.traversal.step.sideEffect.InjectStep;
-import org.apache.tinkerpop.gremlin.process.traversal.strategy.AbstractTraversalStrategy;
-import org.apache.tinkerpop.gremlin.process.traversal.strategy.verification.ReadOnlyStrategy;
-import org.apache.tinkerpop.gremlin.process.traversal.strategy.verification.VerificationException;
-import org.apache.tinkerpop.gremlin.process.traversal.util.TraversalHelper;
-import org.apache.tinkerpop.gremlin.structure.Graph;
-
-import java.util.Collections;
-import java.util.Set;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-public final class ActorVerificationStrategy extends AbstractTraversalStrategy<TraversalStrategy.VerificationStrategy> implements TraversalStrategy.VerificationStrategy {
-
- private static final ActorVerificationStrategy INSTANCE = new ActorVerificationStrategy();
-
- private ActorVerificationStrategy() {
- }
-
- @Override
- public void apply(final Traversal.Admin<?, ?> traversal) {
- if (!TraversalHelper.getStepsOfAssignableClass(InjectStep.class, traversal).isEmpty())
- throw new VerificationException("Inject traversal currently not supported", traversal);
- }
-
- public static ActorVerificationStrategy instance() {
- return INSTANCE;
- }
-}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/util/DefaultActorsResult.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/util/DefaultActorsResult.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/util/DefaultActorsResult.java
deleted file mode 100644
index c650ba1..0000000
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/util/DefaultActorsResult.java
+++ /dev/null
@@ -1,42 +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.util;
-
-import org.apache.tinkerpop.gremlin.process.actor.ActorsResult;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-public final class DefaultActorsResult<R> implements ActorsResult<R> {
-
- private R result;
-
- public DefaultActorsResult() {
-
- }
-
- public R getResult() {
- return this.result;
- }
-
- public void setResult(final R result) {
- this.result = result;
- }
-}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/util/GraphActorsHelper.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/util/GraphActorsHelper.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/util/GraphActorsHelper.java
deleted file mode 100644
index eebee17..0000000
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/util/GraphActorsHelper.java
+++ /dev/null
@@ -1,48 +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.util;
-
-import org.apache.commons.configuration.Configuration;
-import org.apache.tinkerpop.gremlin.process.actor.GraphActors;
-import org.apache.tinkerpop.gremlin.util.iterator.IteratorUtils;
-
-import java.util.Iterator;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-public final class GraphActorsHelper {
-
- private GraphActorsHelper() {
-
- }
-
- public static GraphActors configure(GraphActors actors, final Configuration configuration) {
- final Iterator<String> keys = IteratorUtils.asList(configuration.getKeys()).iterator();
- while (keys.hasNext()) {
- final String key = keys.next();
- if (key.equals(GraphActors.GRAPH_ACTORS_WORKERS))
- actors = actors.workers(configuration.getInt(GraphActors.GRAPH_ACTORS_WORKERS));
- else if (!key.equals(GraphActors.GRAPH_ACTORS))
- actors = actors.configure(key, configuration.getProperty(key));
- }
- return actors;
- }
-}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/Actor.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/Actor.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/Actor.java
new file mode 100644
index 0000000..0445968
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/Actor.java
@@ -0,0 +1,95 @@
+/*
+ * 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.actors;
+
+import org.apache.tinkerpop.gremlin.structure.Partition;
+import org.apache.tinkerpop.gremlin.structure.Partitioner;
+
+import java.util.List;
+
+/**
+ * An Actor represents an isolated processing unit that can only be interacted with via messages.
+ * Actors are able to send and receive messages. The {@link GraphActors} framework has two types of actors:
+ * {@link Master} and {@link Worker}. A master actors is not associated with a particular graph {@link Partition}.
+ * Instead, its role is to coordinate the workers and ultimately, yield the final result of the submitted
+ * {@link ActorProgram}.
+ *
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public interface Actor {
+
+ /**
+ * Get the {@link Partitioner} associated with the {@link GraphActors} system.
+ *
+ * @return the partitioner used to partition (logically and/or physically) the {@link org.apache.tinkerpop.gremlin.structure.Graph}
+ */
+ public Partitioner partitioner();
+
+ /**
+ * Get the {@link Address} of the actors.
+ *
+ * @return the actors's address
+ */
+ public Address address();
+
+ /**
+ * Get a list of the {@link Address} values of all the workers in {@link GraphActors} system.
+ *
+ * @return the worker's addresses
+ */
+ public List<Address.Worker> workers();
+
+ /**
+ * Send a message from this actors to another actors given their {@link Address}.
+ *
+ * @param toActor the actors to receive the messages
+ * @param message the message being sent
+ * @param <M> the message type
+ */
+ public <M> void send(final Address toActor, final M message);
+
+ public interface Master extends Actor {
+
+ public Address.Master address();
+
+ public void close();
+
+ public <R> ActorsResult<R> result();
+
+ }
+
+ public interface Worker extends Actor {
+
+ public Address.Worker address();
+
+ public Address.Master master();
+
+ /**
+ * Get the {@link Partition} associated with this worker.
+ * In principle, this is the subset of the {@link org.apache.tinkerpop.gremlin.structure.Graph} that
+ * the worker is "data-local" to.
+ *
+ * @return the worker's partition
+ */
+ public Partition partition();
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/ActorProgram.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/ActorProgram.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/ActorProgram.java
new file mode 100644
index 0000000..b1e3065
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/ActorProgram.java
@@ -0,0 +1,145 @@
+/*
+ * 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.actors;
+
+import org.apache.commons.configuration.Configuration;
+import org.apache.tinkerpop.gremlin.structure.Graph;
+
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public interface ActorProgram extends Cloneable {
+
+ public static final String ACTOR_PROGRAM = "gremlin.actorProgram";
+
+ /**
+ * When it is necessary to store the state of the ActorProgram, this method is called.
+ * This is typically required when the ActorProgram needs to be serialized to another machine.
+ * Note that what is stored is simply the instance/configuration state, not any processed data.
+ * The default implementation provided simply stores the ActorProgarm class name for reflective reconstruction.
+ * It is typically a good idea to ActorProgram.super.storeState().
+ *
+ * @param configuration the configuration to store the state of the ActorProgram in.
+ */
+ public default void storeState(final Configuration configuration) {
+ configuration.setProperty(ACTOR_PROGRAM, this.getClass().getCanonicalName());
+ }
+
+ /**
+ * When it is necessary to load the state of the ActorProgram, this method is called.
+ * This is typically required when the ActorProgram needs to be serialized to another machine.
+ * Note that what is loaded is simply the instance state, not any processed data.
+ *
+ * @param graph the graph that the ActorProgram will run against
+ * @param configuration the configuration to load the state of the ActorProgram from.
+ */
+ public default void loadState(final Graph graph, final Configuration configuration) {
+
+ }
+
+ /**
+ * Create the {@link org.apache.tinkerpop.gremlin.process.actors.Actor.Worker} program.
+ * This is typically used by {@link Worker} to spawn its program.
+ *
+ * @param worker the worker actors creating the worker program
+ * @return the worker program
+ */
+ public ActorProgram.Worker createWorkerProgram(final Actor.Worker worker);
+
+ /**
+ * Create the {@link org.apache.tinkerpop.gremlin.process.actors.Actor.Master} program.
+ * This is typically used by {@link Master} to spawn its program.
+ *
+ * @param master the master actors creating the master program
+ * @return the master program
+ */
+ public ActorProgram.Master createMasterProgram(final Actor.Master master);
+
+ /**
+ * Get the ordered list of message classes where order determines the priority
+ * of message reception by the respective {@link Actor}. For instance,
+ * if an {@link Actor} has a message of type {@code X} and a message of type {@code Y}
+ * in its message buffer, and {@code X} has a higher priority, it will be fetched
+ * first from the buffer. If no list is provided then its FIFO.
+ * The default implementation returns an {@link Optional#empty()}.
+ *
+ * @return the optional ordered list of message priorities.
+ */
+ public default Optional<List<Class>> getMessagePriorities() {
+ return Optional.empty();
+ }
+
+ /**
+ * When multiple workers on a single machine need ActorProgram instances, it is possible to use clone.
+ * This will provide a speedier way of generating instances, over the {@link ActorProgram#storeState} and {@link ActorProgram#loadState} model.
+ * The default implementation simply returns the object as it assumes that the ActorProgram instance is a stateless singleton.
+ *
+ * @return A clone of the VertexProgram object
+ */
+ @SuppressWarnings("CloneDoesntDeclareCloneNotSupportedException")
+ public ActorProgram clone();
+
+ /**
+ * The Worker program is executed by a worker process in the {@link GraphActors} system.
+ * There are many workers and a single master.
+ * All workers execute the same program.
+ *
+ * @param <M> The message type accepted by the worker
+ */
+ public static interface Worker<M> {
+
+ /**
+ * This method is evaluated when the worker process is spawned.
+ */
+ public void setup();
+
+ /**
+ * This method is evaluated when the worker receives a new message.
+ *
+ * @param message the received message
+ */
+ public void execute(final M message);
+
+ /**
+ * This method is evaluated when the worker process is destroyed.
+ */
+ public void terminate();
+
+ }
+
+ /**
+ * The Master program is executed by the master process in the {@link GraphActors} system.
+ * There are many workers and a single master.
+ *
+ * @param <M> The message type accepted by the master
+ */
+ public static interface Master<M> {
+ public void setup();
+
+ public void execute(final M message);
+
+ public void terminate();
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/ActorsResult.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/ActorsResult.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/ActorsResult.java
new file mode 100644
index 0000000..beb7ab9
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/ActorsResult.java
@@ -0,0 +1,30 @@
+/*
+ * 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.actors;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public interface ActorsResult<R> {
+
+ public R getResult();
+
+ public void setResult(final R result);
+}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/Address.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/Address.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/Address.java
new file mode 100644
index 0000000..894a961
--- /dev/null
+++ b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actors/Address.java
@@ -0,0 +1,76 @@
+/*
+ * 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.actors;
+
+import java.io.Serializable;
+import java.net.InetAddress;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public abstract class Address implements Serializable {
+
+ private final String id;
+ private final InetAddress location;
+
+ public Address(final String id, final InetAddress location) {
+ this.id = id;
+ this.location = location;
+ }
+
+ public InetAddress getLocation() {
+ return this.location;
+ }
+
+ public String getId() {
+ return this.id;
+ }
+
+ @Override
+ public boolean equals(final Object other) {
+ return other instanceof Address && ((Address) other).id.equals(this.id);
+ }
+
+ @Override
+ public int hashCode() {
+ return this.id.hashCode();
+ }
+
+ @Override
+ public String toString() {
+ return this.id;
+ }
+
+ public static final class Master extends Address {
+
+ public Master(final String id, final InetAddress location) {
+ super(id, location);
+ }
+
+ }
+
+ public static final class Worker extends Address {
+
+ public Worker(final String id, final InetAddress location) {
+ super(id, location);
+ }
+
+ }
+}
[3/3] tinkerpop git commit: refactored GraphActors packaging -- its
not actor/, but actors/. JavaDoc and various cleanups. Also,
about to NOT serialize a traversal but instead use Bytecode. Next push will do
this with TraveraslVertexProgram.
Posted by ok...@apache.org.
refactored GraphActors packaging -- its not actor/, but actors/. JavaDoc and various cleanups. Also, about to NOT serialize a traversal but instead use Bytecode. Next push will do this with TraveraslVertexProgram.
Project: http://git-wip-us.apache.org/repos/asf/tinkerpop/repo
Commit: http://git-wip-us.apache.org/repos/asf/tinkerpop/commit/7ee2d372
Tree: http://git-wip-us.apache.org/repos/asf/tinkerpop/tree/7ee2d372
Diff: http://git-wip-us.apache.org/repos/asf/tinkerpop/diff/7ee2d372
Branch: refs/heads/TINKERPOP-1564
Commit: 7ee2d37268e42c76f2e8c9d5c964a5c734e1b2c1
Parents: 0aa3076
Author: Marko A. Rodriguez <ok...@gmail.com>
Authored: Wed Jan 4 08:01:29 2017 -0700
Committer: Marko A. Rodriguez <ok...@gmail.com>
Committed: Wed Jan 4 08:01:29 2017 -0700
----------------------------------------------------------------------
.../gremlin/akka/jsr223/AkkaGremlinPlugin.java | 2 +-
.../akka/process/actor/ActorMailbox.java | 141 ---------------
.../akka/process/actor/AkkaGraphActors.java | 130 --------------
.../gremlin/akka/process/actor/MasterActor.java | 117 ------------
.../gremlin/akka/process/actor/WorkerActor.java | 112 ------------
.../akka/process/actors/ActorMailbox.java | 141 +++++++++++++++
.../akka/process/actors/AkkaGraphActors.java | 130 ++++++++++++++
.../akka/process/actors/MasterActor.java | 117 ++++++++++++
.../akka/process/actors/WorkerActor.java | 112 ++++++++++++
.../src/main/resources/application.conf | 6 +-
.../process/AkkaActorsProcessActorsTest.java | 34 ----
.../akka/process/AkkaActorsProvider.java | 158 ----------------
.../gremlin/akka/process/AkkaPlayTest.java | 89 ---------
.../actors/AkkaActorsProcessActorsTest.java | 33 ++++
.../akka/process/actors/AkkaActorsProvider.java | 158 ++++++++++++++++
.../akka/process/actors/AkkaPlayTest.java | 89 +++++++++
.../tinkerpop/gremlin/process/actor/Actor.java | 95 ----------
.../gremlin/process/actor/ActorProgram.java | 145 ---------------
.../gremlin/process/actor/ActorsResult.java | 30 ----
.../gremlin/process/actor/Address.java | 76 --------
.../gremlin/process/actor/GraphActors.java | 98 ----------
.../actor/traversal/TraversalActorProgram.java | 111 ------------
.../actor/traversal/TraversalMasterProgram.java | 179 -------------------
.../actor/traversal/TraversalWorkerProgram.java | 170 ------------------
.../traversal/WorkerTraversalSideEffects.java | 148 ---------------
.../traversal/message/BarrierAddMessage.java | 47 -----
.../traversal/message/BarrierDoneMessage.java | 41 -----
.../traversal/message/SideEffectAddMessage.java | 43 -----
.../traversal/message/SideEffectSetMessage.java | 42 -----
.../actor/traversal/message/StartMessage.java | 35 ----
.../actor/traversal/message/Terminate.java | 28 ---
.../step/map/TraversalActorProgramStep.java | 73 --------
.../decoration/ActorProgramStrategy.java | 94 ----------
.../verification/ActorVerificationStrategy.java | 55 ------
.../process/actor/util/DefaultActorsResult.java | 42 -----
.../process/actor/util/GraphActorsHelper.java | 48 -----
.../tinkerpop/gremlin/process/actors/Actor.java | 95 ++++++++++
.../gremlin/process/actors/ActorProgram.java | 145 +++++++++++++++
.../gremlin/process/actors/ActorsResult.java | 30 ++++
.../gremlin/process/actors/Address.java | 76 ++++++++
.../gremlin/process/actors/GraphActors.java | 107 +++++++++++
.../actors/traversal/TraversalActorProgram.java | 129 +++++++++++++
.../traversal/TraversalMasterProgram.java | 179 +++++++++++++++++++
.../traversal/TraversalWorkerProgram.java | 170 ++++++++++++++++++
.../traversal/WorkerTraversalSideEffects.java | 148 +++++++++++++++
.../traversal/message/BarrierAddMessage.java | 47 +++++
.../traversal/message/BarrierDoneMessage.java | 41 +++++
.../traversal/message/SideEffectAddMessage.java | 43 +++++
.../traversal/message/SideEffectSetMessage.java | 42 +++++
.../actors/traversal/message/StartMessage.java | 35 ++++
.../actors/traversal/message/Terminate.java | 28 +++
.../step/map/TraversalActorProgramStep.java | 73 ++++++++
.../decoration/ActorProgramStrategy.java | 94 ++++++++++
.../verification/ActorVerificationStrategy.java | 48 +++++
.../actors/util/DefaultActorsResult.java | 42 +++++
.../process/actors/util/GraphActorsHelper.java | 48 +++++
.../process/traversal/TraversalStrategies.java | 2 +-
.../gremlin/structure/util/StringFactory.java | 2 +-
.../apache/tinkerpop/gremlin/GraphManager.java | 2 +-
.../apache/tinkerpop/gremlin/GraphProvider.java | 2 +-
.../gremlin/process/actors/GraphActorsTest.java | 1 -
61 files changed, 2408 insertions(+), 2390 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/jsr223/AkkaGremlinPlugin.java
----------------------------------------------------------------------
diff --git a/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/jsr223/AkkaGremlinPlugin.java b/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/jsr223/AkkaGremlinPlugin.java
index 049c5b7..5c06bff 100644
--- a/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/jsr223/AkkaGremlinPlugin.java
+++ b/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/jsr223/AkkaGremlinPlugin.java
@@ -19,7 +19,7 @@
package org.apache.tinkerpop.gremlin.akka.jsr223;
-import org.apache.tinkerpop.gremlin.akka.process.actor.AkkaGraphActors;
+import org.apache.tinkerpop.gremlin.akka.process.actors.AkkaGraphActors;
import org.apache.tinkerpop.gremlin.jsr223.AbstractGremlinPlugin;
import org.apache.tinkerpop.gremlin.jsr223.DefaultImportCustomizer;
import org.apache.tinkerpop.gremlin.jsr223.ImportCustomizer;
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/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
deleted file mode 100644
index c8e5fde..0000000
--- a/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actor/ActorMailbox.java
+++ /dev/null
@@ -1,141 +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.LinkedList;
-import java.util.List;
-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<Class> messagePriorities;
- private final List<Queue> messages;
- private final Object MUTEX = new Object();
-
- public ActorMessageQueue(final List<Class> messagePriorities) {
- this.messagePriorities = messagePriorities;
- this.messages = new ArrayList<>(this.messagePriorities.size());
- for (final Class clazz : this.messagePriorities) {
- final Queue queue;
- if (Traverser.class.isAssignableFrom(clazz))
- queue = new TraverserSet<>();
- else
- queue = new LinkedList<>();
- this.messages.add(queue);
- }
- }
-
- public void enqueue(final ActorRef receiver, final Envelope handle) {
- synchronized (MUTEX) {
- final Object message = handle.message();
- for (int i = 0; i < this.messagePriorities.size(); i++) {
- final Class clazz = this.messagePriorities.get(i);
- if (clazz.isInstance(message)) {
- this.messages.get(i).offer(message instanceof Traverser ? message : handle);
- return;
- }
- }
- throw new IllegalArgumentException("The provided message type is not registered: " + handle.message().getClass());
- }
- }
-
- 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/7ee2d372/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
deleted file mode 100644
index bc692c0..0000000
--- a/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actor/AkkaGraphActors.java
+++ /dev/null
@@ -1,130 +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.commons.configuration.BaseConfiguration;
-import org.apache.commons.configuration.Configuration;
-import org.apache.commons.configuration.ConfigurationUtils;
-import org.apache.tinkerpop.gremlin.process.actor.ActorProgram;
-import org.apache.tinkerpop.gremlin.process.actor.ActorsResult;
-import org.apache.tinkerpop.gremlin.process.actor.Address;
-import org.apache.tinkerpop.gremlin.process.actor.GraphActors;
-import org.apache.tinkerpop.gremlin.process.actor.util.DefaultActorsResult;
-import org.apache.tinkerpop.gremlin.process.actor.util.GraphActorsHelper;
-import org.apache.tinkerpop.gremlin.structure.Graph;
-import org.apache.tinkerpop.gremlin.structure.Partitioner;
-import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
-import org.apache.tinkerpop.gremlin.structure.util.partitioner.HashPartitioner;
-
-import java.net.InetAddress;
-import java.net.UnknownHostException;
-import java.util.Collections;
-import java.util.UUID;
-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 ActorProgram<R> actorProgram;
- private int workers = 1;
- private Configuration configuration;
- private boolean executed = false;
-
- private AkkaGraphActors(final Configuration configuration) {
- this.configuration = new BaseConfiguration();
- ConfigurationUtils.copy(configuration, this.configuration);
- this.configuration.setProperty(GRAPH_ACTORS, AkkaGraphActors.class.getCanonicalName());
- GraphActorsHelper.configure(this, this.configuration);
- }
-
- @Override
- public String toString() {
- return StringFactory.graphActorsString(this);
- }
-
- @Override
- public GraphActors<R> program(final ActorProgram<R> actorProgram) {
- this.actorProgram = actorProgram;
- return this;
- }
-
- @Override
- public GraphActors<R> workers(final int workers) {
- this.workers = workers;
- this.configuration.setProperty(GRAPH_ACTORS_WORKERS, workers);
- return this;
- }
-
- @Override
- public GraphActors<R> configure(final String key, final Object value) {
- this.configuration.setProperty(key, value);
- return this;
- }
-
- @Override
- public Future<R> submit(final Graph graph) {
- if (this.executed)
- throw new IllegalStateException("Can not execute twice");
- this.executed = true;
- final Config config = ConfigFactory.defaultApplication().withValue("message-priorities",
- ConfigValueFactory.fromAnyRef(this.actorProgram.getMessagePriorities().
- orElse(Collections.singletonList(Object.class)).
- stream().
- map(Class::getCanonicalName).
- collect(Collectors.toList()).toString()));
- final ActorSystem system = ActorSystem.create("traversal-" + UUID.randomUUID(), config);
- final ActorsResult<R> result = new DefaultActorsResult<>();
- final Partitioner partitioner = this.workers == 1 ? graph.partitioner() : new HashPartitioner(graph.partitioner(), this.workers);
- try {
- new Address.Master(system.actorOf(Props.create(MasterActor.class, this.actorProgram, partitioner, result), "master").path().toString(), InetAddress.getLocalHost());
- } catch (final UnknownHostException e) {
- throw new IllegalStateException(e.getMessage(), e);
- }
- return CompletableFuture.supplyAsync(() -> {
- while (!system.isTerminated()) {
-
- }
- return result.getResult();
- });
- }
-
- @Override
- public Configuration configuration() {
- return this.configuration;
- }
-
- public static AkkaGraphActors open(final Configuration configuration) {
- return new AkkaGraphActors(configuration);
- }
-
- public static AkkaGraphActors open() {
- return new AkkaGraphActors(new BaseConfiguration());
- }
-}
-
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/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
deleted file mode 100644
index 0173a8f..0000000
--- a/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actor/MasterActor.java
+++ /dev/null
@@ -1,117 +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.AbstractActor;
-import akka.actor.ActorSelection;
-import akka.actor.Props;
-import akka.dispatch.RequiresMessageQueue;
-import akka.japi.pf.ReceiveBuilder;
-import org.apache.tinkerpop.gremlin.process.actor.Actor;
-import org.apache.tinkerpop.gremlin.process.actor.ActorProgram;
-import org.apache.tinkerpop.gremlin.process.actor.ActorsResult;
-import org.apache.tinkerpop.gremlin.process.actor.Address;
-import org.apache.tinkerpop.gremlin.structure.Partition;
-import org.apache.tinkerpop.gremlin.structure.Partitioner;
-
-import java.net.InetAddress;
-import java.net.UnknownHostException;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-public final class MasterActor extends AbstractActor implements RequiresMessageQueue<ActorMailbox.ActorSemantics>, Actor.Master {
-
- private final ActorProgram.Master masterProgram;
- private final Address.Master master;
- private final List<Address.Worker> workers;
- private final Map<Address, ActorSelection> actors = new HashMap<>();
- private final ActorsResult<?> result;
- private final Partitioner partitioner;
-
- public MasterActor(final ActorProgram program, final Partitioner partitioner, final ActorsResult<?> result) {
- this.partitioner = partitioner;
- this.result = result;
- try {
- this.master = new Address.Master(self().path().toString(), InetAddress.getLocalHost());
- } catch (final UnknownHostException e) {
- throw new IllegalStateException(e.getMessage(), e);
- }
- this.workers = new ArrayList<>();
- final List<Partition> partitions = partitioner.getPartitions();
- for (final Partition partition : partitions) {
- final String workerPathString = "worker-" + partition.id();
- this.workers.add(new Address.Worker(workerPathString, partition.location()));
- context().actorOf(Props.create(WorkerActor.class, program, this.master, partition, partitioner), workerPathString);
- }
- this.masterProgram = program.createMasterProgram(this);
- receive(ReceiveBuilder.matchAny(this.masterProgram::execute).build());
- }
-
- @Override
- public void preStart() {
- this.masterProgram.setup();
- }
-
- @Override
- public void postStop() {
- this.masterProgram.terminate();
- }
-
- @Override
- public <M> void send(final Address toActor, final M message) {
- ActorSelection actor = this.actors.get(toActor);
- if (null == actor) {
- actor = context().actorSelection(toActor.getId());
- this.actors.put(toActor, actor);
- }
- actor.tell(message, self());
- }
-
- @Override
- public List<Address.Worker> workers() {
- return this.workers;
- }
-
- @Override
- public Partitioner partitioner() {
- return this.partitioner;
- }
-
- @Override
- public Address.Master address() {
- return this.master;
- }
-
- @Override
- public void close() {
- context().system().terminate();
- }
-
- @Override
- public <R> ActorsResult<R> result() {
- return (ActorsResult<R>) this.result;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/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
deleted file mode 100644
index 27f942a..0000000
--- a/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actor/WorkerActor.java
+++ /dev/null
@@ -1,112 +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.AbstractActor;
-import akka.actor.ActorSelection;
-import akka.dispatch.RequiresMessageQueue;
-import akka.japi.pf.ReceiveBuilder;
-import org.apache.tinkerpop.gremlin.process.actor.Actor;
-import org.apache.tinkerpop.gremlin.process.actor.ActorProgram;
-import org.apache.tinkerpop.gremlin.process.actor.Address;
-import org.apache.tinkerpop.gremlin.structure.Partition;
-import org.apache.tinkerpop.gremlin.structure.Partitioner;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-public final class WorkerActor extends AbstractActor implements RequiresMessageQueue<ActorMailbox.ActorSemantics>, Actor.Worker {
-
- private final ActorProgram.Worker workerProgram;
- private final Partition localPartition;
- private final Partitioner partitioner;
- private final Address.Worker self;
- private final Address.Master master;
- private final List<Address.Worker> workers;
- private final Map<Address, ActorSelection> actors = new HashMap<>();
-
- public WorkerActor(final ActorProgram program, final Address.Master master, final Partition localPartition, final Partitioner partitioner) {
- this.localPartition = localPartition;
- this.partitioner = partitioner;
- this.self = new Address.Worker(this.createWorkerAddress(localPartition), localPartition.location());
- this.master = master;
- this.workers = new ArrayList<>();
- for (final Partition partition : partitioner.getPartitions()) {
- this.workers.add(new Address.Worker(this.createWorkerAddress(partition), partition.location()));
- }
- this.workerProgram = program.createWorkerProgram(this);
- receive(ReceiveBuilder.matchAny(this.workerProgram::execute).build());
- }
-
- @Override
- public void preStart() {
- this.workerProgram.setup();
- }
-
- @Override
- public void postStop() {
- this.workerProgram.terminate();
- }
-
- @Override
- public <M> void send(final Address toActor, final M message) {
- ActorSelection actor = this.actors.get(toActor);
- if (null == actor) {
- actor = context().actorSelection(toActor.getId());
- this.actors.put(toActor, actor);
- }
- actor.tell(message, self());
- }
-
- @Override
- public List<Address.Worker> workers() {
- return this.workers;
- }
-
- @Override
- public Partition partition() {
- return this.localPartition;
- }
-
- @Override
- public Partitioner partitioner() {
- return this.partitioner;
- }
-
- @Override
- public Address.Worker address() {
- return this.self;
- }
-
- @Override
- public Address.Master master() {
- return this.master;
- }
-
- private String createWorkerAddress(final Partition partition) {
- return "../worker-" + partition.id();
- }
-}
-
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actors/ActorMailbox.java
----------------------------------------------------------------------
diff --git a/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actors/ActorMailbox.java b/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actors/ActorMailbox.java
new file mode 100644
index 0000000..8087038
--- /dev/null
+++ b/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actors/ActorMailbox.java
@@ -0,0 +1,141 @@
+/*
+ * 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.actors;
+
+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.LinkedList;
+import java.util.List;
+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<Class> messagePriorities;
+ private final List<Queue> messages;
+ private final Object MUTEX = new Object();
+
+ public ActorMessageQueue(final List<Class> messagePriorities) {
+ this.messagePriorities = messagePriorities;
+ this.messages = new ArrayList<>(this.messagePriorities.size());
+ for (final Class clazz : this.messagePriorities) {
+ final Queue queue;
+ if (Traverser.class.isAssignableFrom(clazz))
+ queue = new TraverserSet<>();
+ else
+ queue = new LinkedList<>();
+ this.messages.add(queue);
+ }
+ }
+
+ public void enqueue(final ActorRef receiver, final Envelope handle) {
+ synchronized (MUTEX) {
+ final Object message = handle.message();
+ for (int i = 0; i < this.messagePriorities.size(); i++) {
+ final Class clazz = this.messagePriorities.get(i);
+ if (clazz.isInstance(message)) {
+ this.messages.get(i).offer(message instanceof Traverser ? message : handle);
+ return;
+ }
+ }
+ throw new IllegalArgumentException("The provided message type is not registered: " + handle.message().getClass());
+ }
+ }
+
+ 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/7ee2d372/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actors/AkkaGraphActors.java
----------------------------------------------------------------------
diff --git a/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actors/AkkaGraphActors.java b/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actors/AkkaGraphActors.java
new file mode 100644
index 0000000..3bd5fa6
--- /dev/null
+++ b/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actors/AkkaGraphActors.java
@@ -0,0 +1,130 @@
+/*
+ * 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.actors;
+
+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.commons.configuration.BaseConfiguration;
+import org.apache.commons.configuration.Configuration;
+import org.apache.commons.configuration.ConfigurationUtils;
+import org.apache.tinkerpop.gremlin.process.actors.ActorProgram;
+import org.apache.tinkerpop.gremlin.process.actors.ActorsResult;
+import org.apache.tinkerpop.gremlin.process.actors.Address;
+import org.apache.tinkerpop.gremlin.process.actors.GraphActors;
+import org.apache.tinkerpop.gremlin.process.actors.util.DefaultActorsResult;
+import org.apache.tinkerpop.gremlin.process.actors.util.GraphActorsHelper;
+import org.apache.tinkerpop.gremlin.structure.Graph;
+import org.apache.tinkerpop.gremlin.structure.Partitioner;
+import org.apache.tinkerpop.gremlin.structure.util.StringFactory;
+import org.apache.tinkerpop.gremlin.structure.util.partitioner.HashPartitioner;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.Collections;
+import java.util.UUID;
+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 ActorProgram actorProgram;
+ private int workers = 1;
+ private Configuration configuration;
+ private boolean executed = false;
+
+ private AkkaGraphActors(final Configuration configuration) {
+ this.configuration = new BaseConfiguration();
+ ConfigurationUtils.copy(configuration, this.configuration);
+ this.configuration.setProperty(GRAPH_ACTORS, AkkaGraphActors.class.getCanonicalName());
+ GraphActorsHelper.configure(this, this.configuration);
+ }
+
+ @Override
+ public String toString() {
+ return StringFactory.graphActorsString(this);
+ }
+
+ @Override
+ public GraphActors<R> program(final ActorProgram actorProgram) {
+ this.actorProgram = actorProgram;
+ return this;
+ }
+
+ @Override
+ public GraphActors<R> workers(final int workers) {
+ this.workers = workers;
+ this.configuration.setProperty(GRAPH_ACTORS_WORKERS, workers);
+ return this;
+ }
+
+ @Override
+ public GraphActors<R> configure(final String key, final Object value) {
+ this.configuration.setProperty(key, value);
+ return this;
+ }
+
+ @Override
+ public Future<R> submit(final Graph graph) {
+ if (this.executed)
+ throw new IllegalStateException("Can not execute twice");
+ this.executed = true;
+ final Config config = ConfigFactory.defaultApplication().withValue("message-priorities",
+ ConfigValueFactory.fromAnyRef(this.actorProgram.getMessagePriorities().
+ orElse(Collections.singletonList(Object.class)).
+ stream().
+ map(Class::getCanonicalName).
+ collect(Collectors.toList()).toString()));
+ final ActorSystem system = ActorSystem.create("traversal-" + UUID.randomUUID(), config);
+ final ActorsResult<R> result = new DefaultActorsResult<>();
+ final Partitioner partitioner = this.workers == 1 ? graph.partitioner() : new HashPartitioner(graph.partitioner(), this.workers);
+ try {
+ new Address.Master(system.actorOf(Props.create(MasterActor.class, this.actorProgram, partitioner, result), "master").path().toString(), InetAddress.getLocalHost());
+ } catch (final UnknownHostException e) {
+ throw new IllegalStateException(e.getMessage(), e);
+ }
+ return CompletableFuture.supplyAsync(() -> {
+ while (!system.isTerminated()) {
+
+ }
+ return result.getResult();
+ });
+ }
+
+ @Override
+ public Configuration configuration() {
+ return this.configuration;
+ }
+
+ public static AkkaGraphActors open(final Configuration configuration) {
+ return new AkkaGraphActors(configuration);
+ }
+
+ public static AkkaGraphActors open() {
+ return new AkkaGraphActors(new BaseConfiguration());
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actors/MasterActor.java
----------------------------------------------------------------------
diff --git a/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actors/MasterActor.java b/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actors/MasterActor.java
new file mode 100644
index 0000000..97951a8
--- /dev/null
+++ b/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actors/MasterActor.java
@@ -0,0 +1,117 @@
+/*
+ * 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.actors;
+
+import akka.actor.AbstractActor;
+import akka.actor.ActorSelection;
+import akka.actor.Props;
+import akka.dispatch.RequiresMessageQueue;
+import akka.japi.pf.ReceiveBuilder;
+import org.apache.tinkerpop.gremlin.process.actors.Actor;
+import org.apache.tinkerpop.gremlin.process.actors.ActorProgram;
+import org.apache.tinkerpop.gremlin.process.actors.ActorsResult;
+import org.apache.tinkerpop.gremlin.process.actors.Address;
+import org.apache.tinkerpop.gremlin.structure.Partition;
+import org.apache.tinkerpop.gremlin.structure.Partitioner;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public final class MasterActor extends AbstractActor implements RequiresMessageQueue<ActorMailbox.ActorSemantics>, Actor.Master {
+
+ private final ActorProgram.Master masterProgram;
+ private final Address.Master master;
+ private final List<Address.Worker> workers;
+ private final Map<Address, ActorSelection> actors = new HashMap<>();
+ private final ActorsResult<?> result;
+ private final Partitioner partitioner;
+
+ public MasterActor(final ActorProgram program, final Partitioner partitioner, final ActorsResult<?> result) {
+ this.partitioner = partitioner;
+ this.result = result;
+ try {
+ this.master = new Address.Master(self().path().toString(), InetAddress.getLocalHost());
+ } catch (final UnknownHostException e) {
+ throw new IllegalStateException(e.getMessage(), e);
+ }
+ this.workers = new ArrayList<>();
+ final List<Partition> partitions = partitioner.getPartitions();
+ for (final Partition partition : partitions) {
+ final String workerPathString = "worker-" + partition.id();
+ this.workers.add(new Address.Worker(workerPathString, partition.location()));
+ context().actorOf(Props.create(WorkerActor.class, program, this.master, partition, partitioner), workerPathString);
+ }
+ this.masterProgram = program.createMasterProgram(this);
+ receive(ReceiveBuilder.matchAny(this.masterProgram::execute).build());
+ }
+
+ @Override
+ public void preStart() {
+ this.masterProgram.setup();
+ }
+
+ @Override
+ public void postStop() {
+ this.masterProgram.terminate();
+ }
+
+ @Override
+ public <M> void send(final Address toActor, final M message) {
+ ActorSelection actor = this.actors.get(toActor);
+ if (null == actor) {
+ actor = context().actorSelection(toActor.getId());
+ this.actors.put(toActor, actor);
+ }
+ actor.tell(message, self());
+ }
+
+ @Override
+ public List<Address.Worker> workers() {
+ return this.workers;
+ }
+
+ @Override
+ public Partitioner partitioner() {
+ return this.partitioner;
+ }
+
+ @Override
+ public Address.Master address() {
+ return this.master;
+ }
+
+ @Override
+ public void close() {
+ context().system().terminate();
+ }
+
+ @Override
+ public <R> ActorsResult<R> result() {
+ return (ActorsResult<R>) this.result;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actors/WorkerActor.java
----------------------------------------------------------------------
diff --git a/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actors/WorkerActor.java b/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actors/WorkerActor.java
new file mode 100644
index 0000000..7520ce4
--- /dev/null
+++ b/akka-gremlin/src/main/java/org/apache/tinkerpop/gremlin/akka/process/actors/WorkerActor.java
@@ -0,0 +1,112 @@
+/*
+ * 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.actors;
+
+import akka.actor.AbstractActor;
+import akka.actor.ActorSelection;
+import akka.dispatch.RequiresMessageQueue;
+import akka.japi.pf.ReceiveBuilder;
+import org.apache.tinkerpop.gremlin.process.actors.Actor;
+import org.apache.tinkerpop.gremlin.process.actors.ActorProgram;
+import org.apache.tinkerpop.gremlin.process.actors.Address;
+import org.apache.tinkerpop.gremlin.structure.Partition;
+import org.apache.tinkerpop.gremlin.structure.Partitioner;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public final class WorkerActor extends AbstractActor implements RequiresMessageQueue<ActorMailbox.ActorSemantics>, Actor.Worker {
+
+ private final ActorProgram.Worker workerProgram;
+ private final Partition localPartition;
+ private final Partitioner partitioner;
+ private final Address.Worker self;
+ private final Address.Master master;
+ private final List<Address.Worker> workers;
+ private final Map<Address, ActorSelection> actors = new HashMap<>();
+
+ public WorkerActor(final ActorProgram program, final Address.Master master, final Partition localPartition, final Partitioner partitioner) {
+ this.localPartition = localPartition;
+ this.partitioner = partitioner;
+ this.self = new Address.Worker(this.createWorkerAddress(localPartition), localPartition.location());
+ this.master = master;
+ this.workers = new ArrayList<>();
+ for (final Partition partition : partitioner.getPartitions()) {
+ this.workers.add(new Address.Worker(this.createWorkerAddress(partition), partition.location()));
+ }
+ this.workerProgram = program.createWorkerProgram(this);
+ receive(ReceiveBuilder.matchAny(this.workerProgram::execute).build());
+ }
+
+ @Override
+ public void preStart() {
+ this.workerProgram.setup();
+ }
+
+ @Override
+ public void postStop() {
+ this.workerProgram.terminate();
+ }
+
+ @Override
+ public <M> void send(final Address toActor, final M message) {
+ ActorSelection actor = this.actors.get(toActor);
+ if (null == actor) {
+ actor = context().actorSelection(toActor.getId());
+ this.actors.put(toActor, actor);
+ }
+ actor.tell(message, self());
+ }
+
+ @Override
+ public List<Address.Worker> workers() {
+ return this.workers;
+ }
+
+ @Override
+ public Partition partition() {
+ return this.localPartition;
+ }
+
+ @Override
+ public Partitioner partitioner() {
+ return this.partitioner;
+ }
+
+ @Override
+ public Address.Worker address() {
+ return this.self;
+ }
+
+ @Override
+ public Address.Master master() {
+ return this.master;
+ }
+
+ private String createWorkerAddress(final Partition partition) {
+ return "../worker-" + partition.id();
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/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 7ced92c..7ee599a 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.ActorMailbox$ActorSemantics"
+ mailbox-requirement = "org.apache.tinkerpop.gremlin.akka.process.actors.ActorMailbox$ActorSemantics"
}
akka.actor.mailbox.requirements {
- "org.apache.tinkerpop.gremlin.akka.process.actor.ActorMailbox$ActorSemantics" = custom-dispatcher-mailbox
+ "org.apache.tinkerpop.gremlin.akka.process.actors.ActorMailbox$ActorSemantics" = custom-dispatcher-mailbox
}
akka {
@@ -11,5 +11,5 @@ akka {
}
custom-dispatcher-mailbox {
- mailbox-type = "org.apache.tinkerpop.gremlin.akka.process.actor.ActorMailbox"
+ mailbox-type = "org.apache.tinkerpop.gremlin.akka.process.actors.ActorMailbox"
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/akka-gremlin/src/test/java/org/apache/tinkerpop/gremlin/akka/process/AkkaActorsProcessActorsTest.java
----------------------------------------------------------------------
diff --git a/akka-gremlin/src/test/java/org/apache/tinkerpop/gremlin/akka/process/AkkaActorsProcessActorsTest.java b/akka-gremlin/src/test/java/org/apache/tinkerpop/gremlin/akka/process/AkkaActorsProcessActorsTest.java
deleted file mode 100644
index 2c1aa57..0000000
--- a/akka-gremlin/src/test/java/org/apache/tinkerpop/gremlin/akka/process/AkkaActorsProcessActorsTest.java
+++ /dev/null
@@ -1,34 +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;
-
-import org.apache.tinkerpop.gremlin.GraphProviderClass;
-import org.apache.tinkerpop.gremlin.process.ProcessActorsSuite;
-import org.apache.tinkerpop.gremlin.process.ProcessStandardSuite;
-import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerGraph;
-import org.junit.runner.RunWith;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-@RunWith(ProcessActorsSuite.class)
-@GraphProviderClass(provider = AkkaActorsProvider.class, graph = TinkerGraph.class)
-public class AkkaActorsProcessActorsTest {
-}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/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
deleted file mode 100644
index 6756e0c..0000000
--- a/akka-gremlin/src/test/java/org/apache/tinkerpop/gremlin/akka/process/AkkaActorsProvider.java
+++ /dev/null
@@ -1,158 +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;
-
-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.AkkaGraphActors;
-import org.apache.tinkerpop.gremlin.process.actor.GraphActors;
-import org.apache.tinkerpop.gremlin.process.traversal.TraversalInterruptionTest;
-import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
-import org.apache.tinkerpop.gremlin.process.traversal.step.ComplexTest;
-import org.apache.tinkerpop.gremlin.process.traversal.step.map.GraphTest;
-import org.apache.tinkerpop.gremlin.process.traversal.step.map.ProfileTest;
-import org.apache.tinkerpop.gremlin.process.traversal.step.map.ProgramTest;
-import org.apache.tinkerpop.gremlin.process.traversal.step.sideEffect.GroupTest;
-import org.apache.tinkerpop.gremlin.process.traversal.step.sideEffect.SideEffectTest;
-import org.apache.tinkerpop.gremlin.process.traversal.step.sideEffect.SubgraphTest;
-import org.apache.tinkerpop.gremlin.process.traversal.strategy.decoration.ElementIdStrategyProcessTest;
-import org.apache.tinkerpop.gremlin.process.traversal.strategy.decoration.EventStrategyProcessTest;
-import org.apache.tinkerpop.gremlin.process.traversal.strategy.decoration.PartitionStrategyProcessTest;
-import org.apache.tinkerpop.gremlin.process.traversal.strategy.decoration.SubgraphStrategyProcessTest;
-import org.apache.tinkerpop.gremlin.structure.Graph;
-import org.apache.tinkerpop.gremlin.structure.VertexProperty;
-import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerEdge;
-import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerElement;
-import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerGraph;
-import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerGraphVariables;
-import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerProperty;
-import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerVertex;
-import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerVertexProperty;
-
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Random;
-import java.util.Set;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-public class AkkaActorsProvider extends AbstractGraphProvider {
-
- private static final Random RANDOM = new Random();
-
- private static Set<String> SKIP_TESTS = new HashSet<>(Arrays.asList(
- "g_V_hasLabelXpersonX_V_hasLabelXsoftwareX_name",
- "g_VX1X_repeatXbothEXcreatedX_whereXwithoutXeXX_aggregateXeX_otherVX_emit_path",
- "g_withBulkXfalseX_withSackX1_sumX_V_out_barrier_sack",
- "g_V_both_groupCountXaX_out_capXaX_selectXkeysX_unfold_both_groupCountXaX_capXaX",
- GraphTest.Traversals.class.getCanonicalName(),
- GroupTest.Traversals.class.getCanonicalName(),
- ComplexTest.Traversals.class.getCanonicalName(),
- SubgraphTest.Traversals.class.getCanonicalName(),
- SideEffectTest.Traversals.class.getCanonicalName(),
- SubgraphStrategyProcessTest.class.getCanonicalName(),
- ProfileTest.Traversals.class.getCanonicalName(),
- PartitionStrategyProcessTest.class.getCanonicalName(),
- EventStrategyProcessTest.class.getCanonicalName(),
- ElementIdStrategyProcessTest.class.getCanonicalName(),
- TraversalInterruptionTest.class.getCanonicalName(),
- ProgramTest.Traversals.class.getCanonicalName()));
-
- private static final Set<Class> IMPLEMENTATION = new HashSet<Class>() {{
- add(TinkerEdge.class);
- add(TinkerElement.class);
- add(TinkerGraph.class);
- add(TinkerGraphVariables.class);
- add(TinkerProperty.class);
- add(TinkerVertex.class);
- add(TinkerVertexProperty.class);
- }};
-
- @Override
- public Map<String, Object> getBaseConfiguration(final String graphName, final Class<?> test, final String testMethodName,
- final LoadGraphWith.GraphData loadGraphWith) {
-
- final TinkerGraph.DefaultIdManager idManager = selectIdMakerFromGraphData(loadGraphWith);
- final String idMaker = (idManager.equals(TinkerGraph.DefaultIdManager.ANY) ? selectIdMakerFromGraphData(loadGraphWith) : idManager).name();
- return new HashMap<String, Object>() {{
- put(Graph.GRAPH, TinkerGraph.class.getName());
- put(TinkerGraph.GREMLIN_TINKERGRAPH_VERTEX_ID_MANAGER, idMaker);
- put(TinkerGraph.GREMLIN_TINKERGRAPH_EDGE_ID_MANAGER, idMaker);
- put(TinkerGraph.GREMLIN_TINKERGRAPH_VERTEX_PROPERTY_ID_MANAGER, idMaker);
- put("skipTest", SKIP_TESTS.contains(testMethodName) || SKIP_TESTS.contains(test.getCanonicalName()));
- if (loadGraphWith == LoadGraphWith.GraphData.CREW)
- put(TinkerGraph.GREMLIN_TINKERGRAPH_DEFAULT_VERTEX_PROPERTY_CARDINALITY, VertexProperty.Cardinality.list.name());
- }};
- }
-
- @Override
- public void clear(final Graph graph, final Configuration configuration) throws Exception {
- if (graph != null) graph.close();
- }
-
- @Override
- public Set<Class> getImplementations() {
- return IMPLEMENTATION;
- }
-
- /**
- * Test that load with specific graph data can be configured with a specific id manager as the data type to
- * be used in the test for that graph is known.
- */
- protected TinkerGraph.DefaultIdManager selectIdMakerFromGraphData(final LoadGraphWith.GraphData loadGraphWith) {
- if (null == loadGraphWith) return TinkerGraph.DefaultIdManager.ANY;
- if (loadGraphWith.equals(LoadGraphWith.GraphData.CLASSIC))
- return TinkerGraph.DefaultIdManager.INTEGER;
- else if (loadGraphWith.equals(LoadGraphWith.GraphData.MODERN))
- return TinkerGraph.DefaultIdManager.INTEGER;
- else if (loadGraphWith.equals(LoadGraphWith.GraphData.CREW))
- return TinkerGraph.DefaultIdManager.INTEGER;
- else if (loadGraphWith.equals(LoadGraphWith.GraphData.GRATEFUL))
- return TinkerGraph.DefaultIdManager.INTEGER;
- else
- throw new IllegalStateException(String.format("Need to define a new %s for %s", TinkerGraph.IdManager.class.getName(), loadGraphWith.name()));
- }
-
-/////////////////////////////
-/////////////////////////////
-/////////////////////////////
-
- @Override
- public GraphTraversalSource traversal(final Graph graph) {
- if ((Boolean) graph.configuration().getProperty("skipTest"))
- return graph.traversal();
- //throw new VerificationException("This test current does not work with Gremlin-Python", EmptyTraversal.instance());
- else {
- final GraphTraversalSource g = graph.traversal();
- return RANDOM.nextBoolean() ?
- g.withProcessor(AkkaGraphActors.open().workers(new Random().nextInt(15) + 1)) :
- g.withProcessor(GraphActors.open(AkkaGraphActors.class));
- }
- }
-
- @Override
- public GraphActors getGraphActors(final Graph graph) {
- return AkkaGraphActors.open().workers(new Random().nextInt(15) + 1);
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/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
deleted file mode 100644
index df40748..0000000
--- a/akka-gremlin/src/test/java/org/apache/tinkerpop/gremlin/akka/process/AkkaPlayTest.java
+++ /dev/null
@@ -1,89 +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;
-
-import org.apache.tinkerpop.gremlin.akka.process.actor.AkkaGraphActors;
-import org.apache.tinkerpop.gremlin.process.actor.GraphActors;
-import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
-import org.apache.tinkerpop.gremlin.structure.Graph;
-import org.apache.tinkerpop.gremlin.structure.io.gryo.GryoIo;
-import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerGraph;
-import org.junit.Ignore;
-import org.junit.Test;
-
-import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.in;
-import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.out;
-
-/**
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-public class AkkaPlayTest {
-
- @Test
- @Ignore
- public void testPlay1() throws Exception {
- final Graph graph = TinkerGraph.open();
- graph.io(GryoIo.build()).readGraph("../data/tinkerpop-modern.kryo");
- GraphTraversalSource g = graph.traversal().withProcessor(GraphActors.open(AkkaGraphActors.class).workers(3));
- // System.out.println(g.V().group().by("name").by(outE().values("weight").fold()).toList());
-
- for (int i = 0; i < 1000; i++) {
- if (12l != g.V().union(out(), in()).values("name").count().next())
- System.out.println(i);
- }
-
- //3, 1.9, 1
- /*for (int i = 0; i < 10000; i++) {
- final Graph graph = TinkerGraph.open();
- graph.io(GryoIo.build()).readGraph("data/tinkerpop-modern.kryo");
- final GraphTraversalSource g = graph.traversal().withComputer();
- final List<Pair<Integer, Traversal.Admin<?, ?>>> traversals = Arrays.asList(
- // match() works
- Pair.with(6, g.V().match(
- as("a").out("created").as("b"),
- as("b").in("created").as("c"),
- as("b").has("name", P.eq("lop"))).where("a", P.neq("c")).select("a", "b", "c").by("name").asAdmin()),
- // side-effects work
- Pair.with(3, g.V().repeat(both()).times(2).
- groupCount("a").by("name").
- cap("a").unfold().order().by(Column.values, Order.decr).limit(3).asAdmin()),
- // barriers work and beyond the local star graph works
- Pair.with(1, g.V().repeat(both()).times(2).hasLabel("person").
- group().
- by("name").
- by(out("created").values("name").dedup().fold()).asAdmin()),
- // no results works
- Pair.with(0, g.V().out("blah").asAdmin())
- );
- for (final Pair<Integer,Traversal.Admin<?, ?>> pair : traversals) {
- final Integer count = pair.getValue0();
- final Traversal.Admin<?,?> traversal = pair.getValue1();
- System.out.println("EXECUTING: " + traversal.getBytecode());
- final TinkerActorSystem<?,?> actors = new TinkerActorSystem<>(traversal.clone(),new HashPartitioner(graph.partitioner(), 3));
- System.out.println(IteratorUtils.asList(actors.getResults().get()));
- if(IteratorUtils.count(actors.getResults().get()) != count)
- throw new IllegalStateException();
- System.out.println("//////////////////////////////////\n");
- }
- }
- }*/
-
- }
-}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/akka-gremlin/src/test/java/org/apache/tinkerpop/gremlin/akka/process/actors/AkkaActorsProcessActorsTest.java
----------------------------------------------------------------------
diff --git a/akka-gremlin/src/test/java/org/apache/tinkerpop/gremlin/akka/process/actors/AkkaActorsProcessActorsTest.java b/akka-gremlin/src/test/java/org/apache/tinkerpop/gremlin/akka/process/actors/AkkaActorsProcessActorsTest.java
new file mode 100644
index 0000000..e0feef0
--- /dev/null
+++ b/akka-gremlin/src/test/java/org/apache/tinkerpop/gremlin/akka/process/actors/AkkaActorsProcessActorsTest.java
@@ -0,0 +1,33 @@
+/*
+ * 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.actors;
+
+import org.apache.tinkerpop.gremlin.GraphProviderClass;
+import org.apache.tinkerpop.gremlin.process.ProcessActorsSuite;
+import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerGraph;
+import org.junit.runner.RunWith;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+@RunWith(ProcessActorsSuite.class)
+@GraphProviderClass(provider = AkkaActorsProvider.class, graph = TinkerGraph.class)
+public class AkkaActorsProcessActorsTest {
+}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/akka-gremlin/src/test/java/org/apache/tinkerpop/gremlin/akka/process/actors/AkkaActorsProvider.java
----------------------------------------------------------------------
diff --git a/akka-gremlin/src/test/java/org/apache/tinkerpop/gremlin/akka/process/actors/AkkaActorsProvider.java b/akka-gremlin/src/test/java/org/apache/tinkerpop/gremlin/akka/process/actors/AkkaActorsProvider.java
new file mode 100644
index 0000000..4168445
--- /dev/null
+++ b/akka-gremlin/src/test/java/org/apache/tinkerpop/gremlin/akka/process/actors/AkkaActorsProvider.java
@@ -0,0 +1,158 @@
+/*
+ * 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.actors;
+
+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.actors.AkkaGraphActors;
+import org.apache.tinkerpop.gremlin.process.actors.GraphActors;
+import org.apache.tinkerpop.gremlin.process.traversal.TraversalInterruptionTest;
+import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
+import org.apache.tinkerpop.gremlin.process.traversal.step.ComplexTest;
+import org.apache.tinkerpop.gremlin.process.traversal.step.map.GraphTest;
+import org.apache.tinkerpop.gremlin.process.traversal.step.map.ProfileTest;
+import org.apache.tinkerpop.gremlin.process.traversal.step.map.ProgramTest;
+import org.apache.tinkerpop.gremlin.process.traversal.step.sideEffect.GroupTest;
+import org.apache.tinkerpop.gremlin.process.traversal.step.sideEffect.SideEffectTest;
+import org.apache.tinkerpop.gremlin.process.traversal.step.sideEffect.SubgraphTest;
+import org.apache.tinkerpop.gremlin.process.traversal.strategy.decoration.ElementIdStrategyProcessTest;
+import org.apache.tinkerpop.gremlin.process.traversal.strategy.decoration.EventStrategyProcessTest;
+import org.apache.tinkerpop.gremlin.process.traversal.strategy.decoration.PartitionStrategyProcessTest;
+import org.apache.tinkerpop.gremlin.process.traversal.strategy.decoration.SubgraphStrategyProcessTest;
+import org.apache.tinkerpop.gremlin.structure.Graph;
+import org.apache.tinkerpop.gremlin.structure.VertexProperty;
+import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerEdge;
+import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerElement;
+import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerGraph;
+import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerGraphVariables;
+import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerProperty;
+import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerVertex;
+import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerVertexProperty;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public class AkkaActorsProvider extends AbstractGraphProvider {
+
+ private static final Random RANDOM = new Random();
+
+ private static Set<String> SKIP_TESTS = new HashSet<>(Arrays.asList(
+ "g_V_hasLabelXpersonX_V_hasLabelXsoftwareX_name",
+ "g_VX1X_repeatXbothEXcreatedX_whereXwithoutXeXX_aggregateXeX_otherVX_emit_path",
+ "g_withBulkXfalseX_withSackX1_sumX_V_out_barrier_sack",
+ "g_V_both_groupCountXaX_out_capXaX_selectXkeysX_unfold_both_groupCountXaX_capXaX",
+ GraphTest.Traversals.class.getCanonicalName(),
+ GroupTest.Traversals.class.getCanonicalName(),
+ ComplexTest.Traversals.class.getCanonicalName(),
+ SubgraphTest.Traversals.class.getCanonicalName(),
+ SideEffectTest.Traversals.class.getCanonicalName(),
+ SubgraphStrategyProcessTest.class.getCanonicalName(),
+ ProfileTest.Traversals.class.getCanonicalName(),
+ PartitionStrategyProcessTest.class.getCanonicalName(),
+ EventStrategyProcessTest.class.getCanonicalName(),
+ ElementIdStrategyProcessTest.class.getCanonicalName(),
+ TraversalInterruptionTest.class.getCanonicalName(),
+ ProgramTest.Traversals.class.getCanonicalName()));
+
+ private static final Set<Class> IMPLEMENTATION = new HashSet<Class>() {{
+ add(TinkerEdge.class);
+ add(TinkerElement.class);
+ add(TinkerGraph.class);
+ add(TinkerGraphVariables.class);
+ add(TinkerProperty.class);
+ add(TinkerVertex.class);
+ add(TinkerVertexProperty.class);
+ }};
+
+ @Override
+ public Map<String, Object> getBaseConfiguration(final String graphName, final Class<?> test, final String testMethodName,
+ final LoadGraphWith.GraphData loadGraphWith) {
+
+ final TinkerGraph.DefaultIdManager idManager = selectIdMakerFromGraphData(loadGraphWith);
+ final String idMaker = (idManager.equals(TinkerGraph.DefaultIdManager.ANY) ? selectIdMakerFromGraphData(loadGraphWith) : idManager).name();
+ return new HashMap<String, Object>() {{
+ put(Graph.GRAPH, TinkerGraph.class.getName());
+ put(TinkerGraph.GREMLIN_TINKERGRAPH_VERTEX_ID_MANAGER, idMaker);
+ put(TinkerGraph.GREMLIN_TINKERGRAPH_EDGE_ID_MANAGER, idMaker);
+ put(TinkerGraph.GREMLIN_TINKERGRAPH_VERTEX_PROPERTY_ID_MANAGER, idMaker);
+ put("skipTest", SKIP_TESTS.contains(testMethodName) || SKIP_TESTS.contains(test.getCanonicalName()));
+ if (loadGraphWith == LoadGraphWith.GraphData.CREW)
+ put(TinkerGraph.GREMLIN_TINKERGRAPH_DEFAULT_VERTEX_PROPERTY_CARDINALITY, VertexProperty.Cardinality.list.name());
+ }};
+ }
+
+ @Override
+ public void clear(final Graph graph, final Configuration configuration) throws Exception {
+ if (graph != null) graph.close();
+ }
+
+ @Override
+ public Set<Class> getImplementations() {
+ return IMPLEMENTATION;
+ }
+
+ /**
+ * Test that load with specific graph data can be configured with a specific id manager as the data type to
+ * be used in the test for that graph is known.
+ */
+ protected TinkerGraph.DefaultIdManager selectIdMakerFromGraphData(final LoadGraphWith.GraphData loadGraphWith) {
+ if (null == loadGraphWith) return TinkerGraph.DefaultIdManager.ANY;
+ if (loadGraphWith.equals(LoadGraphWith.GraphData.CLASSIC))
+ return TinkerGraph.DefaultIdManager.INTEGER;
+ else if (loadGraphWith.equals(LoadGraphWith.GraphData.MODERN))
+ return TinkerGraph.DefaultIdManager.INTEGER;
+ else if (loadGraphWith.equals(LoadGraphWith.GraphData.CREW))
+ return TinkerGraph.DefaultIdManager.INTEGER;
+ else if (loadGraphWith.equals(LoadGraphWith.GraphData.GRATEFUL))
+ return TinkerGraph.DefaultIdManager.INTEGER;
+ else
+ throw new IllegalStateException(String.format("Need to define a new %s for %s", TinkerGraph.IdManager.class.getName(), loadGraphWith.name()));
+ }
+
+/////////////////////////////
+/////////////////////////////
+/////////////////////////////
+
+ @Override
+ public GraphTraversalSource traversal(final Graph graph) {
+ if ((Boolean) graph.configuration().getProperty("skipTest"))
+ return graph.traversal();
+ //throw new VerificationException("This test current does not work with Gremlin-Python", EmptyTraversal.instance());
+ else {
+ final GraphTraversalSource g = graph.traversal();
+ return RANDOM.nextBoolean() ?
+ g.withProcessor(AkkaGraphActors.open().workers(new Random().nextInt(15) + 1)) :
+ g.withProcessor(GraphActors.open(AkkaGraphActors.class));
+ }
+ }
+
+ @Override
+ public GraphActors getGraphActors(final Graph graph) {
+ return AkkaGraphActors.open().workers(new Random().nextInt(15) + 1);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/akka-gremlin/src/test/java/org/apache/tinkerpop/gremlin/akka/process/actors/AkkaPlayTest.java
----------------------------------------------------------------------
diff --git a/akka-gremlin/src/test/java/org/apache/tinkerpop/gremlin/akka/process/actors/AkkaPlayTest.java b/akka-gremlin/src/test/java/org/apache/tinkerpop/gremlin/akka/process/actors/AkkaPlayTest.java
new file mode 100644
index 0000000..d4562eb
--- /dev/null
+++ b/akka-gremlin/src/test/java/org/apache/tinkerpop/gremlin/akka/process/actors/AkkaPlayTest.java
@@ -0,0 +1,89 @@
+/*
+ * 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.actors;
+
+import org.apache.tinkerpop.gremlin.akka.process.actors.AkkaGraphActors;
+import org.apache.tinkerpop.gremlin.process.actors.GraphActors;
+import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
+import org.apache.tinkerpop.gremlin.structure.Graph;
+import org.apache.tinkerpop.gremlin.structure.io.gryo.GryoIo;
+import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerGraph;
+import org.junit.Ignore;
+import org.junit.Test;
+
+import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.in;
+import static org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.__.out;
+
+/**
+ * @author Marko A. Rodriguez (http://markorodriguez.com)
+ */
+public class AkkaPlayTest {
+
+ @Test
+ @Ignore
+ public void testPlay1() throws Exception {
+ final Graph graph = TinkerGraph.open();
+ graph.io(GryoIo.build()).readGraph("../data/tinkerpop-modern.kryo");
+ GraphTraversalSource g = graph.traversal().withProcessor(GraphActors.open(AkkaGraphActors.class).workers(3));
+ // System.out.println(g.V().group().by("name").by(outE().values("weight").fold()).toList());
+
+ for (int i = 0; i < 1000; i++) {
+ if (12l != g.V().union(out(), in()).values("name").count().next())
+ System.out.println(i);
+ }
+
+ //3, 1.9, 1
+ /*for (int i = 0; i < 10000; i++) {
+ final Graph graph = TinkerGraph.open();
+ graph.io(GryoIo.build()).readGraph("data/tinkerpop-modern.kryo");
+ final GraphTraversalSource g = graph.traversal().withComputer();
+ final List<Pair<Integer, Traversal.Admin<?, ?>>> traversals = Arrays.asList(
+ // match() works
+ Pair.with(6, g.V().match(
+ as("a").out("created").as("b"),
+ as("b").in("created").as("c"),
+ as("b").has("name", P.eq("lop"))).where("a", P.neq("c")).select("a", "b", "c").by("name").asAdmin()),
+ // side-effects work
+ Pair.with(3, g.V().repeat(both()).times(2).
+ groupCount("a").by("name").
+ cap("a").unfold().order().by(Column.values, Order.decr).limit(3).asAdmin()),
+ // barriers work and beyond the local star graph works
+ Pair.with(1, g.V().repeat(both()).times(2).hasLabel("person").
+ group().
+ by("name").
+ by(out("created").values("name").dedup().fold()).asAdmin()),
+ // no results works
+ Pair.with(0, g.V().out("blah").asAdmin())
+ );
+ for (final Pair<Integer,Traversal.Admin<?, ?>> pair : traversals) {
+ final Integer count = pair.getValue0();
+ final Traversal.Admin<?,?> traversal = pair.getValue1();
+ System.out.println("EXECUTING: " + traversal.getBytecode());
+ final TinkerActorSystem<?,?> actors = new TinkerActorSystem<>(traversal.clone(),new HashPartitioner(graph.partitioner(), 3));
+ System.out.println(IteratorUtils.asList(actors.getResults().get()));
+ if(IteratorUtils.count(actors.getResults().get()) != count)
+ throw new IllegalStateException();
+ System.out.println("//////////////////////////////////\n");
+ }
+ }
+ }*/
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/tinkerpop/blob/7ee2d372/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/Actor.java
----------------------------------------------------------------------
diff --git a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/Actor.java b/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/Actor.java
deleted file mode 100644
index 5a0b869..0000000
--- a/gremlin-core/src/main/java/org/apache/tinkerpop/gremlin/process/actor/Actor.java
+++ /dev/null
@@ -1,95 +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 org.apache.tinkerpop.gremlin.structure.Partition;
-import org.apache.tinkerpop.gremlin.structure.Partitioner;
-
-import java.util.List;
-
-/**
- * An Actor represents an isolated processing unit that can only be interacted with via messages.
- * Actors are able to send and receive messages. The {@link GraphActors} framework has two types of actors:
- * {@link Master} and {@link Worker}. A master actor is not associated with a particular graph {@link Partition}.
- * Instead, its role is to coordinate the workers and ultimately, yield the final result of the submitted
- * {@link ActorProgram}.
- *
- * @author Marko A. Rodriguez (http://markorodriguez.com)
- */
-public interface Actor {
-
- /**
- * Get the {@link Partitioner} associated with the {@link GraphActors} system.
- *
- * @return the partitioner used to partition (logically and/or physically) the {@link org.apache.tinkerpop.gremlin.structure.Graph}
- */
- public Partitioner partitioner();
-
- /**
- * Get the {@link Address} of the actor.
- *
- * @return the actor's address
- */
- public Address address();
-
- /**
- * Get a list of the {@link Address} values of all the workers in {@link GraphActors} system.
- *
- * @return the worker's addresses
- */
- public List<Address.Worker> workers();
-
- /**
- * Send a message from this actor to another actor given their {@link Address}.
- *
- * @param toActor the actor to receive the messages
- * @param message the message being sent
- * @param <M> the message type
- */
- public <M> void send(final Address toActor, final M message);
-
- public interface Master extends Actor {
-
- public Address.Master address();
-
- public void close();
-
- public <R> ActorsResult<R> result();
-
- }
-
- public interface Worker extends Actor {
-
- public Address.Worker address();
-
- public Address.Master master();
-
- /**
- * Get the {@link Partition} associated with this worker.
- * In principle, this is the subset of the {@link org.apache.tinkerpop.gremlin.structure.Graph} that
- * the worker is "data-local" to.
- *
- * @return the worker's partition
- */
- public Partition partition();
- }
-
-
-}