Add pipeline framework to make parallel processing simpler #1077

ajsutton · 2019-03-08T06:41:54Z

PR description

Adds a new pipeline module to provide a simple way of chaining processing that needs to happen in parallel. This will wind up being used in the WorldStateDownloader to improve its performance.

World state downloading will come out something like:

final Pipeline pipeline =
          createPipelineFrom(new TaskQueueIterator(newDownloadState), persistenceQueueCapacity)
              .thenProcess(this::checkForLocalData)
              .inBatches(hashCountPerRequest)
              .thenProcessAsync(
                  requestTasks -> sendRequestForData(requestTasks, header, newDownloadState),
                  maxOutstandingRequests)
              .thenProcess(tasks -> persistBatch(tasks, header, newDownloadState))
              .thenFlatMap(List::stream, persistenceQueueCapacity)
              .thenProcessInParallel(task -> enqueueChildren(task, header, newDownloadState), 3)
              .andFinishWith(task -> markAsCompleteOrFailed(header, newDownloadState, task));

…ed before all threads finished processing their last entry.

…sume that earlier entries have completed processing by then.

…to stall.

…-framework

…rvices take an unusually long time to start.

This reverts commit 3515376.

…-framework

…ich are not).

mbaxter

A few minor questions / comments, but LGTM!

mbaxter · 2019-03-12T15:47:14Z

services/pipeline/src/main/java/tech/pegasys/pantheon/services/pipeline/OutputPipe.java

+package tech.pegasys.pantheon.services.pipeline;
+
+/**
+ * The end of the pipe that stages write their output to.


Suggested change

* The end of the pipe that stages write their output to.

* The start of the pipe that stages write their output to.

mbaxter · 2019-03-12T15:50:06Z

services/pipeline/src/main/java/tech/pegasys/pantheon/services/pipeline/OutputPipe.java

+ *
+ * @param <T> the type of output.
+ */
+public interface OutputPipe<T> {


So, a Pipe is composed of an OutputPipe and an InputPipe. Which means OutputPipe and InputPipe are really part of a Pipe not "pipes" themselves. I think the relationship might be clearer if OutputPipe became PipeInput and InputPipe became PipeOutput ... As confusing as that seems :P

It depends which way you look at it. The place where pipes are actually held is mostly inside a stage where it reads inputs from it's InputPipe and writes outputs to it's OutputPipe. Admittedly it is a bit weird to think of a pipe as being made up of an output which leads to an input but most of the time we're looking at the pipe from the point of view of a stage which the naming works.

When I tried renaming I inevitably wound up with APIs like:
void processNextInput(PipeOutput inputPipe, PipeInput outputPipe)

So I've side-stepped the issue a bit and used ReadPipe and WritePipe. The variable names are unchanged so a stage has an API like:
void processNextInput(ReadPipe inputPipe, WritePipe outputPipe)

mbaxter · 2019-03-12T16:31:59Z

services/pipeline/src/main/java/tech/pegasys/pantheon/services/pipeline/Pipeline.java

+  private final CompletableFuture<Void> overallFuture = new CompletableFuture<>();
+  private volatile List<Future<?>> futures;
+
+  public Pipeline(


Can this constructor be package-private?

mbaxter · 2019-03-12T16:47:53Z

services/pipeline/src/main/java/tech/pegasys/pantheon/services/pipeline/PipelineBuilder.java

+   */
+  public PipelineBuilder<List<T>> inBatches(final String stageName, final int maximumBatchSize) {
+    return addStage(
+        new BatchingProcessor<>(maximumBatchSize), bufferSize / maximumBatchSize + 1, stageName);


You might want to add an argument check on maximumBatchSize: > 0

Seems reasonable.

mbaxter · 2019-03-12T17:08:35Z

services/pipeline/src/main/java/tech/pegasys/pantheon/services/pipeline/Pipe.java

+  }
+
+  @Override
+  public boolean hasRemainingCapacity() {


If the pipe is not open, should this return 0 ??

Also seems reasonable.

… confusion of which way to look at input and output.

…-framework

…ity when they're closed.

ajsutton added 30 commits March 8, 2019 11:02

Add a basic pipeline framework.

a6ba7bf

Support introducing batches into the pipeline.

2a75bd9

Make buffer size configurable.

1eecb21

Add async processing and parallel processing support to pipelines.

fa06ace

Extract common looping code.

7fa7c59

Introduce finalize step to make InputOutputStep more useful.

91b53ca

Move AsyncProcessStage over to SingleStepStage.

a3970c1

Reduce duplication.

c9b23f1

Add tests for flat map.

2ed23fc

Fix issue where output pipe from a parallel processing stage was clos…

2f2ce8a

…ed before all threads finished processing their last entry.

Support aborting pipelines.

ea8e0c4

Abort the pipeline if any stage throws an unhandled exception.

55ccbcc

Tidy up access levels.

6811215

Add some javadoc.

02e4aa1

Exceptions cause the pipeline to abort quite violently so we can't as…

ebf0feb

…sume that earlier entries have completed processing by then.

Add missing javadoc parameters.

be717e0

Don't use a static executor service.

4b8c544

Simplify test & fix spotless.

8181abb

Ignore PipelineBuilderTest to see if that's what's causing the build …

7b9961f

…to stall.

Try to find the one test case that's stalling.

6a78181

Unignore some more tests.

e3f0f84

Unignore one more test.

1e3a6b0

Unignore final test.

0a4f8cf

Synchronize on overallFuture.

e92fce2

Merge branch 'master' of github.com:PegaSysEng/pantheon into pipeline…

f12e681

…-framework

Add more logging.

3515376

Remove unused executor.

c9e671e

Don't wait forever for vertx services to shutdown. Log warnings if se…

5f9899a

…rvices take an unusually long time to start.

Merge branch 'runner-fix' into pipeline-framework

da999d6

Revert "Add more logging."

38b3b66

This reverts commit 3515376.

ajsutton added 10 commits March 12, 2019 05:56

Rename PipelineSource to ProducingStage.

3d8c447

Merge branch 'master' of github.com:PegaSysEng/pantheon into pipeline…

088838d

…-framework

Fix race condition when aborting.

ca65957

Spotless.

803784e

Add javadoc.

7c799c0

Merge branch 'master' of github.com:PegaSysEng/pantheon into pipeline…

824867b

…-framework

More clearly separate stages (which are runnable) from processors (wh…

dc87314

…ich are not).

Add a counter to track metrics for each stage output count.

714d62c

Put the metrics in the pipe not the stage.

6b2cfb9

Abort the pipeline when the returned future is cancelled.

4cf7cca

mbaxter approved these changes Mar 12, 2019

View reviewed changes

ajsutton added 4 commits March 13, 2019 06:23

Rename InputPipe to ReadPipe and OutputPipe to WritePipe to avoid the…

f8df301

… confusion of which way to look at input and output.

Make Pipeline constructor package private.

9d0be66

Merge branch 'master' of github.com:PegaSysEng/pantheon into pipeline…

30d2893

…-framework

Check batch size is greater than 0. Pipe's don't have remaining capac…

5b336aa

…ity when they're closed.

ajsutton merged commit bf1c861 into PegaSysEng:master Mar 12, 2019

ajsutton deleted the pipeline-framework branch March 12, 2019 20:45

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add pipeline framework to make parallel processing simpler #1077

Add pipeline framework to make parallel processing simpler #1077

ajsutton commented Mar 8, 2019

mbaxter left a comment

mbaxter Mar 12, 2019

mbaxter Mar 12, 2019

ajsutton Mar 12, 2019

mbaxter Mar 12, 2019

ajsutton Mar 12, 2019

mbaxter Mar 12, 2019

ajsutton Mar 12, 2019

mbaxter Mar 12, 2019

ajsutton Mar 12, 2019

	* The end of the pipe that stages write their output to.
	* The start of the pipe that stages write their output to.

Add pipeline framework to make parallel processing simpler #1077

Add pipeline framework to make parallel processing simpler #1077

Conversation

ajsutton commented Mar 8, 2019

PR description

mbaxter left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment