Good implementation strategies for encapsulating shared data in a software pipeline

Question

I'm working on re-factoring certain aspects of an existing web service. The way the service APIs are implemented is by having a kind of "processing pipeline", where there are tasks that are performed in sequence. Unsurprisingly, later tasks may need information computed by earlier tasks, and currently the way this is done is by adding fields to a "pipeline state" class.

I've been thinking (and hoping?) that there's a better way to share information between pipeline steps than having a data object with a zillion fields, some of which make sense to some processing steps and not to others. It would be a major pain to make this class thread-safe (I don't know if it would even be possible), there is no way to reason about its invariants (and it's likely it doesn't have any).

I was paging through the Gang of Four design patterns book to find some inspiration, but I didn't feel like there was a solution in there (Memento was somewhat in the same spirit, but not quite). I also looked online, but the second you search for "pipeline" or "workflow" you get flooded with either Unix pipes information, or proprietary workflow engines and frameworks.

My question is - how would you approach the issue of recording the execution state of a software processing pipeline, so that later tasks can use information computed by earlier ones? I guess the major difference with Unix pipes is that you don't just care about the output of the immediately preceding task.

---

As requested, some pseudocode to illustrate my use case:

The "pipeline context" object has a bunch of fields that the different pipeline steps can populate/read:

public class PipelineCtx {
    ... // fields
    public Foo getFoo() { return this.foo; }
    public void setFoo(Foo aFoo) { this.foo = aFoo; }
    public Bar getBar() { return this.bar; }
    public void setBar(Bar aBar) { this.bar = aBar; }
    ... // more methods
}

Each of the pipeline steps is also an object:

public abstract class PipelineStep {
    public abstract PipelineCtx doWork(PipelineCtx ctx);
}

public class BarStep extends PipelineStep {
    @Override
    public PipelineCtx doWork(PipelieCtx ctx) {
        // do work based on the stuff in ctx
        Bar theBar = ...; // compute it
        ctx.setBar(theBar);

        return ctx;
    }
}

Similarly for a hypothetical FooStep, which might need the Bar computed by BarStep before it, along with other data. And then we have the real API call:

public class BlahOperation extends ProprietaryWebServiceApiBase {
    public BlahResponse handle(BlahRequest request) {
        PipelineCtx ctx = PipelineCtx.from(request);

        // some steps happen here
        // ...

        BarStep barStep = new BarStep();
        barStep.doWork(crx);

        // some more steps maybe
        // ...

        FooStep fooStep = new FooStep();
        fooStep.doWork(ctx);

        // final steps ...

        return BlahResponse.from(ctx);
    }
}

Answer 1

The main reason to use a pipeline design is that you want to decouple the stages. Either because one stage may be used in multiple pipelines (like the Unix shell tools), or because you gain some scaling benefit (ie, you can easily move from a single-node architecture to a multi-node architecture).

In either case, each stage in the pipeline needs to be given everything that it needs to do its job. There's no reason that you can't use an external store (eg, database), but in most cases it's better to pass the data from one stage to another.

However, that doesn't mean that you must or should pass one big message object with every possible field (although see below). Instead, each stage in the pipeline should define interfaces for its input and output messages, that identify just the data that stage needs.

You then have a lot of flexibility in how you implement your actual message objects. One approach is to use a huge data object that implements all the necessary interfaces. Another is to create wrapper classes around a simple Map. Still another is to create a wrapper class around a database.

Answer 2

This looks like a Chain Pattern in GoF.

A good starting point would be to look at what commons-chain does.

A popular technique for organizing the execution of complex processing flows is the "Chain of Responsibility" pattern, as described (among many other places) in the classic "Gang of Four" design patterns book. Although the fundamental API contracts required to implement this design patten are extremely simple, it is useful to have a base API that facilitates using the pattern, and (more importantly) encouraging composition of command implementations from multiple diverse sources.

Towards that end, the Chain API models a computation as a series of "commands" that can be combined into a "chain". The API for a command consists of a single method (execute()), which is passed a "context" parameter containing the dynamic state of the computation, and whose return value is a boolean that determines whether or not processing for the current chain has been completed (true), or whether processing should be delegated to the next command in the chain (false).

The "context" abstraction is designed to isolate command implementations from the environment in which they are run (such as a command that can be used in either a Servlet or Portlet, without being tied directly to the API contracts of either of these environments). For commands that need to allocate resources prior to delegation, and then release them upon return (even if a delegated-to command throws an exception), the "filter" extension to "command" provides a postprocess() method for this cleanup. Finally, commands can be stored and looked up in a "catalog" to allow deferral of the decision on which command (or chain) is actually executed.

To maximize the usefulness of the Chain of Responsibility pattern APIs, the fundamental interface contracts are defined in a manner with zero dependencies other than an appropriate JDK. Convenience base class implementations of these APIs are provided, as well as more specialized (but optional) implementations for the web environment (i.e. servlets and portlets).

Given that command implementations are designed to conform with these recommendations, it should be feasible to utilize the Chain of Responsibility APIs in the "front controller" of a web application framework (such as Struts), but also be able to use it in the business logic and persistence tiers to model complex computational requirements via composition. In addition, separation of a computation into discrete commands that operate on a general purpose context allows easier creation of commands that are unit testable, because the impact of executing a command can be directly measured by observing the corresponding state changes in the context that is supplied...

Answer 3

There are a few thoughts that leap to mind, first of which is that I don't have enough information.

• Does each step produce data used beyond the pipeline, or do we only care about the results of the last stage?
• Are there many big data concerns? ie. memory concerns, speed concerns, etc

The answers would probably make me think more carefully about the design, however based on what you said there are 2 approaches I would probably consider first.

Structure each stage as it's own object. The nth stage would have 1 through n-1 stages as a list of delegates. The each stage encapsulates the data and the processing of the data; reducing overall complexity and fields within each object. You can also have later stages access the data as needed from much earlier stages by traversing the delegates. You still have pretty tight coupling across all the objects because it's the results of the stages (ie. all the attrs) that are important, but it's significantly reduced and each stage/object is probably more readable and understandable. You could make it thread safe by making the list of delegates lazy and using a thread safe queue to populate the delegate list in each object as needed.

Alternatively I would probably do something similar to what your doing. A massive data object that goes through functions representing each stage. This is often much faster and light weight, but more complex and error prone because of it's just a big pile of data attributes. Obviously not thread-safe.

Honestly I done the later one more often for ETL and some other similar problems. I was focused on performance because of the amount of data rather than maintainability. Also, they were one-offs which wouldn't be used again.

Answer 4

A first solution I can imagine is to make the steps explicit. Each of them becomes an object able to process a piece of data and transmit it to the next process object. Each process produces a new (ideally immutable) product, so that there is no interaction between the processes and then there is no risk due to data sharing. If some processes are more time consuming than some others, you can place some a buffer between two process. If you correctly exploit a scheduler for the multithreading, it will allocate more ressources to flush the buffers.

A second solution could be to think "message" instead of pipeline, possibly with a dedicated framework. You have then some "actors" receiving messages from other actors and sending other messages to other actors. You organize your actors in a pipeline and give your primary data to a first actor who initiates the chain. There is no data sharing since the sharing is replaced by messages sending. I know the Scala's actor model can be used in Java, since there is nothing Scala specific here, but I have never use it in a Java program.

The solutions are similar and you can implement the second one with the first one. Basically, the main concepts are to deal with immutable data to avoid the traditional problems due to the data sharing and to create explicit and independent entities representing the processes in your pipeline. If you satisfy these conditions, you can easily create clear, simple pipelines and use them in a parallel program.