co.cask.cdap.messaging.cache

Class MessageCache<T>

java.lang.Object

co.cask.cdap.messaging.cache.MessageCache<T>

Type Parameters:

T - type of entry stored in the cache

public class MessageCache<T>
extends Object

An in-memory cache for messages. This cache is expected to be shared between publishers and fetchers of the same topic. This cache is specifically designed for the TMS operations, hence leveraging certain properties from TMS: - Single writer, concurrent fetchers - Ordered, unique entry (row key) This cache uses three memory limits to balance between publish and consume efficiency as well as bounding the memory usage. It uses a provided MessageCache.Weigher to compute the weight of each entry being stored inside the cache. - Hard limit. This is the upper bound weight for the cache and it won't grow beyond this. - Min retain. This is the minimum weight that the cache will try to maintain. - Reduce trigger. This is the cache weight that triggers the logic for reducing the cache size back to the min retain weight. When the cache weight is larger than this limit, weight reduction logic will be executed by the consumer. On adding entries to the cache, the cache can keep growing without blocking as long as the hard limit is not hit so that the publisher doesn't need to be blocked. The room between the reduce trigger and hard limits is basically the buffer for non-blocking addition. - On addition, once the hard limit is reached, a blocking operation is needed to reduce the weight of the cache back to min retain. - On fetching entries from the cache, the fetcher will check whether it needs to reduce the cache weight and reduce it if needed. This essentially is to amortize the cost of the blocking weight reduction operations among all fetchers (which typically has multiple of them), without blocking the single publish as much as possible.

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	MessageCache.EntryUpdater<T> A updater for updating an entry.
static class	MessageCache.Limits Carries the limits for the MessageCache.
static interface	MessageCache.Scanner<T> This interface is for accessing cached entries.
static interface	MessageCache.Weigher<T> This interface is for calculating the weight of a cache entry.

Constructor Summary

Constructors

Constructor and Description
MessageCache(Comparator<T> comparator, MessageCache.Weigher<T> weigher, MessageCache.Limits limits, co.cask.cdap.api.metrics.MetricsContext metricsContext) Creates a new instance of the cache.

Method Summary

Modifier and Type	Method and Description
void	addAll(Iterator<T> entries) Adds a list of entries to the cache.
void	clear() Clears the cache.
Comparator<T>	getComparator() Returns the Comparator used by this cache.
MessageCache.Limits	getLimits() Returns the current soft and hard limits of this cache.
void	resize(MessageCache.Limits limits) Resize the cache limits.
MessageCache.Scanner<T>	scan(T startEntry, boolean includeStart, int limit, MessageFilter<T> filter) Creates a MessageCache.Scanner for fetching cached entries in ascending order.
void	updateEntries(T startEntry, T endEntry, MessageCache.EntryUpdater<T> updater) Updates entries in the cache.

Methods inherited from class java.lang.Object

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

Constructor Detail

MessageCache

public MessageCache(Comparator<T> comparator,
                    MessageCache.Weigher<T> weigher,
                    MessageCache.Limits limits,
                    co.cask.cdap.api.metrics.MetricsContext metricsContext)

Creates a new instance of the cache.

Parameters:

comparator - a Comparator for ordering cache entries

weigher - a MessageCache.Weigher for computing the weight of each cache entry

limits - the limits for maintaining cache weight; see class description for more detail

metricsContext - a MetricsContext for emitting metrics about this cache.

Method Detail

getComparator

public Comparator<T> getComparator()

Returns the Comparator used by this cache.

addAll

public void addAll(Iterator<T> entries)

Adds a list of entries to the cache. The entries provided must be in strictly increasing order and should be larger than existing entries in the cache. Also, this method doesn't allow concurrent invocation.

Parameters:

entries - a Iterator to provide entries to be added to the cache

Throws:

ConcurrentModificationException - if called by multiple threads concurrently

IllegalArgumentException - if the entries provided are not in strictly increasing order or not larger existing cached entries

scan

public MessageCache.Scanner<T> scan(T startEntry,
                                    boolean includeStart,
                                    int limit,
                                    MessageFilter<T> filter)

Creates a MessageCache.Scanner for fetching cached entries in ascending order.

Parameters:

startEntry - the entry to start fetching from

includeStart - true to include the startEntry in the resulting MessageCache.Scanner if it exists in the cache

limit - maximum number of entries to fetch

Returns:

a MessageCache.Scanner for accessing to the fetched entries

updateEntries

public void updateEntries(T startEntry,
                          T endEntry,
                          MessageCache.EntryUpdater<T> updater)

Updates entries in the cache. Update to each entry shouldn't change the ordering of the entry based on the Comparator provided to this cache.

Parameters:

startEntry - the starting entry for the update to start (inclusive)

endEntry - the ending entry for the update to end (inclusive)

updater - a MessageCache.EntryUpdater to update the content of a entry

clear

public void clear()

Clears the cache. The caller is responsible to make sure there is no concurrent call to the addAll(Iterator) method.

resize

public void resize(MessageCache.Limits limits)

Resize the cache limits.

Parameters:

limits - the new limits for this cache.

getLimits

public MessageCache.Limits getLimits()

Returns the current soft and hard limits of this cache.