K
- type of key.N
- type of namespace.S
- type of value.public class CopyOnWriteStateTable<K,N,S> extends StateTable<K,N,S> implements Iterable<StateEntry<K,N,S>>
CopyOnWriteStateTable
sacrifices some peak performance and memory efficiency for features like incremental
rehashing and asynchronous snapshots through copy-on-write. Copy-on-write tries to minimize the amount of copying by
maintaining version meta data for both, the map structure and the state objects. However, we must often proactively
copy state objects when we hand them to the user.
As for any state backend, user should not keep references on state objects that they obtained from state backends outside the scope of the user function calls.
Some brief maintenance notes:
1) Flattening the underlying data structure from nested maps (namespace) -> (key) -> (state) to one flat map
(key, namespace) -> (state) brings certain performance trade-offs. In theory, the flat map has one less level of
indirection compared to the nested map. However, the nested map naturally de-duplicates namespace objects for which
#equals() is true. This leads to potentially a lot of redundant namespace objects for the flattened version. Those,
in turn, can again introduce more cache misses because we need to follow the namespace object on all operations to
ensure entry identities. Obviously, copy-on-write can also add memory overhead. So does the meta data to track
copy-on-write requirement (state and entry versions on StateTableEntry
).
2) A flat map structure is a lot easier when it comes to tracking copy-on-write of the map structure.
3) Nested structure had the (never used) advantage that we can easily drop and iterate whole namespaces. This could give locality advantages for certain access pattern, e.g. iterating a namespace.
4) Serialization format is changed from namespace-prefix compressed (as naturally provided from the old nested structure) to making all entries self contained as (key, namespace, state).
5) We got rid of having multiple nested tables, one for each key-group. Instead, we partition state into key-groups on-the-fly, during the asynchronous part of a snapshot.
6) Currently, a state table can only grow, but never shrinks on low load. We could easily add this if required.
7) Heap based state backends like this can easily cause a lot of GC activity. Besides using G1 as garbage collector, we should provide an additional state backend that operates on off-heap memory. This would sacrifice peak performance (due to de/serialization of objects) for a lower, but more constant throughput and potentially huge simplifications w.r.t. copy-on-write.
8) We could try a hybrid of a serialized and object based backends, where key and namespace of the entries are both serialized in one byte-array.
9) We could consider smaller types (e.g. short) for the version counting and think about some reset strategy before overflows, when there is no snapshot running. However, this would have to touch all entries in the map.
This class was initially based on the HashMap
implementation of the Android JDK, but is now heavily
customized towards the use case of table for state entries.
IMPORTANT: the contracts for this class rely on the user not holding any references to objects returned by this map
beyond the life cycle of per-element operations. Or phrased differently, all get-update-put operations on a mapping
should be within one call of processElement. Otherwise, the user must take care of taking deep copies, e.g. for
caching purposes.
keyContext, metaInfo
Modifier and Type | Method and Description |
---|---|
boolean |
containsKey(N namespace)
Returns whether this table contains a mapping for the composite of active key and given namespace.
|
CopyOnWriteStateTableSnapshot<K,N,S> |
createSnapshot()
Creates a snapshot of this
CopyOnWriteStateTable , to be written in checkpointing. |
S |
get(K key,
N namespace)
Returns the state for the composite of active key and given namespace.
|
S |
get(N namespace)
Returns the state of the mapping for the composite of active key and given namespace.
|
java.util.stream.Stream<K> |
getKeys(N namespace) |
RegisteredKeyedBackendStateMetaInfo<N,S> |
getMetaInfo() |
TypeSerializer<N> |
getNamespaceSerializer() |
TypeSerializer<S> |
getStateSerializer() |
Iterator<StateEntry<K,N,S>> |
iterator() |
void |
put(K key,
int keyGroup,
N namespace,
S state) |
void |
put(N namespace,
S state)
Maps the composite of active key and given namespace to the specified state.
|
S |
putAndGetOld(N namespace,
S state)
Maps the composite of active key and given namespace to the specified state.
|
void |
remove(N namespace)
Removes the mapping for the composite of active key and given namespace.
|
S |
removeAndGetOld(N namespace)
Removes the mapping for the composite of active key and given namespace, returning the state that was
found under the entry.
|
void |
setMetaInfo(RegisteredKeyedBackendStateMetaInfo<N,S> metaInfo) |
int |
size()
Returns the total number of entries in this
CopyOnWriteStateTable . |
int |
sizeOfNamespace(Object namespace) |
<T> void |
transform(N namespace,
T value,
StateTransformationFunction<S,T> transformation)
Applies the given
StateTransformationFunction to the state (1st input argument), using the given value as
second input argument. |
isEmpty
clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait
forEach, spliterator
public int size()
CopyOnWriteStateTable
. This is the sum of both sub-tables.size
in class StateTable<K,N,S>
CopyOnWriteStateTable
.public S get(K key, N namespace)
StateTable
get
in class StateTable<K,N,S>
key
- the key. Not null.namespace
- the namespace. Not null.null
if no mapping for the specified key is found.public java.util.stream.Stream<K> getKeys(N namespace)
getKeys
in class StateTable<K,N,S>
public void put(K key, int keyGroup, N namespace, S state)
put
in class StateTable<K,N,S>
public S get(N namespace)
StateTable
get
in class StateTable<K,N,S>
namespace
- the namespace. Not null.null
if no mapping for the specified key is found.public boolean containsKey(N namespace)
StateTable
containsKey
in class StateTable<K,N,S>
namespace
- the namespace in the composite key to search for. Not null.true
if this map contains the specified key/namespace composite key,
false
otherwise.public void put(N namespace, S state)
StateTable
#putAndGetOld(N, S)
(Namespace, State)} when the caller is not interested in the old state.put
in class StateTable<K,N,S>
namespace
- the namespace. Not null.state
- the state. Can be null.public S putAndGetOld(N namespace, S state)
StateTable
putAndGetOld
in class StateTable<K,N,S>
namespace
- the namespace. Not null.state
- the state. Can be null.null
if there was no such mapping.public void remove(N namespace)
StateTable
#removeAndGetOld(N)
when the caller is not interested in the old state.remove
in class StateTable<K,N,S>
namespace
- the namespace of the mapping to remove. Not null.public S removeAndGetOld(N namespace)
StateTable
removeAndGetOld
in class StateTable<K,N,S>
namespace
- the namespace of the mapping to remove. Not null.null
if no mapping
for the specified key was found.public <T> void transform(N namespace, T value, StateTransformationFunction<S,T> transformation) throws Exception
StateTable
StateTransformationFunction
to the state (1st input argument), using the given value as
second input argument. The result of StateTransformationFunction.apply(Object, Object)
is then stored as
the new state. This function is basically an optimization for get-update-put pattern.public TypeSerializer<S> getStateSerializer()
getStateSerializer
in class StateTable<K,N,S>
public TypeSerializer<N> getNamespaceSerializer()
getNamespaceSerializer
in class StateTable<K,N,S>
public RegisteredKeyedBackendStateMetaInfo<N,S> getMetaInfo()
getMetaInfo
in class StateTable<K,N,S>
public void setMetaInfo(RegisteredKeyedBackendStateMetaInfo<N,S> metaInfo)
setMetaInfo
in class StateTable<K,N,S>
public Iterator<StateEntry<K,N,S>> iterator()
public CopyOnWriteStateTableSnapshot<K,N,S> createSnapshot()
CopyOnWriteStateTable
, to be written in checkpointing. The snapshot integrity
is protected through copy-on-write from the CopyOnWriteStateTable
. Users should call
releaseSnapshot(CopyOnWriteStateTableSnapshot)
after using the returned object.CopyOnWriteStateTable
, for checkpointing.public int sizeOfNamespace(Object namespace)
sizeOfNamespace
in class StateTable<K,N,S>
Copyright © 2014–2019 The Apache Software Foundation. All rights reserved.