org.infinispan.iteration.impl

Class DistributedEntryRetriever<K,V>

java.lang.Object

org.infinispan.iteration.impl.LocalEntryRetriever<K,V>

org.infinispan.iteration.impl.DistributedEntryRetriever<K,V>

All Implemented Interfaces:

EntryRetriever<K,V>

public class DistributedEntryRetriever<K,V>
extends LocalEntryRetriever<K,V>

This is an entry retriever designed to retrieve values for a distributed cache. This requests entries by segments to further optimize when a rehash occurs to request less data each time.

The way this works is when a new entry retriever is acquired it first calculates the remote node that has the most segments and sends a request to it (this is done asynchronously). Also another thread is spawned off that processes the local data asynchronously. When either is done (remote sends response) then it will process the entries found and complete all segments that were marked as completed. If it is a remote invocation then it will send a new remote request to the node that now has the most remaining segments. If it is local it will complete those segments and stop running, unless a rehash has caused it to regain new local segments. If a rehash occurs during a remote node processing then those segments will automatically marked as suspect to not complete them.

Values retrieved for suspect segments are sent to the iterator and it is noted that they were raised. When these segments are requested again the noted keys are sent along to reduce value payload size.

Since:

7.0

Author:

wburns

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
protected class	DistributedEntryRetriever.DistributedItr<C>

Nested classes/interfaces inherited from class org.infinispan.iteration.impl.LocalEntryRetriever

LocalEntryRetriever.BatchHandler<K,C>, LocalEntryRetriever.DataContainerIterator<C>, LocalEntryRetriever.Itr<C>, LocalEntryRetriever.ItrQueuerHandler<C>, LocalEntryRetriever.KeyValueActionForCacheLoaderTask, LocalEntryRetriever.PartitionListener, LocalEntryRetriever.PassivationListener<K,V>

Nested classes/interfaces inherited from interface org.infinispan.iteration.impl.EntryRetriever

EntryRetriever.SegmentListener

Field Summary

Fields inherited from class org.infinispan.iteration.impl.LocalEntryRetriever

batchSize, cache, componentRegistry, dataContainer, entryFactory, executorService, keyEquivalence, log, partitionListener, timeout, timeService, unit, withinThreadExecutor

Constructor Summary

Constructors

Constructor and Description
DistributedEntryRetriever(int batchSize, long timeout, TimeUnit unit)

Method Summary

Modifier and Type	Method and Description
void	dataRehashed(DataRehashedEvent<K,V> event) We need to listen to data rehash events in case if data moves while we are iterating over it.
void	initialize(DistributionManager distributionManager, PersistenceManager persistenceManager, CommandsFactory commandsFactory, RpcManager rpcManager, ExecutorService remoteExecutorService)
<C> void	receiveResponse(UUID identifier, Address origin, Set<Integer> completedSegments, Set<Integer> inDoubtSegments, Collection<CacheEntry<K,C>> entries, CacheException e) This method is invoked on the local node who started the iteration process for each batch of values.
<C> CloseableIterator<CacheEntry<K,C>>	retrieveEntries(KeyValueFilter<? super K,? super V> filter, Converter<? super K,? super V,? extends C> converter, Set<Flag> flags, EntryRetriever.SegmentListener listener) This is invoked locally on the node that requested the iteration process.
void	start()
<C> void	startRetrievingValues(UUID identifier, Address origin, Set<Integer> segments, Set<K> keysToFilter, KeyValueFilter<? super K,? super V> filter, Converter<? super K,? super V,C> converter, Set<Flag> flags) This method is intended to be ran remotely on a node who has segments that the values have been requested.
void	topologyChanged(TopologyChangedEvent<K,V> event) We need to listen for topology change events.

Methods inherited from class org.infinispan.iteration.impl.LocalEntryRetriever

inject, registerIterator, shouldUseLoader, unwrapMarshalledvalue, wireFilterAndConverterDependencies

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Detail

DistributedEntryRetriever

public DistributedEntryRetriever(int batchSize,
                                 long timeout,
                                 TimeUnit unit)

Method Detail

dataRehashed

public void dataRehashed(DataRehashedEvent<K,V> event)

We need to listen to data rehash events in case if data moves while we are iterating over it. This is both important for the originator of the entry retriever request and remote nodes. If a rehash occurs causing this node to lose a segment and there is something iterating over the data container looking for values of that segment, we can't guarantee that the data has all been seen correctly, so we must therefore suspect that node and subsequently request it again from the new owner later.

Parameters:

event - The data rehash event

topologyChanged

public void topologyChanged(TopologyChangedEvent<K,V> event)

We need to listen for topology change events. This is important for the originator so it can know when a node goes down that it needs to now send the new request to the next remote node. Also if the originator has acquired some of the segments

Parameters:

event - The topology change event

initialize

public void initialize(DistributionManager distributionManager,
                       PersistenceManager persistenceManager,
                       CommandsFactory commandsFactory,
                       RpcManager rpcManager,
                       ExecutorService remoteExecutorService)

start

public void start()

Overrides:

start in class LocalEntryRetriever<K,V>

startRetrievingValues

public <C> void startRetrievingValues(UUID identifier,
                                      Address origin,
                                      Set<Integer> segments,
                                      Set<K> keysToFilter,
                                      KeyValueFilter<? super K,? super V> filter,
                                      Converter<? super K,? super V,C> converter,
                                      Set<Flag> flags)

Description copied from interface: EntryRetriever

This method is intended to be ran remotely on a node who has segments that the values have been requested.

Specified by:

startRetrievingValues in interface EntryRetriever<K,V>

Overrides:

startRetrievingValues in class LocalEntryRetriever<K,V>

Type Parameters:

C - The resulting type of the Converter

Parameters:

identifier - The unique identifier of the iteration request

origin - The node that sent the iteration request

segments - The segments this node wants

keysToFilter - The keys to filter out (can be null)

filter - The filter to be applied to determine if a value should be used

converter - The converter to run on the values retrieved before returning

retrieveEntries

public <C> CloseableIterator<CacheEntry<K,C>> retrieveEntries(KeyValueFilter<? super K,? super V> filter,
                                                              Converter<? super K,? super V,? extends C> converter,
                                                              Set<Flag> flags,
                                                              EntryRetriever.SegmentListener listener)

Description copied from interface: EntryRetriever

This is invoked locally on the node that requested the iteration process. This method will return immediately with the iterator and will process the request asynchronously making more values available as they are received.

Specified by:

retrieveEntries in interface EntryRetriever<K,V>

Overrides:

retrieveEntries in class LocalEntryRetriever<K,V>

Type Parameters:

C - The type of the resulting values from the converter

Parameters:

filter - An optional filter that will be ran on each key/value to determine if it should be returned.

converter - An optional converter that will be ran on each key/value that will be returned to transform the value to a different value if desired

flags - An optional set of flags to modify behavior. For example Flag.CACHE_MODE_LOCAL will prevent the retriever from retrieving remote values and Flag.SKIP_CACHE_LOAD will prevent the retriever from getting values from the configured loader if present.

listener - An optional segment listener that can be used to tell the invoker when segments and the iteration process is completed

Returns:

An iterator that should be closed when done working it it, especially if not fully iterated over

receiveResponse

public <C> void receiveResponse(UUID identifier,
                                Address origin,
                                Set<Integer> completedSegments,
                                Set<Integer> inDoubtSegments,
                                Collection<CacheEntry<K,C>> entries,
                                CacheException e)

Description copied from interface: EntryRetriever

This method is invoked on the local node who started the iteration process for each batch of values. When either completedSegments or inDoubtSegments is not empty (not both could be as well) then the iteration process on this node is complete.

Specified by:

receiveResponse in interface EntryRetriever<K,V>

Overrides:

receiveResponse in class LocalEntryRetriever<K,V>

Type Parameters:

C - The type of entries values sent back

Parameters:

identifier - The unique identifier of the iteration request

origin - The node where the response came from

completedSegments - Which segments have been completed

inDoubtSegments - Which segments are now in doubt due to a rehash

entries - The entries retrieved

e - If an exception handled while processing the data on the remote node