狠狠撸

ConcurrentHashMap
Code Reading
角田直行 <kakuda@gmail.com>

Agenda

? 準备
? Overview
? 読む
? まとめ

準备するもの
? Eclipse

? ソース

? ConcurrentHashMap.java

? テストケース

? JSR166TestCase.java

? ConcurrentHashMapTest.java

? ConcurrentHashMapに関する参照資料

? 気合い

ソースの取得場所

? ConcurrentHashMap.javaはJDKより
? テストケースはDoug Leaさんのサイトより
http://gee.cs.oswego.edu/dl/concurrency-interest/

参照した資料

? JavaDoc
http://java.sun.com/javase/ja/6/docs/ja/api/java/util/concurrent/
ConcurrentHashMap.html

? IBM developerWorks:優れたHashMapの構築
http://www.ibm.com/developerworks/jp/java/library/j-jtp08223/

? ITアーキテクト J2SE 5.0の新機能
http://www.itarchitect.jp/technology_and_programming/-/24161-3.html

? Servlet Gardern@はてな
http://d.hatena.ne.jp/yokolet/20071005

Overview

? Constants: 6つ

? Field: 6つ

? ユーティリティ: 2つ(hash(), segmentFor())

? インナークラス: 2つ(HashEntry, Segment)

? Public メソッド: 24コ(Constructorは5つ)

? Iteratorサポート: 8つ

? Serializationサポート: 2つ(writeObject(), readObject())

多分重要なトコ

? Constants: 6つ

? Field: 6つ






今回は割愛するトコ

? Constants: 6つ

? Field: 6つ






コンストラクタ
public ConcurrentHashMap() {
this(DEFAULT_INITIAL_CAPACITY,
DEFAULT_LOAD_FACTOR,
DEFAULT_CONCURRENCY_LEVEL);
}

public ConcurrentHashMap(int initialCapacity) {
this(initialCapacity,
DEFAULT_LOAD_FACTOR,
}

public ConcurrentHashMap(int initialCapacity, float loadFactor) {
this(initialCapacity,
loadFactor,
}


/** デフォルトの初期容量 */
static final int DEFAULT_INITIAL_CAPACITY = 16;
/** デフォルトの負荷係数 */
static final float DEFAULT_LOAD_FACTOR = 0.75f;
/** デフォルトの並行処理レベル */
static final int DEFAULT_CONCURRENCY_LEVEL = 16;

/** 最大Segment数?1を16ビット分左にシフトしてるので65536 */
static final int MAX_SEGMENTS = 1 << 16;
...
public ConcurrentHashMap(int initialCapacity,
float loadFactor, int concurrencyLevel) {
if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)
throw new IllegalArgumentException();

// 並行処理レベルが最大セグメント数を超えてはならない
if (concurrencyLevel > MAX_SEGMENTS)
concurrencyLevel = MAX_SEGMENTS;
...

Segment数の決定
/** どのSegmentにあるかを特定するためのマスク値 */
final int segmentMask;
/** どのSegmentにあるかを特定するためのシフト値 */
final int segmentShift;

// SegmentとHashエントリの最適な引数を見つける
int sshift = 0;
int ssize = 1;
while (ssize < concurrencyLevel) {
++sshift;
ssize <<= 1;
}
// currencyLevelが16(デフォルト)の時、sshiftは4、ssizeは16になる
segmentShift = 32 - sshift; // 28(32は32bitのこと?)
segmentMask = ssize - 1; // 15(0b0111)
this.segments = Segment.newArray(ssize);

Segmentの生成
/** 最大容量?1を30ビット分左にシフトしてるので1073741824 */
static final int MAXIMUM_CAPACITY = 1 << 30;

// Segmentの初期容量の算出
if (initialCapacity > MAXIMUM_CAPACITY)
initialCapacity = MAXIMUM_CAPACITY;
int c = initialCapacity / ssize;
if (c * ssize < initialCapacity)
++c;
int cap = 1;
while (cap < c)
cap <<= 1;

for (int i = 0; i < this.segments.length; ++i)
this.segments[i] = new Segment<K,V>(cap, loadFactor);
}

Segmentとは

? ConcurrentHashMap.javaにて定義されて
いるインナークラス

? 1つのConcurrentHashMapインスタンス
に複数(デフォルト16)のSegmentを持つ

? ハッシュテーブルの特別バージョン
? ReentrantLockのサブクラス

Segmentのコンストラクタ
static final class Segment<K,V> extends ReentrantLock implements Serializable {
...
transient int threshold;
// 各SegmentにHashEntryの配列を保持
transient volatile HashEntry<K,V>[] table;
...
Segment(int initialCapacity, float lf) {
loadFactor = lf;
setTable(HashEntry.<K,V>newArray(initialCapacity));
}
...
void setTable(HashEntry<K,V>[] newTable) {
threshold = (int)(newTable.length * loadFactor);
table = newTable;
}

HashEntryとは

? ConcurrentHashMap.javaにて定義されて
いるインナークラス

? 1つのSegmentインスタンスに1つ以上の
HashEntryを持つ

? フィールドにkey, hash値, value, next(次の
HashEntry)を持つ

? key, hash, nextは?nal

HashEntry
static final class HashEntry<K,V> {
final K key;
final int hash;
volatile V value;
final HashEntry<K,V> next;
HashEntry(K key, int hash, HashEntry<K,V> next, V value) {
this.key = key;
this.hash = hash;
this.next = next;
this.value = value;
}
@SuppressWarnings("unchecked")
static final <K,V> HashEntry<K,V>[] newArray(int i) {
return new HashEntry[i];
}
}

厂别驳尘别苍迟、贬补蝉丑贰苍迟谤测の构造


ConcurrentHashMap


Segment

Segment

ConcurrentHashMap
?
?
?
Segment


Hash
Entry
Segment

Segment

ConcurrentHashMap
?
?
?
Segment


Hash
Entry
Segment

Hash
Entry
Segment

ConcurrentHashMap
?
?
?
Segment


Hash Hash
Entry
Segment
Entry

Hash
Entry
Segment

ConcurrentHashMap
?
?
?
Segment


Hash Hash
Entry
Segment
Entry

Hash Hash
Entry
Segment
Entry

ConcurrentHashMap
?
?
?
Segment


Hash Hash Hash
Entry
Segment
Entry Entry

Hash Hash
Entry
Segment
Entry

ConcurrentHashMap
?
?
?
Segment


Hash Hash Hash
Entry
Segment
Entry Entry
実際のチェーン数は
1～2個くらい
Hash Hash
Entry
Segment
Entry

ConcurrentHashMap
?
?
?
Segment


Hash Hash Hash
Entry
Segment
Entry Entry
1～2個くらい
Hash Hash
Entry
Segment
Entry

ConcurrentHashMap
?
?
?
Hash
Entry
Segment


Hash Hash Hash
Entry
Segment
Entry Entry
1～2個くらい
Hash Hash
Entry
Segment
Entry

ConcurrentHashMap
?
? 全Segmentが埋まったらSegment
? のサイズは2倍に拡張される
Hash
Entry
Segment

ConcurrentHashMap#get

public V get(Object key) {
// 与えられたハッシュ値を元にちゃんとしたハッシュ値を求める
int hash = hash(key.hashCode());
// ハッシュ値に該当するセグメントを見つけSegment#getをcall
return segmentFor(hash).get(key, hash);
}

他の主要メソッドも
似たような処理
public boolean remove(Object key, Object value) {
public boolean containsKey(Object key) {
if (value == null)
return false;
return segmentFor(hash).containsKey(key, hash);
return segmentFor(hash).remove(key, hash, value) !=
}
null;
public V put(K key, V value) {
}
if (value == null)
public boolean replace(K key, V oldValue, V newValue) {
throw new NullPointerException();
if (oldValue == null || newValue == null)
return segmentFor(hash).put(key, hash, value, false);
}
return segmentFor(hash).replace(key, hash, oldValue,
public V putIfAbsent(K key, V value) {
newValue);
if (value == null)
}
public V replace(K key, V value) {
if (value == null)
return segmentFor(hash).put(key, hash, value, true);
}
return segmentFor(hash).replace(key, hash, value);
}

ConcurrentHashMap#hash

何やっているかわかりません???orz
private static int hash(int h) {
// Spread bits to regularize both segment and index locations,
// using variant of single-word Wang/Jenkins hash.
h += (h << 15) ^ 0xffffcd7d;
h ^= (h >>> 10);
h += (h << 3);
h ^= (h >>> 6);
h += (h << 2) + (h << 14);
return h ^ (h >>> 16);
}

HashMap#hashよりBetterな実装に置き換えてる

http://www.concentric.net/ Ttwang/tech/inthash.htm が参考ソース？

いいハッシュ値を返すと???

Segment

Segment

ConcurrentHashMap
?
?
?
Segment


Segment

(“1”, “a”)
Segment

PUT ConcurrentHashMap
?
?
?
Segment


(“1”, “a”) Segment

Segment

ConcurrentHashMap
?
?
?
Segment



(“2”, “b”)
Segment

?
?
?
Segment



(“2”, “b”) Segment

ConcurrentHashMap
?
?
?
Segment



(“3”, “c”)

?
?
?
Segment




ConcurrentHashMap
?
?
?
(“3”, “c”) Segment




ちゃんとバラけてくれる！
ConcurrentHashMap
?
?
?
(“3”, “c”) Segment

悪いハッシュ値を返すと???

Segment

Segment

ConcurrentHashMap
?
?
?
Segment


Segment

(“1”, “a”)
Segment

?
?
?
Segment



Segment

ConcurrentHashMap
?
?
?
Segment



(“2”, “b”)
Segment

?
?
?
Segment


(“1”, “a”) (“2”, Segment
“b”)

Segment

ConcurrentHashMap
?
?
?
Segment


(“1”, “a”) (“2”, Segment
“b”)

(“3”, “c”)
Segment

?
?
?
Segment


(“1”, “a”) (“2”, Segment
“b”) (“3”, “c”)

Segment

ConcurrentHashMap
?
?
?
Segment


(“1”, “a”) (“2”, Segment
“b”) (“3”, “c”)

Segment
偏ってしまい、
ConcurrentHashMap
?
すごく効率が悪くなる！
?
?
Segment

Segment#get
V get(Object key, int hash) {
if (count != 0) { // read-volatile
// hashに相当するHashEntryの要素を取り出す
HashEntry<K,V> e = getFirst(hash);
while (e != null) {
if (e.hash == hash && key.equals(e.key)) {
// HashEntryのハッシュ値とキーが両方とも合えば
V v = e.value;
if (v != null) return v;
return readValueUnderLock(e); // recheck
}
e = e.next;
}
}
return null;
}

Segment#containsKey
基本Segment#getと同じ
boolean containsKey(Object key, int hash) {
while (e != null) {
if (e.hash == hash && key.equals(e.key))
return true;
e = e.next;
}
}
return false;
}

Segment#containsValue
boolean containsValue(Object value) {
HashEntry<K,V>[] tab = table; // 変化する恐れがあるので作業用に置いてる？
int len = tab.length;
for (int i = 0 ; i < len; i++) {
for (HashEntry<K,V> e = tab[i]; e != null; e = e.next) {
V v = e.value;
if (v == null) // recheck
v = readValueUnderLock(e); // 理屈上こういうことがありうるらしい
if (value.equals(v))
return true;
}
}
}
return false;
}

Segment#put

V put(K key, int hash, V value, boolean onlyIfAbsent) {
lock(); // HashEntryの更新系(put, remove, replace...)はロックが必要
try {
int c = count;
if (c++ > threshold) // ensure capacity
rehash(); // サイズがしきい値を超えたら各要素のハッシュ値を再計算
HashEntry<K,V>[] tab = table;
int index = hash & (tab.length - 1);
HashEntry<K,V> first = tab[index];
HashEntry<K,V> e = first;
while (e != null && (e.hash != hash || !key.equals(e.key))) // どういうケース?
e = e.next;
...

Segment#put
V oldValue;
if (e != null) {
oldValue = e.value;
if (! onlyIfAbsent) e.value = value;
}
else { // 基本はここを通る
oldValue = null;
++modCount; // HashEntryの構造が変更された回数を増やす
tab[index] = new HashEntry<K,V>(key, hash, first, value);
count = c; // write-volatile
}
return oldValue;
} finally {
unlock();
}
}

Segment#remove
前半部分はSegment#putと大体同じ
V remove(Object key, int hash, Object value) {
lock();
try {
int c = count - 1; // サイズを1つ減らす
int index = hash & (tab.length - 1);
HashEntry<K,V> first = tab[index];
HashEntry<K,V> e = first;
while (e != null && (e.hash != hash || !key.equals(e.key)))
e = e.next;
V oldValue = null;
...

Segment#remove
if (e != null) {
V v = e.value;
if (value == null || value.equals(v)) {
oldValue = v;
++modCount;
HashEntry<K,V> newFirst = e.next;
for (HashEntry<K,V> p = first; p != e; p = p.next)
newFirst = new HashEntry<K,V>(p.key, p.hash, newFirst, p.value);
tab[index] = newFirst;
count = c; // write-volatile
}
}
return oldValue;
} finally {
unlock();
}
}

Segment#remove

? HashEntryにはnextという次の要素を表す
フィールドがある

? removeが行われるとnextの付け替えが発生
する

? nextフィールドは?nalなので削除した要素よ
り前を作り直さなければならない

A B C D E
Segment
Next = B Next = C Next = D Next = E Next = null

A B C D E
Segment
Next = B Next = C Next = D Next = E Next = null

remove(“C”)

A B D E
Segment
Next = B Next = C Next = E Next = null

Nextが?nalなので
A B Dに変更できない！D E
Segment

Nextが?nalなので
A B Dに変更できない！D E
Segment

AとBは作り直し

Segment#rehash
put時にサイズがcapacityを超えた時に発生
void rehash() {
HashEntry<K,V>[] oldTable = table;
int oldCapacity = oldTable.length;
if (oldCapacity >= MAXIMUM_CAPACITY)
return;

/*
* Reclassify nodes in each list to new Map. Because we are
* using power-of-two expansion, the elements from each bin
* must either stay at same index, or move with a power of two
* offset. We eliminate unnecessary node creation by catching
...

コメントの超意訳

? 各リストのノードを新しいMapに再分類する。2つの強力
な拡張(SegmentとHashEntry)を使っているため、同じ
indexを保つか2つのo?setを同時に移動させなければなら
ない。nextフィールドは変わらないので、古いノードは再
利用でき不必要なノードを作らなくてすむ。統計的に、デ
フォルトのthresholdだとテーブルを2倍にする時にクロー
ンする必要があるのは約1/6。置き換わる古いノードは
readerスレッドがすぐにテーブルを走査することで参照さ
れなくなりGC対象となる。

Segment#clear
void clear() {
if (count != 0) {
lock();
try {
for (int i = 0; i < tab.length ; i++)
tab[i] = null;
++modCount;
count = 0; // write-volatile
} finally {
unlock();
}
}
}

Segment#replace
V replace(K key, int hash, V newValue) {
lock();
try {
e = e.next;

V oldValue = null;
if (e != null) {
oldValue = e.value;
e.value = newValue;
}
return oldValue;
} finally {
unlock();
}
}

Segment#replace その2
boolean replace(K key, int hash, V oldValue, V newValue) {
lock();
try {
e = e.next;

boolean replaced = false;
if (e != null && oldValue.equals(e.value)) {
replaced = true;
e.value = newValue;
}
return replaced;
} finally {
unlock();
}
}

ConcurrentHashMap#isEmpty

public boolean isEmpty() {
final Segment<K,V>[] segments = this.segments;
int[] mc = new int[segments.length];
int mcsum = 0;
for (int i = 0; i < segments.length; ++i) {
if (segments[i].count != 0)
return false;
else
// 全segmentのサイズが0の時、それぞれの変更回数の和を計算する
mcsum += mc[i] = segments[i].modCount;
}
...

ConcurrentHashMap#isEmpty

if (mcsum != 0) {
// 何かしらの変更 (ABA問題) があった場合
if (segments[i].count != 0 || mc[i] != segments[i].modCount)
return false;
}
}
return true;
}

isEmptyのコメント超意訳

? いつの時点でもテーブルが空にならなかった場合、ある
segmentの一要素が追加され別で走査中に削除されるとい
うABA問題(「Java並行処理プログラミング」15章参照)を
避けるために各segmentのmodCountを追跡する。他に
ABA問題の影響を受ける可能性があるsize()や
containsValue()メソッドでも同様にmodCountを使ってい
る。

ConcurrentHashMap#size

? まずはロック無しで2回までトライする
? 数えている間に、別で変更処理が行われた
らやり直す

? 2回とも邪魔されてしまったら全Segmentに
ロックをかけて数える

/** size()とcontainsValue()で使われる */
static final int RETRIES_BEFORE_LOCK = 2;

public int size() {
long sum = 0;
long check = 0;
// Try a few times to get accurate count. On failure due to
// continuous async changes in table, resort to locking.
for (int k = 0; k < RETRIES_BEFORE_LOCK; ++k) {
check = 0;
sum = 0;
int mcsum = 0;
...

sum += segments[i].count;
}
if (mcsum != 0) { // 初期状態から変更が行われた
check += segments[i].count;
if (mc[i] != segments[i].modCount) { // 変更された
check = -1; // force retry(「Java並行処理プログラミング」P269)
break;
}
}
}
if (check == sum) break;
}

if (check != sum) { // Resort to locking all segments
sum = 0;
for (int i = 0; i < segments.length; ++i) //全Segmentをロック
segments[i].lock();
for (int i = 0; i < segments.length; ++i)
sum += segments[i].count;
segments[i].unlock();
}
if (sum > Integer.MAX_VALUE)
return Integer.MAX_VALUE;
else
return (int)sum;
}

ConcurrentHashMap#containsValue

ConcurrentHashMap#sizeと大体同じ
public boolean containsValue(Object value) {
if (value == null)
// See explanation of modCount use above

// Try a few times without locking
for (int k = 0; k < RETRIES_BEFORE_LOCK; ++k) {
int sum = 0;
int mcsum = 0;
...

int c = segments[i].count; // メモリを同期化
if (segments[i].containsValue(value)) return true;
}
boolean cleanSweep = true;
if (mcsum != 0) { // 初期状態から変更された
int c = segments[i].count; // メモリを同期化
if (mc[i] != segments[i].modCount) { // 変更されたのでやり直し
cleanSweep = false;
break;
}
}
}
if (cleanSweep) return false;
}

// Resort to locking all segments
segments[i].lock();
boolean found = false;
try {
if (segments[i].containsValue(value)) { // Segment#containsValue
found = true;
break;
}
}
} finally {
segments[i].unlock();
}
return found;
}

まとめ

? ConcurrentHashMapは、Segmentと
HashEntryというデータ構造が肝

? 高速化のために色んなことをやってる
? ?nalでread時はロックしないように
? hash値計算でちゃんとばらけさせる
? ロック処理を最小限に抑える

ちなみに

? ConcurrentHashMapより速い
NonBlockingHashMapというのがある

? Azul SystemsのCli? Click氏作 (DN$&6;OK)$P$&$QA/R$5F*.

:L$@8M+0 :D$@8M+0
77768)*+.-./01.6591

!' !'

&< HIE"" &<

JKDE""
&' &'

DE9F.G.05
DE9F.G.05

HIE><
:< :<

:'
JKDE>< :'

< < HI

' '
JKD
: & ! ; < = > ? : & ! ; < = > ?
@AB08C. @AB08C.

$
#$$$$%&''!$()*+$,-./01.2$3456
!"

詳しく知りたい人は

? JavaOne資料「A Lock-Free Hash Table」
http://www.azulsystems.com/events/javaone_2007/2007_LockFreeHash.pdf

? Blog記事「A Non-Blocking HashTable」
http://blogs.azulsystems.com/cli?/2007/03/a_nonblocking_h.html

? Highly Scalable Java
http://sourceforge.net/projects/high-scale-lib

ご清聴ありがとう
ございました！

狠狠撸

ConcurrentHashMap Code Reading

More Related Content

ConcurrentHashMap Code Reading