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
/**
	 * Helper method that is the basis for operations that add mappings.
	 */
Helper method that is the basis for operations that add mappings.
Helper method that is the basis for operations that add mappings .
private StateTableEntry<K, N, S> putEntry(K key, N namespace) {
		final int hash = computeHashForOperationAndDoIncrementalRehash(key, namespace);
		final StateTableEntry<K, N, S>[] tab = selectActiveTable(hash);
		int index = hash & (tab.length - 1);
		for (StateTableEntry<K, N, S> e = tab[index]; e != null; e = e.next) {
			if (e.hash == hash && key.equals(e.key) && namespace.equals(e.namespace)) {
				if (e.entryVersion < highestRequiredSnapshotVersion) {
					e = handleChainedEntryCopyOnWrite(tab, index, e);
				}
				return e;
			}
		}
		++modCount;
		if (size() > threshold) {
			doubleCapacity();
		}
		return addNewStateTableEntry(tab, key, namespace, hash);
	}
private StateTableEntry < K , N , S > putEntry ( K key , N namespace ) { final int hash = computeHashForOperationAndDoIncrementalRehash ( key , namespace ) ; final StateTableEntry < K , N , S > [ ] tab = selectActiveTable ( hash ) ; int index = hash & ( tab . length - 1 ) ; for ( StateTableEntry < K , N , S > e = tab [ index ] ; e != null ; e = e . next ) { if ( e . hash == hash && key . equals ( e . key ) && namespace . equals ( e . namespace ) ) { if ( e . entryVersion < highestRequiredSnapshotVersion ) { e = handleChainedEntryCopyOnWrite ( tab , index , e ) ; } return e ; } } ++ modCount ; if ( size ( ) > threshold ) { doubleCapacity ( ) ; } return addNewStateTableEntry ( tab , key , namespace , hash ) ; }
Helper method that is the basis for operations that add mappings.
Helper method that is the basis for operations that add mappings .
// copy-on-write check for entry