Package com.hazelcast.core

Class Pipelining<E>

java.lang.Object
com.hazelcast.core.Pipelining<E>
Type Parameters:
E - the result type of the Pipelining
@Beta public class Pipelining<E> extends Object
The Pipelining can be used to speed up requests. It is build on top of asynchronous requests like e.g. IMap.getAsync(Object) or any other asynchronous call.

The main purpose of the Pipelining is to control the number of concurrent requests when using asynchronous invocations. This can be done by setting the depth using the constructor. So you could set the depth to e.g. 100 and do 1000 calls. That means that at any given moment, there will only be 100 concurrent requests.

It depends on the situation what the optimal depth (number of invocations in flight) should be. If it is too high, you can run into memory related problems. If it is too low, it will provide little or no performance advantage at all. In most cases a Pipelining and a few hundred map/cache puts/gets should not lead to any problems. For testing purposes we frequently have a Pipelining of 1000 or more concurrent requests to be able to saturate the system.

The Pipelining can't be used for transaction purposes. So you can't create a Pipelining, add a set of asynchronous request and then not call results() to prevent executing these requests. Invocations can be executed before the results() is called.

Pipelining can be used by both clients or members.

The Pipelining isn't thread safe. So only a single thread should add requests to the Pipelining and wait for results.

Currently, all CompletionStage and their responses are stored in the Pipelining. So be careful executing a huge number of request with a single Pipelining because it can lead to a huge memory bubble. In this case it is better to periodically, after waiting for completion, to replace the Pipelining by a new one. In the future we might provide this as an out-of-the-box experience, but currently we do not.

A Pipelining provides its own backpressure on the system. So there will not be more in flight invocations than the depth of the Pipelining. This means that the Pipelining will work fine when backpressure on the client/member is disabled (default). Also, when it is enabled it will work fine, but keep in mind that the number of concurrent invocations in the Pipelining could be lower than the configured number of invocation of the Pipelining because the backpressure on the client/member is leading.

The Pipelining has been marked as Beta since we need to see how the API needs to evolve. But there is no problem using it in production. We use similar techniques to achieve high performance.

  • Constructor Details

    • Pipelining

      public Pipelining(int depth)
      Creates a Pipelining with the given depth.
      Parameters:
      depth - the maximum number of concurrent calls allowed in this Pipelining.
      Throws:
      IllegalArgumentException - if depth smaller than 1. But if you use depth 1, it means that every call is sync, and you will not benefit from pipelining at all.

  • Method Details

    • results

      public List<E> results() throws ExecutionException, InterruptedException
      Returns the results.

      The results are returned in the order the requests were done.

      This call waits till all requests have completed.

      Returns:
      the List of results.
      Throws:
      InterruptedException - if the Thread got interrupted while waiting for the result.
      CancellationException - if any of the futures was cancelled
      ExecutionException - if any of the futures completed exceptionally

    • add

      public CompletionStage<E> add(CompletionStage<E> future) throws InterruptedException
      Adds a future to this Pipelining or blocks until there is capacity to add the future to the Pipelining.

      This call blocks until there is space in the Pipelining, but it doesn't mean that the invocation that returned the CompletionStage got blocked.

      Parameters:
      future - the future to add.
      Returns:
      the future added.
      Throws:
      InterruptedException - if the Thread got interrupted while adding the request to the Pipelining.
      NullPointerException - if future is null.