flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/CopyOnWriteStateTable.java
/**
	 * Runs a number of steps for incremental rehashing.
	 */
Runs a number of steps for incremental rehashing.
Runs a number of steps for incremental rehashing .
@SuppressWarnings("unchecked")
	private void incrementalRehash() {
		StateTableEntry<K, N, S>[] oldTable = primaryTable;
		StateTableEntry<K, N, S>[] newTable = incrementalRehashTable;
		int oldCapacity = oldTable.length;
		int newMask = newTable.length - 1;
		int requiredVersion = highestRequiredSnapshotVersion;
		int rhIdx = rehashIndex;
		int transferred = 0;
		while (transferred < MIN_TRANSFERRED_PER_INCREMENTAL_REHASH) {
			StateTableEntry<K, N, S> e = oldTable[rhIdx];
			while (e != null) {
				if (e.entryVersion < requiredVersion) {
					e = new StateTableEntry<>(e, stateTableVersion);
				}
				StateTableEntry<K, N, S> n = e.next;
				int pos = e.hash & newMask;
				e.next = newTable[pos];
				newTable[pos] = e;
				e = n;
				++transferred;
			}
			oldTable[rhIdx] = null;
			if (++rhIdx == oldCapacity) {
				primaryTable = newTable;
				incrementalRehashTable = (StateTableEntry<K, N, S>[]) EMPTY_TABLE;
				primaryTableSize += incrementalRehashTableSize;
				incrementalRehashTableSize = 0;
				rehashIndex = 0;
				return;
			}
		}
		primaryTableSize -= transferred;
		incrementalRehashTableSize += transferred;
		rehashIndex = rhIdx;
	}
@ SuppressWarnings ( "unchecked" ) private void incrementalRehash ( ) { StateTableEntry < K , N , S > [ ] oldTable = primaryTable ; StateTableEntry < K , N , S > [ ] newTable = incrementalRehashTable ; int oldCapacity = oldTable . length ; int newMask = newTable . length - 1 ; int requiredVersion = highestRequiredSnapshotVersion ; int rhIdx = rehashIndex ; int transferred = 0 ; while ( transferred < MIN_TRANSFERRED_PER_INCREMENTAL_REHASH ) { StateTableEntry < K , N , S > e = oldTable [ rhIdx ] ; while ( e != null ) { if ( e . entryVersion < requiredVersion ) { e = new StateTableEntry < > ( e , stateTableVersion ) ; } StateTableEntry < K , N , S > n = e . next ; int pos = e . hash & newMask ; e . next = newTable [ pos ] ; newTable [ pos ] = e ; e = n ; ++ transferred ; } oldTable [ rhIdx ] = null ; if ( ++ rhIdx == oldCapacity ) { primaryTable = newTable ; incrementalRehashTable = ( StateTableEntry < K , N , S > [ ] ) EMPTY_TABLE ; primaryTableSize += incrementalRehashTableSize ; incrementalRehashTableSize = 0 ; rehashIndex = 0 ; return ; } } primaryTableSize -= transferred ; incrementalRehashTableSize += transferred ; rehashIndex = rhIdx ; }
Runs a number of steps for incremental rehashing.
Runs a number of steps for incremental rehashing .
// we migrate a certain minimum amount of entries from the old to the new table // copy-on-write check for entry //here, the rehash is complete and we release resources and reset fields // sync our local bookkeeping the with official bookkeeping fields