Class LockedStreamImpl<K,V>
- java.lang.Object
-
- org.infinispan.stream.impl.LockedStreamImpl<K,V>
-
- All Implemented Interfaces:
AutoCloseable
,BaseStream<CacheEntry<K,V>,LockedStream<K,V>>
,BaseCacheStream<CacheEntry<K,V>,LockedStream<K,V>>
,LockedStream<K,V>
- Direct Known Subclasses:
TxLockedStreamImpl
public class LockedStreamImpl<K,V> extends Object implements LockedStream<K,V>
Lock Stream implementation that locks a key using theLockManager
before and after executing the various code.This implementation doesn't work properly when using an optimistic transactional cache. Care should be made to prevent that usage if possible.
- Since:
- 9.0
- Author:
- wburns
-
-
Nested Class Summary
-
Nested classes/interfaces inherited from interface org.infinispan.BaseCacheStream
BaseCacheStream.SegmentCompletionListener
-
-
Constructor Summary
Constructors Constructor Description LockedStreamImpl(CacheStream<CacheEntry<K,V>> realStream, long time, TimeUnit unit)
-
Method Summary
All Methods Instance Methods Concrete Methods Modifier and Type Method Description void
close()
LockedStream<K,V>
disableRehashAware()
Disables tracking of rehash events that could occur to the underlying cache.LockedStream<K,V>
distributedBatchSize(int batchSize)
Controls how many keys are returned from a remote node when using a stream terminal operation with a distributed cache to back this stream.LockedStream<K,V>
filter(Predicate<? super CacheEntry<K,V>> predicate)
Returns a locked stream consisting of the elements of this stream that match the given predicate.LockedStream<K,V>
filterKeys(Set<?> keys)
Filters which entries are returned by only returning ones that map to the given key.LockedStream<K,V>
filterKeySegments(Set<Integer> segments)
Filters which entries are returned by what segment they are present in.LockedStream<K,V>
filterKeySegments(org.infinispan.commons.util.IntSet segments)
Filters which entries are returned by what segment they are present in.void
forEach(BiConsumer<Cache<K,V>,? super CacheEntry<K,V>> biConsumer)
Performs an action for each element of this stream on the primary owner of the given key.<R> Map<K,R>
invokeAll(BiFunction<Cache<K,V>,? super CacheEntry<K,V>,R> biFunction)
Performs a BiFunction for each element of this stream on the primary owner of each entry returning a value.boolean
isParallel()
Iterator<CacheEntry<K,V>>
iterator()
This method is not supported when using aLockedStream
LockedStream<K,V>
onClose(Runnable closeHandler)
LockedStream<K,V>
parallel()
LockedStream<K,V>
parallelDistribution()
This would enable sending requests to all other remote nodes when a terminal operator is performed.LockedStream
segmentCompletionListener(BaseCacheStream.SegmentCompletionListener listener)
This method is not supported when using aLockedStream
LockedStream<K,V>
sequential()
LockedStream<K,V>
sequentialDistribution()
This would disable sending requests to all other remote nodes compared to one at a time.Spliterator<CacheEntry<K,V>>
spliterator()
This method is not supported when using aLockedStream
LockedStream
timeout(long timeout, TimeUnit unit)
Sets the timeout for the acquisition of the lock for each entry.LockedStream<K,V>
unordered()
-
Methods inherited from class java.lang.Object
clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait
-
Methods inherited from interface org.infinispan.LockedStream
filter, forEach, invokeAll
-
-
-
-
Constructor Detail
-
LockedStreamImpl
public LockedStreamImpl(CacheStream<CacheEntry<K,V>> realStream, long time, TimeUnit unit)
-
-
Method Detail
-
filter
public LockedStream<K,V> filter(Predicate<? super CacheEntry<K,V>> predicate)
Description copied from interface:LockedStream
Returns a locked stream consisting of the elements of this stream that match the given predicate.This filter is after the lock is acquired for the given key. This way the filter will see the same value as the consumer is given.
- Specified by:
filter
in interfaceLockedStream<K,V>
- Parameters:
predicate
- predicate- Returns:
- a LockedStream with the filter applied
-
forEach
public void forEach(BiConsumer<Cache<K,V>,? super CacheEntry<K,V>> biConsumer)
Description copied from interface:LockedStream
Performs an action for each element of this stream on the primary owner of the given key.This method is performed while holding exclusive lock over the given entry and will be released only after the consumer has completed. In the function,
entry.setValue(newValue)
is equivalent tocache.put(entry.getKey(), newValue)
.If using pessimistic transactions this lock is not held using a transaction and thus the user can start a transaction in this consumer which also must be completed before returning. A transaction can be started in the consumer and if done it will share the same lock used to obtain the key.
Remember that if you are using an explicit transaction or an async method that these must be completed before the consumer returns to guarantee that they are operating within the scope of the lock for the given key. Failure to do so will lead into possible inconsistency as they will be performing operations without the proper locking.
Some methods on the provided cache may not work as expected. These include
Cache.putForExternalRead(Object, Object)
,AdvancedCache.lock(Object[])
,AdvancedCache.lock(Collection)
, andAdvancedCache.removeGroup(String)
. If these methods are used inside of the Consumer on the cache it will throw aIllegalStateException
. This is due to possible interactions with locks while using these commands.- Specified by:
forEach
in interfaceLockedStream<K,V>
- Parameters:
biConsumer
- the biConsumer to run for each entry under their lock
-
invokeAll
public <R> Map<K,R> invokeAll(BiFunction<Cache<K,V>,? super CacheEntry<K,V>,R> biFunction)
Description copied from interface:LockedStream
Performs a BiFunction for each element of this stream on the primary owner of each entry returning a value. The returned value from the function will be sent back to the user mapped to the key that generated it, with all of these stored in a map. Both the BiFunction and the returned value must be Serializable in some way. This method will return only after all entries have been processed.This method is currently marked as
Experimental
since this method returns a Map and requires blocking. This operation could take a deal of time and as such should be done using an asynchronous API. Most likely this return type will be changed to use some sort of asynchronous return value. This method is here until this can be implemented.This BiFunction is invoked while holding an exclusive lock over the given entry that will be released only after the function has completed. In the function,
entry.setValue(newValue)
is equivalent tocache.put(entry.getKey(), newValue)
.If using pessimistic transactions this lock is not held using a transaction and thus the user can start a transaction in this consumer which also must be completed before returning. A transaction can be started in the biFunction and if done it will share the same lock used to obtain the key.
Remember if you are using an explicit transaction or an async method that these must be completed before the consumer returns to guarantee that they are operating within the scope of the lock for the given key. Failure to do so will lead into possible inconsistency as they will be performing operations without the proper locking.
Some methods on the provided cache may not work as expected. These include
Cache.putForExternalRead(Object, Object)
,AdvancedCache.lock(Object[])
,AdvancedCache.lock(Collection)
, andAdvancedCache.removeGroup(String)
. If these methods are used inside of the Consumer on the cache it will throw aIllegalStateException
. This is due to possible interactions with locks while using these commands.- Specified by:
invokeAll
in interfaceLockedStream<K,V>
- Type Parameters:
R
- the return type- Parameters:
biFunction
- the biFunction to run for each entry under their lock- Returns:
- a map with each key mapped to the value returned from the bi function
-
sequentialDistribution
public LockedStream<K,V> sequentialDistribution()
Description copied from interface:LockedStream
This would disable sending requests to all other remote nodes compared to one at a time. This can reduce memory pressure on the originator node at the cost of performance.Parallel distribution is enabled by default except for
CacheStream.iterator()
andCacheStream.spliterator()
- Specified by:
sequentialDistribution
in interfaceBaseCacheStream<K,V>
- Specified by:
sequentialDistribution
in interfaceLockedStream<K,V>
- Returns:
- a stream with parallel distribution disabled
-
parallelDistribution
public LockedStream<K,V> parallelDistribution()
Description copied from interface:LockedStream
This would enable sending requests to all other remote nodes when a terminal operator is performed. This requires additional overhead as it must process results concurrently from various nodes, but should perform faster in the majority of cases.Parallel distribution is enabled by default except for
CacheStream.iterator()
andCacheStream.spliterator()
- Specified by:
parallelDistribution
in interfaceBaseCacheStream<K,V>
- Specified by:
parallelDistribution
in interfaceLockedStream<K,V>
- Returns:
- a stream with parallel distribution enabled.
-
filterKeySegments
public LockedStream<K,V> filterKeySegments(Set<Integer> segments)
Description copied from interface:LockedStream
Filters which entries are returned by what segment they are present in. This method can be substantially more efficient than using a regularCacheStream.filter(Predicate)
method as this can control what nodes are asked for data and what entries are read from the underlying CacheStore if present.- Specified by:
filterKeySegments
in interfaceBaseCacheStream<K,V>
- Specified by:
filterKeySegments
in interfaceLockedStream<K,V>
- Parameters:
segments
- The segments to use for this stream operation. Any segments not in this set will be ignored.- Returns:
- a stream with the segments filtered.
-
filterKeySegments
public LockedStream<K,V> filterKeySegments(org.infinispan.commons.util.IntSet segments)
Description copied from interface:LockedStream
Filters which entries are returned by what segment they are present in. This method can be substantially more efficient than using a regularCacheStream.filter(Predicate)
method as this can control what nodes are asked for data and what entries are read from the underlying CacheStore if present.- Specified by:
filterKeySegments
in interfaceBaseCacheStream<K,V>
- Specified by:
filterKeySegments
in interfaceLockedStream<K,V>
- Parameters:
segments
- The segments to use for this stream operation. Any segments not in this set will be ignored.- Returns:
- a stream with the segments filtered.
-
filterKeys
public LockedStream<K,V> filterKeys(Set<?> keys)
Description copied from interface:LockedStream
Filters which entries are returned by only returning ones that map to the given key. This method will be faster than a regularCacheStream.filter(Predicate)
if the filter is holding references to the same keys.- Specified by:
filterKeys
in interfaceBaseCacheStream<K,V>
- Specified by:
filterKeys
in interfaceLockedStream<K,V>
- Parameters:
keys
- The keys that this stream will only operate on.- Returns:
- a stream with the keys filtered.
-
distributedBatchSize
public LockedStream<K,V> distributedBatchSize(int batchSize)
Description copied from interface:LockedStream
Controls how many keys are returned from a remote node when using a stream terminal operation with a distributed cache to back this stream. This value is ignored when terminal operators that don't track keys are used. Key tracking terminal operators areCacheStream.iterator()
,CacheStream.spliterator()
,CacheStream.forEach(Consumer)
. Please see those methods for additional information on how this value may affect them.This value may be used in the case of a a terminal operator that doesn't track keys if an intermediate operation is performed that requires bringing keys locally to do computations. Examples of such intermediate operations are
CacheStream.sorted()
,CacheStream.sorted(Comparator)
,CacheStream.distinct()
,CacheStream.limit(long)
,CacheStream.skip(long)
This value is always ignored when this stream is backed by a cache that is not distributed as all values are already local.
- Specified by:
distributedBatchSize
in interfaceBaseCacheStream<K,V>
- Specified by:
distributedBatchSize
in interfaceLockedStream<K,V>
- Parameters:
batchSize
- The size of each batch. This defaults to the state transfer chunk size.- Returns:
- a stream with the batch size updated
-
segmentCompletionListener
public LockedStream segmentCompletionListener(BaseCacheStream.SegmentCompletionListener listener)
Description copied from interface:LockedStream
This method is not supported when using aLockedStream
- Specified by:
segmentCompletionListener
in interfaceBaseCacheStream<K,V>
- Specified by:
segmentCompletionListener
in interfaceLockedStream<K,V>
- Parameters:
listener
- The listener that will be called back as segments are completed.- Returns:
- a stream with the listener registered.
-
disableRehashAware
public LockedStream<K,V> disableRehashAware()
Description copied from interface:LockedStream
Disables tracking of rehash events that could occur to the underlying cache. If a rehash event occurs while a terminal operation is being performed it is possible for some values that are in the cache to not be found. Note that you will never have an entry duplicated when rehash awareness is disabled, only lost values.Most terminal operations will run faster with rehash awareness disabled even without a rehash occuring. However if a rehash occurs with this disabled be prepared to possibly receive only a subset of values.
- Specified by:
disableRehashAware
in interfaceBaseCacheStream<K,V>
- Specified by:
disableRehashAware
in interfaceLockedStream<K,V>
- Returns:
- a stream with rehash awareness disabled.
-
timeout
public LockedStream timeout(long timeout, TimeUnit unit)
Description copied from interface:LockedStream
Sets the timeout for the acquisition of the lock for each entry.- Specified by:
timeout
in interfaceBaseCacheStream<K,V>
- Specified by:
timeout
in interfaceLockedStream<K,V>
- Parameters:
timeout
- the maximum time to waitunit
- the time unit of the timeout argument- Returns:
- a LockedStream with the timeout applied
-
iterator
public Iterator<CacheEntry<K,V>> iterator()
Description copied from interface:LockedStream
This method is not supported when using aLockedStream
- Specified by:
iterator
in interfaceBaseStream<K,V>
- Specified by:
iterator
in interfaceLockedStream<K,V>
-
spliterator
public Spliterator<CacheEntry<K,V>> spliterator()
Description copied from interface:LockedStream
This method is not supported when using aLockedStream
- Specified by:
spliterator
in interfaceBaseStream<K,V>
- Specified by:
spliterator
in interfaceLockedStream<K,V>
-
isParallel
public boolean isParallel()
- Specified by:
isParallel
in interfaceBaseStream<K,V>
-
sequential
public LockedStream<K,V> sequential()
- Specified by:
sequential
in interfaceBaseStream<K,V>
-
parallel
public LockedStream<K,V> parallel()
- Specified by:
parallel
in interfaceBaseStream<K,V>
-
unordered
public LockedStream<K,V> unordered()
- Specified by:
unordered
in interfaceBaseStream<K,V>
-
onClose
public LockedStream<K,V> onClose(Runnable closeHandler)
- Specified by:
onClose
in interfaceBaseStream<K,V>
-
close
public void close()
- Specified by:
close
in interfaceAutoCloseable
- Specified by:
close
in interfaceBaseStream<K,V>
-
-