public abstract class AbstractBundler extends Base
Assume that we receive a continuous and concurrent stream of individual operations on multiple threads in parallel. Let's also assume those individual operations have relatively high latency (network or database-related) and there are functionally analogous [bulk] operations that take a collection of arguments instead of a single one without causing the latency to grow linearly, as a function of the collection size. Examples of operations and topologies that satisfy these assumptions are:
NamedCache
API for the partitioned cache service topology;
NamedCache
API for the partitioned cache service topology;
CacheLoader API for the read-through
backing map topology;
CacheStore API for the write-through
backing map topology.
Under these assumptions, it's quite clear that the bundler could achieve a better utilization of system resources and better throughput if slightly delays the individual execution requests with a purpose of "bundling" them together and passing into a corresponding bulk operation. Additionally, the "bundled" request should be triggered if a bundle reaches a "preferred bundle size" threshold, eliminating a need to wait till a bundle timeout is reached.
Note: we assume that all bundle-able operations are idempotent and could be repeated if un-bundling is necessary due to a bundled operation failure.
| Modifier and Type | Class and Description |
|---|---|
protected class |
AbstractBundler.Bundle
Bundle represents a unit of optimized execution.
|
protected static class |
AbstractBundler.Statistics
Statistics class contains the latest bundler statistics.
|
Base.LoggingWriter, Base.StackFrame| Modifier and Type | Field and Description |
|---|---|
static int |
ADJUSTMENT_FREQUENCY
Frequency of the adjustment attempts.
|
protected AtomicInteger |
m_countThreads
A counter for the total number of threads that have started any bundle
related execution.
|
protected double |
m_dPreviousSizeThreshold
The previous bundle size threshold value.
|
protected List |
m_listBundle
A pool of Bundle objects.
|
| Constructor and Description |
|---|
AbstractBundler()
Construct the bundler.
|
| Modifier and Type | Method and Description |
|---|---|
void |
adjust()
Adjust this Bundler's parameters according to the available statistical
information.
|
long |
getDelayMillis()
Obtain the timeout delay value.
|
protected AbstractBundler.Bundle |
getOpenBundle()
Retrieve any Bundle that is currently in the open state.
|
int |
getSizeThreshold()
Obtain the bundle size threshold value.
|
int |
getThreadThreshold()
Obtains the minimum number of threads that will trigger the bundler to
switch from a pass through to a bundled mode.
|
protected abstract AbstractBundler.Bundle |
instantiateBundle()
Instantiate a new Bundle object.
|
boolean |
isAllowAutoAdjust()
Check whether or not the auto-adjustment is allowed.
|
void |
resetStatistics()
Reset this Bundler statistics.
|
void |
setAllowAutoAdjust(boolean fAutoAdjust)
Specify whether or not the auto-adjustment is allowed..
|
void |
setDelayMillis(long lDelay)
Specify the timeout delay value.
|
void |
setSizeThreshold(int cSize)
Specify the bundle size threshold value.
|
void |
setThreadThreshold(int cThreads)
Specify the minimum number of threads that will trigger the bundler to
switch from a pass through to a bundled mode.
|
String |
toString()
Provide a human readable description for the Bundler object
(for debugging).
|
protected void |
updateStatistics()
Update the statistics for this Bundle.
|
azzert, azzert, azzert, azzertFailed, breakLines, breakLines, capitalize, checkNotEmpty, checkNotNull, checkRange, computeSafeWaitTime, decimalValue, dup, dup, ensureBigDecimal, ensureClassLoader, ensureRuntimeException, ensureRuntimeException, equals, equalsDeep, err, err, err, err, err, escape, formatDateTime, getCallerStackFrame, getCommonMonitor, getCommonMonitor, getCommonMonitor, getContextClassLoader, getContextClassLoader, getDeepMessage, getErr, getLastSafeTimeMillis, getLog, getMaxDecDigits, getMaxHexDigits, getOriginalException, getOut, getRandom, getRandomBinary, getRandomBinary, getRandomString, getSafeTimeMillis, getStackFrame, getStackFrames, getStackTrace, getStackTrace, getStackTrace, getThreadFactory, getTimeZone, getUpTimeMillis, hashCode, hexValue, indentString, indentString, isDecimal, isHex, isLogEcho, isOctal, log, log, log, log, log, makeInteger, makeLong, makeThread, mergeArray, mergeBooleanArray, mergeByteArray, mergeCharArray, mergeDoubleArray, mergeFloatArray, mergeIntArray, mergeLongArray, mod, mod, octalValue, out, out, out, out, out, pad, parseBandwidth, parseBandwidth, parseDelimitedString, parseHex, parseHex, parseMemorySize, parseMemorySize, parsePercentage, parseTime, parseTime, parseTimeNanos, parseTimeNanos, printStackTrace, randomize, randomize, randomize, randomize, read, read, read, read, read, read, read, replace, setErr, setLog, setLogEcho, setOut, sleep, toBandwidthString, toBandwidthString, toCharEscape, toCrc, toCrc, toCrc, toCrc, toCrc, toDecString, toDelimitedString, toDelimitedString, toDelimitedString, toDelimitedString, toHex, toHex, toHexDump, toHexEscape, toHexEscape, toHexEscape, toHexEscape, toHexString, toMemorySizeString, toMemorySizeString, toQuotedCharEscape, toQuotedStringEscape, toSqlString, toString, toString, toStringEscape, toUnicodeEscape, trace, trace, trace, trace, trace, trace, trace, trace, trace, truncateString, truncateString, waitpublic static int ADJUSTMENT_FREQUENCY
protected double m_dPreviousSizeThreshold
protected List m_listBundle
protected AtomicInteger m_countThreads
public AbstractBundler()
public int getSizeThreshold()
AbstractBundler.Bundle.getBundleSize() methodpublic void setSizeThreshold(int cSize)
cSize - the bundle size threshold value; must be positive value
expressed in the same units as the value returned by the
AbstractBundler.Bundle.getBundleSize() methodpublic int getThreadThreshold()
public void setThreadThreshold(int cThreads)
cThreads - the number of threads thresholdpublic long getDelayMillis()
public void setDelayMillis(long lDelay)
lDelay - the timeout delay value in millisecondspublic boolean isAllowAutoAdjust()
public void setAllowAutoAdjust(boolean fAutoAdjust)
fAutoAdjust - true if the auto-adjustment should be allowed;
false otherwiseprotected void updateStatistics()
public void resetStatistics()
public void adjust()
public String toString()
protected AbstractBundler.Bundle getOpenBundle()
protected abstract AbstractBundler.Bundle instantiateBundle()