Overview

Load-balancing is essential in any truly scalable architecture, as it enables scaling beyond the physical resources of a single-server machine. In GigaSpaces, load-balancing is the mechanism used by the clustered proxy to distribute space operations among spaces of a group. Each of these spaces can run on a different physical machine.

If a space belongs to a load-balancing group, the clustered proxy originating from this space contains logical references to all space members in the load-balancing group. The references are "logical", in the sense that no active connection to a space member is opened until it is needed. This is illustrated in the following diagram:

For limitations and considerations when working with a partitioned space; and details regarding multi-space operations, refer to the Partitioned Space Considerations section.

The status of clustered spaces can be viewed using different logging levels. For more details, refer to the Viewing Clustered Space Status section.

Types of Load-Balancing

GigaSpaces Platform ships with a number of built-in load-balancing policies. These range from relatively static policies, where each proxy "attaches" to a specific server and directs all operations to it, to extremely dynamic policies where the target of an operation takes into account the operation data and the relative strength of the machine.

In each load balancing group, a load-balancing policy is specified per basic space operation: write, read, take and notify. This allows you to direct different kinds of operations to different spaces, ensuring correct semantics for the application.

The following table describes the built-in load balancing types.

If you wish, you can implement your own load-balancing policies.

Hash-Based Load-Balancing

When using hash-based load-balancing the client proxy splits new written Entries into relevant cluster space nodes. The relevant space to rout the operation is determined using Entry first indexed field value hashcode. Hash-based load-balancing partition the data based on Entry/template first index field value hash code and the number of partitions to calculate the target space for the operation.
The target space ID is calculated using:

hashcode % (# of partitions)

The indexed attribute must implement the hashCode method to return a number used when your application writes the Entries to the space, and when it reads/takes Entries from the space.
The above formula assumes you got normal distribution of the operations per target space – i.e. same amount of Entries per partition.

It is recommended to use an integer or a long field type for the field used to determine the target space.

Example

A cluster is defined with 3 primary spaces: sp1, sp2, and sp3. Each primary space is associated with a backup space: sp1Backup, sp2Backup, and sp3Backup respectively.

Starting Replication group1:

• Run sp1 – gsInstance: "/./space?schema=cache&cluster_schema=partitioned-sync2backup&total_members=3&id=1"
• Run sp1 Backup – gsInstance: "/./space?schema=cache&cluster_schema=partitioned-sync2backup&total_members=1&id=1&backup_id=1"

Starting Replication group2:

• Run sp2 Backup – gsInstance "/./space?schema=cache&cluster_schema=partitioned-sync2backup&total_members=3&id=2"
  gsInstance "/./space?schema=cache&cluster_schema=partitioned-sync2backup&total_members=2&id=1&backup_id=1"

Starting Replication group3:

• Run sp3 – gsInstance: "/./space?schema=cache&cluster_schema=partitioned-sync2backup&total_members=3&id=3"
• Run sp3 Backup - gsInstance: "/./space?schema=cache&cluster_schema=partitioned-sync2backup&total_members=3&id=1&backup_id=1"

The primary spaces are defined as part of a load-balancing group that uses a hash-based policy. The application writes the Account Entry class objects into the space. The m_accountID attribute as part of the Account class is defined as an indexed attribute using the __getSpaceIndexedFields() method.

public static String[] __getSpaceIndexedFields()
{
String[] indexedFields = { "m_accountID" };
return indexedFields;
}

The Account class implementation:

package com.j_spaces.examples.cluster.lbhashbackup;
public class Account
implements net.jini.core.entry.Entry
{
public Integer m_accountID;
public String m_accountName;
public Account() {}

public Account(int id)
{
m_accountID = new Integer(id);
m_accountName = "Account" + id;
}
public static String[] __getSpaceIndexedFields()
{
String[] indexedFields = { "m_accountID" };
return indexedFields;
}

public String toString()
{
return m_accountName;
}
}

The m_accountID value is used by the space client proxy together with the number of the primary cluster available spaces (3) in the following manner to determine the target space for the write, read, or take operations:

Target Space number = (m_accountID hashCode value) modulo (Number of available spaces)

If we will write 30 Entries with different Account.m_accountID values into a cluster, the Entries will be routed into the 3 partitions in the following manner:

If the relevant target space or its backup space is not active and the cluster does not have a fail-over policy defined, an error message will display.

Weighted-Round-Robin Load-Balancing

The weighted-round-robin will distribute the operations using the defined weights – i.e. will skip to the next space in the list based on its weight.

For example: lets assume we have 3 spaces: SpaceA , SpaceB and SpaceC (Lets assume that these are order of the spaces in the group).

SpaceA weight is 6, SpaceB weight is 3 and SpaceC weight is 1.

The client will perform its first 6 operation with space A, the next 3 operations will be done with space B and the next operation will be done with spaceC.
The 11th operation will again start with spaceA onward.

The above can be configured via the cluster wizard for static clusters.

Weighted-Hash Based Load-Balancing

The weighted-hash based load-balancing allows you to define "regions" per space that are used to direct the Entry/operation to the relevant space. In this way, you direct to each space different amounts of operation/data. The space with high weight will accommodate a large amount of data. Routing to the target space is done based on the Entry's hash code (not the first index field value hash code), and the space weight. This load-balancing policy allows you to direct operations to spaces with a large amount of memory, and a strong or large amount of CPUs.

For example: lets assume we have 3 spaces: SpaceA , SpaceB and SpaceC.

We want SpaceA to accommodate 60% of the data, SpaceB to accommodate 30% of the data, and SpaceC to accommodate the remaining 10%.

The cluster-members cluster configuration should look like the following:

<cluster-members> 
  <member> 
    <member-name>container_SpaceA:SpaceA</member-name> 
    <member-url> 
      jini://*/container_SpaceA/SpaceA 
   </member-url> 
    <param> 
      <param-name>weight</param-name> 
      <param-value>6</param-value> 
    </param> 
  </member> 
  <member> 
    <member-name>container_SpaceB:SpaceB</member-name> 
    <member-url> 
      jini://*/container_SpaceB/SpaceB 
    </member-url> 
    <param> 
      <param-name>weight</param-name> 
      <param-value>3</param-value> 
    </param> 
  </member> 
    <member-name>container_SpaceC:SpaceC</member-name> 
    <member-url> 
      jini://*/container_SpaceC/SpaceC 
    </member-url> 
    <param> 
      <param-name>weight</param-name> 
      <param-value>1</param-value> 
    </param> 
  </member> 
</cluster-members>

Weights can be defined using the Cluster Wizard:

Policy Parameters

The interface that abstracts the load-balancing policy is com.j_spaces.core.cluster.ILoadBalance. The class that implements this interface must be set in the load-balancing group definition. The GigaSpaces class that implements this interface is com.j_spaces.core.cluster.LoadBalanceImpl.

Since the load-balancing policy is designed to be open for many implementations, its parameters are structured in a highly open fashion:

<group>
  ...
  <group-members>...</group-members>
  <load-bal-policy>
    <load-bal-impl-class>
      com.gigaspaces.cluster.loadbalance.LoadBalanceImpl
    </load-bal-impl-class>
    <write>
      <policy-type>select-by-priority</policy-type>
      <param>
        <param-name>A:a</param-name>
        <param-value>5</param-value>
      </param>
      <param>
        <param-name>B:b</param-name>
        <param-value>4</param-value>
      </param>
    </write>
    <default>
      <policy-type>round-robin</policy-type>
    </default>
  </load-bal-policy>
</group>

For each basic operation on the space (write, take, read and notify), a policy type should be defined, except when a policy type is set to the operation. In this case, the default policy is applied for all space operations. Each policy type can take parameters. Each parameter is enclosed by a tag. Each parameter has a name and a value, both of which are interpreted by the load-balancing implementation.

Complex operations inherit their load-balancing policy from the corresponding basic operation. Update gets the policy you define for write; count gets the policy you define for read; and clear gets the policy you define for take.

During runtime, the clustered proxy works with the load-balancing implementation to fulfill the policy.

The hash-based policies take into account the hash code of Entries or templates involved in the operation. Thus, the administrator can influence the load-balancing by overriding the hashCode method in his or her Entries classes.

The weight-based policies take into account the weight set for each space member in the cluster. A member with a higher weight will receive more requests. The weight for a member should be set in the cluster-members section of the cluster configuration file:

<cluster-members>
  <member>
    <member-name>container-name:sp1</member-name>
    <member-url>
      rmi://localhost/container-name/sp1
    </member-url>
    <param>
      <param-name>weight</param-name>
      <param-value>3</param-value>
    </param>
  </member>
  <member>
    <member-name>container-name:sp2</member-name>
    <member-url>
      rmi://localhost/container-name/sp2
    </member-url>
    <param>
      <param-name>weight</param-name>
      <param-value>5</param-value>
    </param>
  </member>
</cluster-members>

You can configure the weight of a space member according to the strength of its physical resources – CPU speed, memory capacity, etc.

Load-balancing with Replication/Failover

Load-balancing can be combined with replication. Depending on the application needs and the load-balancing policy used, this may or may not be necessary. For example, if users tend to perform read operations, and the policy used is round-robin, replication will be necessary to ensure the requested Entry exists on the target space. If, by contrast, the policy used is distribute-by-class, it may not be necessary to replicate, because the class requested should have been written to the same space.

Load-balancing can also be combined with failover, to achieve both scalability and fault tolerance.

</cluster-config>
              <group>
                 <group-name>partitioned_hashbased_group</group-name>
                     <group-members>
                         <xsl:copy-of select="cluster-config/cluster-members/member"/>
                     </group-members>
                     <load-bal-policy>
                          <load-bal-impl-class>com.gigaspaces.cluster.loadbalance.LoadBalanceImpl
                          </load-bal-impl-class>
                              <default>
                                  <policy-type>hash-based</policy-type>
                              </default>
                              <proxy-broadcast-threadpool-min-size>4</proxy-broadcast-threadpool-min-size>
                            <proxy-broadcast-threadpool-max-size>64</proxy-broadcast-threadpool-max-size>
                     </load-bal-policy>
                </group>
         </groups>
</cluster-config>

Parallel Operations - Broadcast

There are three Broadcast options:

• broadcast-if-null-values – for null fields, for hash-based null index. Default for read, take and notify. Its meaning is: the template is a null template (actual null or all fields are null or no fields), or, if the load-blanacing policy is hash-based, the first index (hash index) is null or no index is defined. This option is triggered in the following case: the LB policy is not hash-based and extended-matching (Using SQLQuery, for example) is used by the template, or the LB policy is hash-based and extended matching is used and the match-code for the specific field is not EQ (equal). The latter is designated in order to enable query processor and other extended-matching users to query over multiple spaces.
• unconditional – use broadcast mode anyway – whatever the template field value is.
• broadcast-disabled – disable broadcast. Operations are routed based on a template's first indexed field. If template's first indexed field is null, an error occurs.

Controlling Broadcast Mode

You can control the Broadcast mode for a load-balancing group per space operation. This can be done via the cluster schema configuration file and via the SetLoadBalanceBroadcastCondition API.

The SetLoadBalanceBroadcastCondition returns a LoadBalanceBroadcastInfo object. LoadBalanceBroadcastInfo contains the full information regarding the broadcast of the operation type, including grouping of the cluster members. Get and set functions to handle the LoadBalanceBroadcastInfo exists within the ILoadBalance interface. Broadcast is controlled and integrated with the load-balancer including failover to an alternative member in the same replication group if the broadcast to a source has failed.

LoadBalanceBroadcastInfo indicates how broadcast information for gathering APIs (read, readMultiple,takeMultiple, take etc.) should be distributed between the cluster members by the clustered proxy when the template is a "null" template and taking into account the replication pattern.

IJSpace space = SpaceFinder.find(spaceURL);
JSpaceClusteredProxy clusteredProxy = (JSpaceClusteredProxy) space;
LoadBalanceBroadcastInfo  lbbi = clusteredProxy.setLoadBalanceBroadcastCondition
(LoadBalancingPolicy.BroadcastCondition.BROADCAST_IF_NULL_VALUES , LoadBalancingPolicy.OPERATION_READ);

setLoadBalanceBroadcastCondition enables to change the condition for broadcast
for a load-balancing type of operation (take, read, notify or write)
This api is usually used to set a per-proxy broadcast policy
broadcastCondition : a value from LoadBalancingPolicy.BroadcastCondition
operationType - load balancing operation type : LoadBalancingPolicy.OPERATION_READ ,
LoadBalancingPolicy.OPERATION_TAKE, LoadBalancingPolicy.OPERATION_WRITE or LoadBalancingPolicy.OPERATION_NOTIFY
return the resulted LoadBalanceBroadcastInfo object

public LoadBalanceBroadcastInfo setLoadBalanceBroadcastCondition(LoadBalancingPolicy.BroadcastCondition
broadcastCondition,int operationType )

getLoadBalanceBroadcastCondition gets the current condition for broadcast
for a load-balancing type of operation (take, read, notify or write)
operationType - load balancing operation type : LoadBalancingPolicy.OPERATION_READ ,
LoadBalancingPolicy.OPERATION_TAKE, LoadBalancingPolicy.OPERATION_WRITE or LoadBalancingPolicy.OPERATION_NOTIFY
broadcastCondition : a value from LoadBalancingPolicy.BroadcastCondition for this policy

public LoadBalancingPolicy.BroadcastCondition getLoadBalanceBroadcastCondition(int operationType )

Batch Operation Execution Mode

The following table specifies when the different batch operations executed in parallel manner and when in serial manner when the space running in partitioned mode:

• A Parallel operation on the client means that the partition proxy will run a thread against each of its constituent partition proxies, each thread excuting the operation (take , read). Corollary: every partition will experience a read/take multiple. No partition can guarantee that it will supply objects that the application client obtains in its result.
• A Serial operation on the client means that the partition proxy will execute one read/take multiple on each partition, the partitions being accessed in partition number order, 0 through max partition number. The readMultiple/takeMultiple will complete as soon as the requisite number of objects has been read/taken, or all partitions accessed, whichever comes soonest. Corollary: only partition 0 can guarantee that it will experience a readMultiple/takeMultiple - i.e according to the member ID - Each partition has a numeric running ID associated with it and the processing done according their order(0 to N).
• When executing custom queries against partitioned space (triggers the space filter) you should use the takeMultiple or readMultiple with Integer.MAX_VALUE as the max values to execute the filter operation in parallel across the partitions.
• WriteMultipe/UpdateMultiple with a transaction executed in parallel manner.
• When calling any of the batch operations with a template and set the routing id in the template (making sure all the entries go to the same physical space in a cluster) you do not need to use jini transaction. Local transaction will suffice.
• readMultiple/takeMultiple against any single partition is a synchronous call in the associated partition specific client proxy. The space operation is a synchronous operation. It will return as many requested objects as are currently present in the space, up to the number requested. If no objects are matched an empty array is returned.

Parallel Operation Thread Pool

The client performs parallel operations using a dedicated thread pool managed at the proxy. You may tune this pool size via the following settings at the cluster config:

<proxy-broadcast-threadpool-min-size>4</proxy-broadcast-threadpool-min-size> 
<proxy-broadcast-threadpool-max-size>64</proxy-broadcast-threadpool-max-size>

Limitations

• The replication scheme does not take into account IReplicatable (partial replication) and replication matrix.
• In some cases broadcast can cause ownership/SSI problems to happen.
• Operations under a local transaction are not supported - you should use the Jini Transaction manager.

See the Partitioned Space Considerations section.

Load-Balancing Schema Options