[id='distribution-{context}']
= Distributed Caches
Distribution tries to keep a fixed number of copies of any entry in the cache,
configured as `numOwners`.
This allows the cache to scale linearly, storing more data as nodes are added to the
cluster.

As nodes join and leave the cluster, there will be times when a key has more or less than
`numOwners` copies.
In particular, if `numOwners` nodes leave in quick succession, some entries will be lost,
so we say that a distributed cache tolerates `numOwners - 1` node failures.

The number of copies represents a trade-off between performance and durability of data.
The more copies you maintain, the lower performance will be, but also the lower the risk
of losing data due to server or network failures.
Regardless of how many copies are maintained, distribution still scales linearly, and
this is key to {brandname}'s scalability.

The owners of a key are split into one *primary owner*, which coordinates writes to the
key, and zero or more *backup owners*.
To find more about how primary and backup owners are assigned, please read the
link:#key_ownership[Key Ownership] section.

.Distributed mode
image::clustering-distributed.svg[align="center", title="Distributed mode"]

A read operation will request the value from the primary owner, but if it doesn't respond
in a reasonable amount of time, we request the value from the backup owners as well.
(The `infinispan.stagger.delay` system property, in milliseconds, controls the delay
between requests.)
A read operation may require `0` messages if the key is present in the local cache,
or up to `2 * numOwners` messages if all the owners are slow.

A write operation will also result in at most `2 * numOwners` messages: one message from
the originator to the primary owner, `numOwners - 1` messages from the primary to the
backups, and the corresponding ACK messages.

NOTE: Cache topology changes may cause retries and additional messages, both for reads
and for writes.

Just as replicated mode, distributed mode can also be synchronous or asynchronous.
And as in replicated mode, asynchronous replication is not recommended because it can
lose updates.
In addition to losing updates, asynchronous distributed caches can also see a stale value
when a thread writes to a key and then immediately reads the same key.

Transactional distributed caches use the same kinds of messages as transactional
replicated caches, except lock/prepare/commit/unlock messages are sent only to the
*affected nodes* (all the nodes that own at least one key affected by the transaction)
instead of being broadcast to all the nodes in the cluster.
As an optimization, if the transaction writes to a single key and the originator is the
primary owner of the key, lock messages are not replicated.


== Read consistency
Even with synchronous replication, distributed caches are not linearizable.
(For transactional caches, we say they do not support serialization/snapshot isolation.)
We can have one thread doing a single put:

[source,java]
----
cache.get(k) -> v1
cache.put(k, v2)
cache.get(k) -> v2
----

But another thread might see the values in a different order:

[source,java]
----
cache.get(k) -> v2
cache.get(k) -> v1
----

The reason is that read can return the value from *any* owner, depending on how fast
the primary owner replies.
The write is not atomic across all the owners -- in fact, the primary commits the update
only after it receives a confirmation from the backup.
While the primary is waiting for the confirmation message from the backup, reads from the
backup will see the new value, but reads from the primary will see the old one.

[[key_ownership]]
== Key Ownership
Distributed caches split entries into a fixed number of segments and assign
each segment to a list of owner nodes. Replicated caches do the same, with the
exception that every node is an owner.

The first node in the list of owners is the *primary owner*. The other nodes in
the list are *backup owners*. When the cache topology changes, because a node
joins or leaves the cluster, the segment ownership table is broadcast to every
node. This allows nodes to locate keys without making multicast requests or
maintaining metadata for each key.

The `numSegments` property configures the number of segments available.
However, the number of segments cannot change unless the cluster is restarted.

Likewise the key-to-segment mapping cannot change. Keys must always map to the
same segments regardless of cluster topology changes. It is important that the
key-to-segment mapping evenly distributes the number of segments allocated to each node while minimizing the number of segments that must move when the cluster topology changes.

You can customize the key-to-segment mapping by configuring a
link:{javadocroot}/org/infinispan/distribution/ch/KeyPartitioner.html[KeyPartitioner] or by using the
link:#grouping_api[Grouping API].

However, {brandname} provides the following implementations:

SyncConsistentHashFactory::
Uses an algorithm based on
link:http://en.wikipedia.org/wiki/Consistent_hashing[consistent hashing]. Selected by default when server hinting is disabled.
+
This implementation always assigns keys to the same nodes in every cache as
long as the cluster is symmetric. In other words, all caches run on all nodes.
This implementation does have some negative points in that the load distribution is slightly uneven. It also moves more segments than strictly necessary on a join or leave.

TopologyAwareSyncConsistentHashFactory::
Similar to `SyncConsistentHashFactory`, but adapted for
link:#server_hinting[Server Hinting]. Selected by default when server hinting is enabled.

DefaultConsistentHashFactory::
Achieves a more even distribution than `SyncConsistentHashFactory`, but with one
disadvantage. The order in which nodes join the cluster determines which nodes
own which segments. As a result, keys might be assigned to different nodes in
different caches.
+
Was the default from version 5.2 to version 8.1 with server hinting disabled.

TopologyAwareConsistentHashFactory::
Similar to _DefaultConsistentHashFactory_, but adapted for
link:#server_hinting[Server Hinting].
+
Was the default from version 5.2 to version 8.1 with server hinting enabled.

ReplicatedConsistentHashFactory::
Used internally to implement replicated caches. You should never explicitly
select this algorithm in a distributed cache.

=== Capacity Factors
Capacity factors allocate segment-to-node mappings based on resources available
to nodes.

To configure capacity factors, you specify any non-negative number and the
{brandname} hashing algorithm assigns each node a load weighted by its
capacity factor (both as a primary owner and as a backup owner).

For example, nodeA has 2x the memory available than nodeB in the same
{brandname} cluster. In this case, setting `capacityFactor` to a value of `2`
configures {brandname} to allocate 2x the number of segments to nodeA.

Setting a capacity factor of `0` is possible but is recommended only in cases
where nodes are not joined to the cluster long enough to be useful data owners.

== Zero Capacity Node
You might need to configure a whole node where the capacity factor is `0` for every cache,
user defined caches and internal caches.
When defining a zero capacity node, the node won't hold any data.
This is how you declare a zero capacity node:

[source,xml,options="nowrap",subs=attributes+]
----
include::config_examples/zero_capacity_node.xml[]
----

[source, java]
----
include::code_examples/ZeroCapacityNode.java[]
----

However, note that this will be true for distributed caches only. If you are using replicated
caches, the node will still keep a copy of the value. Use only distributed caches to make the
best use of this feature.

== Hashing Configuration
This is how you configure hashing declaratively, via XML:

[source,xml,options="nowrap",subs=attributes+]
----
include::config_examples/hashing_configuration.xml[]
----

And this is how you can configure it programmatically, in Java:
[source,java]
----
include::code_examples/HashingConfiguration.java[]
----

== Initial cluster size
{brandname}'s very dynamic nature in handling topology changes (i.e. nodes being added / removed
at runtime) means that, normally, a node doesn't wait for the presence of other nodes before
starting. While this is very flexible, it might not be suitable for applications which require
a specific number of nodes to join the cluster before caches are started. For this reason,
you can specify how many nodes should have joined the cluster before proceeding with cache
initialization. To do this, use the `initialClusterSize` and `initialClusterTimeout` transport
properties. The declarative XML configuration:

[source,xml,options="nowrap",subs=attributes+]
----
include::config_examples/initial_cluster_size.xml[]
----

The programmatic Java configuration:
[source,java]
----
include::code_examples/InitialClusterSize.java[]
----

The above configuration will wait for _4_ nodes to join the cluster before initialization. If
the initial nodes do not appear within the specified timeout, the cache manager will fail to
start.

== L1 Caching
When L1 is enabled, a node will keep the result of remote reads locally for a short
period of time (configurable, 10 minutes by default), and repeated lookups will return
the local L1 value instead of asking the owners again.

.L1 caching
image::clustering-l1.svg[align="center", title="L1 caching"]

L1 caching is not free though.
Enabling it comes at a cost, and this cost is that every entry update must broadcast an
invalidation message to all the nodes.
L1 entries can be evicted just like any other entry when the the cache is configured
with a maximum size.
Enabling L1 will improve performance for repeated reads of non-local keys, but it will
slow down writes and it will increase memory consumption to some degree.

Is L1 caching right for you?
The correct approach is to benchmark your application with and without L1 enabled and see
what works best for your access pattern.


[[server_hinting]]
== Server Hinting
The following topology hints can be specified:

Machine::
  This is probably the most useful, when multiple JVM instances run on the
  same node, or even when multiple virtual machines run on the same physical machine.
Rack::
  In larger clusters, nodes located on the same rack are more likely to experience a
  hardware or network failure at the same time.
Site::
  Some clusters may have nodes in multiple physical locations for extra resilience.
  Note that Cross site replication is another alternative for
  clusters that need to span two or more data centres.

All of the above are optional.
When provided, the distribution algorithm will try to spread the ownership of each
segment across as many sites, racks, and machines (in this order) as possible.

=== Configuration
The hints are configured at transport level:

[source,xml,options="nowrap",subs=attributes+]
----
include::config_examples/transport_level_server_hint.xml[]
----

[[key_affinity_service]]
==  Key affinity service
In a distributed cache, a key is allocated to a list of nodes with an opaque algorithm.
There is no easy way to reverse the computation and generate a key that maps to a
particular node.
However, we can generate a sequence of (pseudo-)random keys, see what their primary
owner is, and hand them out to the application when it needs a key mapping to a
particular node.

=== API
Following code snippet depicts how a reference to this service can be obtained and used.

[source,java]
----
include::code_examples/KeyAffinityService.java[]
----

The service is started at step 2: after this point it uses the supplied _Executor_ to
generate and queue keys.
At step 3, we obtain a key from the service, and at step 4 we use it.

=== Lifecycle
`KeyAffinityService` extends `Lifecycle`, which allows stopping and (re)starting it:

[source,java]
----
include::code_examples/Lifecycle.java[]
----

The service is instantiated through `KeyAffinityServiceFactory`. All the factory methods
have an `Executor` parameter, that is used for asynchronous key generation (so that it
won't happen in the caller's thread).
It is the user's responsibility to handle the shutdown of this `Executor`.

The `KeyAffinityService`, once started, needs to be explicitly stopped. This stops the
background key generation and releases other held resources.

The only situation in which `KeyAffinityService` stops by itself is when the cache manager
with which it was registered is shutdown.

=== Topology changes
When the cache topology changes (i.e. nodes join or leave the cluster), the ownership of
the keys generated by the `KeyAffinityService` might change.
The key affinity service keep tracks of these topology changes and doesn't return keys
that would currently map to a different node, but it won't do anything about keys
generated earlier.

As such, applications should treat `KeyAffinityService` purely as an optimization, and
they should not rely on the location of a generated key for correctness.

In particular, applications should not rely on keys generated by `KeyAffinityService`
for the same address to always be located together.
Collocation of keys is only provided by the link:#grouping_api[Grouping API].

[[grouping_api]]
===  The Grouping API
Complementary to link:#key_affinity_service[Key affinity service],
the grouping API allows you to co-locate a group of entries on the same nodes, but without
being able to select the actual nodes.

=== How does it work?
By default, the segment of a key is computed using the key's `hashCode()`.
If you use the grouping API, {brandname} will compute the segment of the group and use
that as the segment of the key.
See the link:#key_ownership[Key Ownership] section for more details on how segments are
then mapped to nodes.

When the group API is in use, it is important that every node can still compute the
owners of every key without contacting other nodes.
For this reason, the group cannot be specified manually.
The group can either be intrinsic to the entry (generated by the key class) or extrinsic
(generated by an external function).

=== How do I use the grouping API?
First, you must enable groups. If you are configuring {brandname} programmatically, then call:

[source,java]
----
include::code_examples/EnableGroups.java[]
----

Or, if you are using XML:

[source,xml,options="nowrap",subs=attributes+]
----
include::config_examples/enable_groups.xml[]
----

If you have control of the key class (you can alter the class definition, it's not part of
an unmodifiable library), then we recommend using an intrinsic group.
The intrinsic group is specified by adding the `@Group` annotation to a method.
Let's take a look at an example:

[source,java]
----
include::code_examples/IntrinsicGroup.java[]
----

NOTE: The group method must return a `String`

If you don't have control over the key class, or the determination of the group is an
orthogonal concern to the key class, we recommend using an extrinsic group.
An extrinsic group is specified by implementing the `Grouper` interface.

[source,java]
----
include::code_examples/ExtrinsicGroup.java[]
----

If multiple `Grouper` classes are configured for the same key type, all of them will be
called, receiving the value computed by the previous one.
If the key class also has a `@Group` annotation, the first `Grouper` will receive the
group computed by the annotated method.
This allows you even greater control over the group when using an intrinsic group.
Let's take a look at an example `Grouper` implementation:

[source,java]
----
include::code_examples/GrouperImplementation.java[]
----

`Grouper` implementations must be registered explicitly in the cache configuration.
If you are configuring {brandname} programmatically:

[source,java]
----
include::code_examples/GrouperRegistration.java[]
----

Or, if you are using XML:

[source,xml,options="nowrap",subs=attributes+]
----
include::config_examples/grouper_registration.xml[]
----

=== Advanced Interface

`AdvancedCache` has two group-specific methods:

link:{javadocroot}/org/infinispan/AdvancedCache.html#getGroup-java.lang.String-[getGroup(groupName)]::
  Retrieves all keys in the cache that belong to a group.
link:{javadocroot}/org/infinispan/AdvancedCache.html#removeGroup-java.lang.String-[removeGroup(groupName)]::
  Removes all the keys in the cache that belong to a group.

Both methods iterate over the entire data container and store (if present), so they can
be slow when a cache contains lots of small groups.
