gistfile1.java

    public ComputeResult submit() {
        final long time = System.currentTimeMillis();
        this.vertexMemory.setComputeKeys(this.vertexProgram.getComputeKeys());
        this.vertexProgram.setup(this.graphMemory);

        boolean done = false;
        while (!done) {

            final ExecutorService executor = Executors.newFixedThreadPool(this.threads);
            final Iterator<Vertex> vertices = this.graph.query().vertices().iterator();
            while (vertices.hasNext()) {
                final Runnable worker = new VertexThread(vertices);
                executor.execute(worker);
            }
            executor.shutdown();
            while (!executor.isTerminated()) {
            }

            this.vertexMemory.completeIteration();
            this.graphMemory.incrIteration();
            done = this.vertexProgram.terminate(this.graphMemory);
        }

        this.graphMemory.setRuntime(System.currentTimeMillis() - time);
        return new ComputeResult() {
            @Override
            public GraphMemory getGraphMemory() {
                return graphMemory;
            }

            @Override
            public VertexMemory getVertexMemory() {
                return vertexMemory;
            }
        };
    }

    public class VertexThread implements Runnable {

        private final List<Vertex> vertices = new ArrayList<>();

        public VertexThread(final Iterator<Vertex> vertices) {
            for (int i = 0; i < chunkSize; i++) {
                if (!vertices.hasNext())
                    break;
                this.vertices.add(vertices.next());
            }
        }

        public void run() {
            final CoreShellVertex coreShellVertex = new CoreShellVertex(vertexMemory);
            for (final Vertex vertex : this.vertices) {
                //System.out.println(this + " " + vertex);
                coreShellVertex.setBaseVertex(vertex);
                vertexProgram.execute(coreShellVertex, graphMemory);
            }
        }
    }