Parallel Queue Pattern

Download
Order Management Processor

 Overview

When building low latency systems, a critical requirement of such systems is to be able to process the incoming data as fast as they can, but also to do that in the exact order the request has been submitted into the system. The system can't process requests associated with the same Order in parallel, but in the exact order these has been created at the client side.

Such a requirement would be relevant with algorithmic trading engines , Order Management Systems , Market Data processing system , High Speed inventory systems, etc.

GigaSpaces introduce the ability to partition data in memory across multiple data-grid partitions. This provides the ability to scale the system, but it does not ensure fast data processing in the correct order. You should add another component that allows the system to "slice" each partition into virtual queues or several buckets. The polling container consuming data in a FIFOClosed manner is the missing required component.

par_q2.jpg

Having multiple polling containers running collocated with each partition allows us to scale at the partition level, forming set of "virtual queues", that consume data pushed into the partition in a parallel manner, but also in the correct order. The amount of polling containers will be usually the number of machine cores. This will optimize the ability to use the machine CPUs in the most efficient manner.

With our example we will simulate a simple Order Management processing system where Orders are sent into the system. An order might have 5 states (state 1-5) where these should be processed in the correct order by a "processor". Different Orders (from different clients) should be processed in parallel, but requests associated with the same order MUST be processed in the exact order they have submitted by the end point client.

Here is an example for the latency duration for the Order request processing time:

par_q1.jpg

The above results retrieved when running the Data-Grid with 4 partitions with a backup.

 The Order Management Processor Example

The following example illustrates a simple Order management processor that includes the following artifacts:

  • The Order Class - This represents an Order request. An order includes a Symbol , requestType, id , orderId and a bucketId field. – The orderId field used to partition the Order requests (its getter method annotated with @SpaceRouting) between the partitions. – The id used as the space object ID (its getter annotated with @SpaceID). – The bucketId used to generate the buckets that each polling container consuming Orders from. – The Order class is decorated with the FIFO mode allows it to be consumed in a FIFO manner.
  • The Feeder - A client running multiple threads pushing requests for different Order.
  • The Processor - Polling Container listener. Running collocated with the data-grid. Each polling container consumes different set of objects from its co located space. Each Polling Container template using different bucketId.
  • The Data-Grid - Stores the incoming Order requests state until these are consumed by the polling containers. It provides the ability to scale the system and also provides high-availability.
  • The ProcessorFactory - A Spring bean. Creates the polling containers and their listeners. The polling containers consumes data based on the Order Symbol. Each associated with a different Symbol. These are running collocated with the Data-Grid.

The bucketId is calculated using the following formula:

(first char of Symbol hashcode) % (# of machine cores)

 Running the Example

You can run the Data-Grid with the collocated Order Processor within your IDEClosed using the following configuration:

par_q4.jpg

Here is a configuration for a data-grid with 2 partitions that will be running the polling containers:

par_q5.jpg

 Example Code and Configuration

package com.gigaspaces.examples.parallelqueue;

import com.gigaspaces.annotation.pojo.FifoSupport;
import com.gigaspaces.annotation.pojo.SpaceClass;
import com.gigaspaces.annotation.pojo.SpaceId;
import com.gigaspaces.annotation.pojo.SpaceIndex;
import com.gigaspaces.annotation.pojo.SpaceRouting;
import com.gigaspaces.metadata.index.SpaceIndexType;

@SpaceClass(fifoSupport=FifoSupport.ALL)
public class Order {

    String requestType;
    String orderId;
    Integer id;
    String symbol;
    Long sendTime;
    Long ackTime;
    Integer bucket;

    public final static String requestTypes[] = new String []{"requestType1" , "requestType2" ,
        "requestType3", "requestType4","requestType5"};
    public final static String symbols[] = new String []{"AAA", "BBB", "CCC","DDD","EEE",
        "FFF","GGG","HHH","III","JJJ","KKK","LLL","MMM", "NNN","OOO","PPP","QQQ", "RRR",
        "SSS" , "TTT", "UUU", "VVV" , "WWW", "XXX" , "YYY", "ZZZ"};

    public String getSymbol() {
        return symbol;
    }
    public void setSymbol(String symbol) {
        this.symbol = symbol;
        setBucket(symbol.substring(0,1).hashCode() % ProcessorFactory.MACHINE_CORES_AMOUNT);
    }
    public Long getSendTime() {
        return sendTime;
    }
    public void setSendTime(Long sendTime) {
        this.sendTime = sendTime;
    }
    public Long getAckTime() {
        return ackTime;
    }
    public void setAckTime(Long ackTime) {
        this.ackTime = ackTime;
    }
    @Override
    public String toString() {
        return "Order ID:"+ orderId+ " symbol " + symbol + " sendTime " +
            sendTime + " requestType "+requestType + " bucket" + bucket;
    }

    @SpaceRouting
    public String getOrderId() {
        return orderId;
    }

    public void setOrderId(String orderId) {
        this.orderId = orderId;
    }

    @SpaceId(autoGenerate=false)
    public Integer getId() {
        return id;
    }
    public void setId(Integer requestOrderId) {
        this.id = requestOrderId;
    }
    public String getRequestType() {
        return requestType;
    }
    public void setRequestType(String requestType) {
        this.requestType = requestType;
    }
    @SpaceIndex (type=SpaceIndexType.BASIC)
    public Integer getBucket() {
        return bucket;
    }
    public void setBucket(Integer bucket) {
        this.bucket= bucket;
    }
}