org.infinispan.distexec.mapreduce

Class MapReduceManagerImpl.MapCombineTask<K,V,KOut,VOut>

java.lang.Object

org.infinispan.distexec.mapreduce.MapReduceManagerImpl.DataContainerTask<K,V>

org.infinispan.distexec.mapreduce.MapReduceManagerImpl.MapCombineTask<K,V,KOut,VOut>

All Implemented Interfaces:

ParallelIterableMap.KeyValueAction<java.lang.Object,org.infinispan.container.entries.InternalCacheEntry>, org.infinispan.persistence.spi.AdvancedCacheLoader.CacheLoaderTask

Enclosing class:

MapReduceManagerImpl

private final class MapReduceManagerImpl.MapCombineTask<K,V,KOut,VOut>
extends MapReduceManagerImpl.DataContainerTask<K,V>
implements org.infinispan.persistence.spi.AdvancedCacheLoader.CacheLoaderTask

This is the parallel staggered map/combine algorithm. Threads from the default fork/join pool traverse container and store key/value pairs in parallel. As one of the threads hits the maxCollectorSize threshold, it takes the snapshot of the current state of the collector and invokes combine on it all while others threads continue to fill up collector up to the point where the threshold is reached again. The thread that broke the collector threshold invokes combine and the algorithm repeats. The benefit of staggered parallel map/combine is manyfold. First, we never exhaust working memory of a node as we batch map/combine execution all while traversal of key/value pairs is in progress. Second, such a staggered combine execution does not cause underlying transport to be completely saturated by intermediate cache put commands; intermediate key/value pairs of map/reduce algorithm are transferred across the cluster smoothly as parallel traversal of container's key/value pairs is progress.

Field Summary

Fields

Modifier and Type	Field and Description
(package private) MapReduceManagerImpl.DefaultCollector<KOut,VOut>	collector
(package private) java.util.Set<KOut>	intermediateKeys
(package private) int	maxCollectorSize
(package private) org.infinispan.commands.read.MapCombineCommand<K,V,KOut,VOut>	mcc

Constructor Summary

Constructors

Constructor and Description
MapReduceManagerImpl.MapCombineTask(MapReduceManagerImpl.DefaultCollector<KOut,VOut> collector, org.infinispan.commands.read.MapCombineCommand<K,V,KOut,VOut> mcc, int maxCollectorSize)

Method Summary

Methods

Modifier and Type	Method and Description
void	apply(java.lang.Object key, org.infinispan.container.entries.InternalCacheEntry v)
protected void	combineAndMigrate()
java.util.Set<KOut>	getIntermediateKeys()
(package private) V	getValue(org.infinispan.container.entries.InternalCacheEntry entry)
private java.lang.Object	getValue(org.infinispan.marshall.core.MarshalledEntry marshalledEntry)
void	processEntry(org.infinispan.marshall.core.MarshalledEntry marshalledEntry, org.infinispan.persistence.spi.AdvancedCacheLoader.TaskContext taskContext)

Methods inherited from class java.lang.Object

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

Field Detail

collector

MapReduceManagerImpl.DefaultCollector<KOut,VOut> collector

mcc

org.infinispan.commands.read.MapCombineCommand<K,V,KOut,VOut> mcc

intermediateKeys

java.util.Set<KOut> intermediateKeys

maxCollectorSize

int maxCollectorSize

Constructor Detail

MapReduceManagerImpl.MapCombineTask

public MapReduceManagerImpl.MapCombineTask(MapReduceManagerImpl.DefaultCollector<KOut,VOut> collector,
                                   org.infinispan.commands.read.MapCombineCommand<K,V,KOut,VOut> mcc,
                                   int maxCollectorSize)

Method Detail

getIntermediateKeys

public java.util.Set<KOut> getIntermediateKeys()

getValue

V getValue(org.infinispan.container.entries.InternalCacheEntry entry)

Overrides:

getValue in class MapReduceManagerImpl.DataContainerTask<K,V>

apply

public void apply(java.lang.Object key,
         org.infinispan.container.entries.InternalCacheEntry v)

Specified by:

apply in interface ParallelIterableMap.KeyValueAction<java.lang.Object,org.infinispan.container.entries.InternalCacheEntry>

combineAndMigrate

protected void combineAndMigrate()
                          throws CacheException

Throws:

CacheException

processEntry

public void processEntry(org.infinispan.marshall.core.MarshalledEntry marshalledEntry,
                org.infinispan.persistence.spi.AdvancedCacheLoader.TaskContext taskContext)
                  throws java.lang.InterruptedException

Specified by:

processEntry in interface org.infinispan.persistence.spi.AdvancedCacheLoader.CacheLoaderTask

Parameters:

marshalledEntry - an iterated entry. Note that MarshalledEntry.getValue() might be null if the fetchValue parameter passed to AdvancedCacheLoader.process(KeyFilter, org.infinispan.persistence.spi.AdvancedCacheLoader.CacheLoaderTask, java.util.concurrent.Executor, boolean, boolean) is false.

taskContext - allows the implementors to decide when to stop the iteration by invoking AdvancedCacheLoader.TaskContext.stop()

Throws:

java.lang.InterruptedException

getValue

private java.lang.Object getValue(org.infinispan.marshall.core.MarshalledEntry marshalledEntry)