Parallel Reduce Operations

When evaluating all objects in a module, a result value can be accumulated and returned to the caller. Inherit from ReduceApiProcessor and register your class to perform parallel reduce operations.

Background

Unlike an API Processor endpoint method, which only operates against a single SOSS object at a time, a ReduceApiProcessor implementation runs against every instance in your module. These distributed “reduce” operations can be used to analyze every SOSS object in your module and return an aggregated result.

This walkthrough will implement a ReduceApiProcessor for the “MyFirstApiPackage” shopping cart project created in the previous topic.

Prerequisites

Java 8 or higher
Maven
The completed ShoppingCart API module

Procedure

1. Create a ReduceApiProcessor subclass

Create a new class named GetTotalValue that inherits from the ReduceApiProcessor class. This class will be used to aggregate the total value of all the shopping cart items in the module.

The base class takes two type parameters: the type of your API module’s SOSS object and the accumulated (aggregated) result type. There are five abstract methods that must be implemented.

2. Implement the Evaluate() method

Your GetTotalValue.Evaluate() method will be called for every SOSS object in the API module. Clients can supply an arbitrary argument value as a byte array, which is available to the Evaluate method as a parameter.

@Override
public EvalResult<Double> evaluate(ApiProcessingContext<ShoppingCart> apiProcessingContext, Double accumulator, ShoppingCart shoppingCart, byte[] bytes) {
    accumulator = accumulator + shoppingCart.getTotalPrice();
    return new EvalResult<>(accumulator, ProcessingResult.NoUpdate);
}

The evaluate implementation must be marked with the @SossEvalMethod attribute, indicating whether the supplied SOSS object should be exclusively locked for the duration of the Evaluate call. Use ApiProcessorLockingMode.ExclusiveLock in the @SossEvalMethod attribute if you need to ensure that only one request can modify the SOSS object at a time.
The accumulated result for the parallel operation is passed into the evaluate method via the accumulator parameter, and the method should return a (possibly) modified accumulator value in an EvalResult object that will be passed to the next evaluate call by the module’s processing pipeline.
In addition to the accumulated result, the returned EvalResult object must return a ProcessingResult value to indicate what should be done with the object in the ScaleOut service after the method has returned:
- DoUpdate: Indicates that the object was (or may have been) modified and must be updated in the ScaleOut service.
- NoUpdate: Indicates the object was not modified and does not need to be updated in the ScaleOut service. (If you are unsure of whether the object was modified, always return DoUpdate.)
- Remove: Remove the SOSS object from the ScaleOut service.

3. Implement the accumulatorFactory() method

Implement the accumulator factory method to initialize result values at the beginning of a reduce operation.

A reduce operation uses multiple threads to concurrently evaluate objects in a module. Each thread maintains its own thread-local result value to minimize locking overhead, so the accumulatorFactory is called multiple times over the course of a reduce operation.

@Override
public Double accumulatorFactory() {
    return 0.0d;
}

6. Implement the Reduce() method

The reduce() method combines partial results from multiple threads on multiple servers into a single value that is ultimately returned to the call made from a client application. The reduce method must merge two partial results into a single combined result.

@Override
public Double reduce(Double r1, Double r2) {
    return r1+r2;
}

6. Implement result serialization methods

Implement serialization of the result type by overriding the abstract serializeResult and deserializeResult methods.

@Override
public Double deserializeResult(byte[] bytes) {
    ByteBuffer buffer = ByteBuffer.wrap(bytes);
    return buffer.getDouble();
}

@Override
public byte[] serializeResult(Double value) {
    ByteBuffer buffer = ByteBuffer.allocate(Double.BYTES);
    buffer.putDouble(value);
    return buffer.array();
}

7. Register your ReduceApiProcessor implementation

In your API module’s Main.java class, find the line where your API module is registered using ModulePackage.addApiModule(). The ApiModule instance that is returned by this call can be used to add parallel “reduce” operations.

apiModule.addReduceOperation("getTotalValue", new GetTotalValue(), new ParallelOperationOptionsBuilder<ShoppingCart, Double>(ShoppingCart.class).build());

8. Expose the reduce operation to clients

In a client application, the ApiModuleClient subclass that was created in the Creating an API Module Project topic can be used to invoke the new getTotalValue operation. Use the ApiModuleClient.invokeAll() method to run a parallel operation, supplying the name of the operation and an optional byte array as an argument (not used in this example):

public double getTotalValue() throws ApiModuleException {
    byte[] ret = invokeAll("getTotalValue", null, Duration.ofSeconds(10));
    ByteBuffer buffer = ByteBuffer.wrap(ret);
    return buffer.getDouble();
}