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
/**
	 * Creates (combined) copy of the table arrays for a snapshot. This method must be called by the same Thread that
	 * does modifications to the {@link CopyOnWriteStateTable}.
	 */
Creates (combined) copy of the table arrays for a snapshot. This method must be called by the same Thread that
does modifications to the  CopyOnWriteStateTable.
Creates ( combined ) copy of the table arrays for a snapshot . This method must be called by the same Thread that does modifications to the CopyOnWriteStateTable .
@VisibleForTesting
	@SuppressWarnings("unchecked")
	StateTableEntry<K, N, S>[] snapshotTableArrays() {
		synchronized (snapshotVersions) {
			if (++stateTableVersion < 0) {
				throw new IllegalStateException("Version count overflow in CopyOnWriteStateTable. Enforcing restart.");
			}
			highestRequiredSnapshotVersion = stateTableVersion;
			snapshotVersions.add(highestRequiredSnapshotVersion);
		}
		StateTableEntry<K, N, S>[] table = primaryTable;
		if (isRehashing()) {
			final int localRehashIndex = rehashIndex;
			final int localCopyLength = table.length - localRehashIndex;
			StateTableEntry<K, N, S>[] copy = new StateTableEntry[localRehashIndex + table.length];
			System.arraycopy(table, localRehashIndex, copy, 0, localCopyLength);
			table = incrementalRehashTable;
			System.arraycopy(table, 0, copy, localCopyLength, localRehashIndex);
			System.arraycopy(table, table.length >>> 1, copy, localCopyLength + localRehashIndex, localRehashIndex);
			return copy;
		} else {
			return Arrays.copyOf(table, table.length);
		}
	}
@ VisibleForTesting @ SuppressWarnings ( "unchecked" ) StateTableEntry < K , N , S > [ ] snapshotTableArrays ( ) { synchronized ( snapshotVersions ) { if ( ++ stateTableVersion < 0 ) { throw new IllegalStateException ( "Version count overflow in CopyOnWriteStateTable. Enforcing restart." ) ; } highestRequiredSnapshotVersion = stateTableVersion ; snapshotVersions . add ( highestRequiredSnapshotVersion ) ; } StateTableEntry < K , N , S > [ ] table = primaryTable ; if ( isRehashing ( ) ) { final int localRehashIndex = rehashIndex ; final int localCopyLength = table . length - localRehashIndex ; StateTableEntry < K , N , S > [ ] copy = new StateTableEntry [ localRehashIndex + table . length ] ; System . arraycopy ( table , localRehashIndex , copy , 0 , localCopyLength ) ; table = incrementalRehashTable ; System . arraycopy ( table , 0 , copy , localCopyLength , localRehashIndex ) ; System . arraycopy ( table , table . length >>> 1 , copy , localCopyLength + localRehashIndex , localRehashIndex ) ; return copy ; } else { return Arrays . copyOf ( table , table . length ) ; } }
Creates (combined) copy of the table arrays for a snapshot.
Creates ( combined ) copy of the table arrays for a snapshot .
// we guard against concurrent modifications of highestRequiredSnapshotVersion between snapshot and release. // Only stale reads of from the result of #releaseSnapshot calls are ok. This is why we must call this method // from the same thread that does all the modifications to the table. // increase the table version for copy-on-write and register the snapshot // this is just a safety net against overflows, but should never happen in practice (i.e., only after 2^31 snapshots) // consider both tables for the snapshot, the rehash index tells us which part of the two tables we need // for the primary table, take every index >= rhIdx. // for the new table, we are sure that two regions contain all the entries: // [0, rhIdx[ AND [table.length / 2, table.length / 2 + rhIdx[ // we only need to copy the primary table