Javaの世界での仕組み。 コンカレントマップ

基本的なプログラミングの原則は次のとおりです。車輪を再発明しないでください。 しかし、時々、何が起こっているのか、ツールを正しく使用する方法を理解するために、これを行う必要があります。 今日、ConcrurrentHashMapを発明しました。

まず、2つのことが必要です。 2つのテストから始めましょう-最初のテストでは、実装にデータ競合がないことを確認します（実際、既知の不正な実装をテストすることでテストが正しいかどうかを確認する必要があります）、2番目のテストを使用してスループットの観点からパフォーマンスをテストします。

Mapインターフェースのいくつかのメソッドのみを検討してください。

public interface Map<K, V> { V put(K key, V value); V get(Object key); V remove(Object key); int size(); }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

スレッドセーフな正当性テスト

スレッドセーフティテストを十分に包括的に記述することはほとんど不可能です。さらに、 JLSの第17章で定義されているすべての側面を考慮する必要があります。さらに、テストはハードウェアメモリまたはJVM実装のモデルに大きく依存します。

スレッドセーフな正当性テストでは、データの不整合を検出するためにコードを実行するjcstressなどの既製のストレステストライブラリの1つを使用します。 jcstressはまだ実験的としてマークされていますが、より良い選択です。 独自の並列処理テストを作成するのが難しい理由-シピレフの講義を参照してください。

jstressを実行するには、jstress jcstress- gradle -pluginを使用します。 完全なソースコードはhow-it-works-concurrent-mapにあります 。

 public class ConcurrentMapThreadSafetyTest { @State public static class MapState { final Map<String, Integer> map = new HashMap<>(3); } @JCStressTest @Description("Test race map get and put") @Outcome(id = "0, 1", expect = ACCEPTABLE, desc = "return 0L and 1L") @Outcome(expect = FORBIDDEN, desc = "Case violating atomicity.") public static class MapPutGetTest { @Actor public void actor1(MapState state, LongResult2 result) { state.map.put("A", 0); Integer r = state.map.get("A"); result.r1 = (r == null ? -1 : r); } @Actor public void actor2(MapState state, LongResult2 result) { state.map.put("B", 1); Integer r = state.map.get("B"); result.r2 = (r == null ? -1 : r); } } @JCStressTest @Description("Test race map check size") @Outcome(id = "2", expect = ACCEPTABLE, desc = "size of map = 2 ") @Outcome(id = "1", expect = FORBIDDEN, desc = "size of map = 1 is race") @Outcome(expect = FORBIDDEN, desc = "Case violating atomicity.") public static class MapSizeTest { @Actor public void actor1(MapState state) { state.map.put("A", 0); } @Actor public void actor2(MapState state) { state.map.put("B", 0); } @Arbiter public void arbiter(MapState state, IntResult1 result) { result.r1 = state.map.size(); } } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

最初のMapPutGetTestテストでは、それぞれメソッドactor1とactor2を同時に実行する2つのスレッドがあります。両方ともマップに値を入れてチェックバックします。データの競合がなければ、両方のスレッドは指定された値を見るはずです。

2番目のMapSizeTestでは、マップに2つの異なるキーを同時に配置し、サイズを確認した後-データの競合がない場合-期待される結果は2になります。

テストの正確さを検証するには、明らかにスレッドセーフなHashMapでテストを実行します。原子性の違反を観察する必要があります。 スレッドセーフなConcurrentHashMapでテストを実行した場合、一貫性の違反は見られません。

HashMapの結果：

 [FAILED] ru.skuptsov.concurrent.map.test.ConcurrentMapTest.MapPutGetTest Observed state Occurrences Expectation Interpretation -1, 1 293,867 FORBIDDEN Case violating atomic 0, -1 282,190 FORBIDDEN Case violating atomic 0, 1 28,013,763 ACCEPTABLE return 0 and 1 [FAILED] ru.skuptsov.concurrent.map.test.ConcurrentMapTest.MapSizeTest Observed state Occurrences Expectation Interpretation 1 1,434,783 FORBIDDEN size of map = 1 race 2 11,733,097 ACCEPTABLE size of map = 2
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

スレッドセーフHashMapでは、統計的な量の一貫性のない結果が見られ、両方のテストが失敗しました。

スレッドセーフなConcurrentHashMapの結果：

 [OK] ru.skuptsov.concurrent.map.test.ConcurrentMapTest.MapPutGetTest Observed state Occurrences Expectation Interpretation 0, 1 20,195,000 ACCEPTABLE [OK] ru.skuptsov.concurrent.map.test.ConcurrentMapTest.MapSizeTest Observed state Occurrences Expectation Interpretation 2 6,573,730 ACCEPTABLE size of map = 2
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

ConcurrentHashMapはテストに合格しました。少なくとも、テストでいくつかの単純な同時実行の問題を検出できることを認識できます。 Collection.synchronizedMapとHashTableについても同じ結果を確認できます。

最も適切なConcurrentHashMapの試行

最初の単純なアプローチは、内部構造（バケットの配列）への各アクセスを単純に同期することです。

実際、送信されたマッププロバイダーに対して並列ラッパーを作成できます。 Java.util.Collections.synchronizedMap、Hashtable、およびGuava synchronizedMultimapは同じことを行います。

 public class SynchrinizedHashMap<K, V> extends BaseMap<K, V> implements Map<K, V>, IMap<K, V> { private final Map<K, V> provider; private final Object monitor; public SynchronizedHashMap(Map<K, V> provider) { this.provider = provider; monitor = this; } @Override public V put(K key, V value) { synchronized (monitor) { return provider.put(key, value); } } @Override public V get(Object key) { synchronized (monitor) { return provider.get(key); } } @Override public int size() { synchronized (monitor) { return provider.size(); } } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

ドキュメントによると、不揮発性マッププロバイダーへの変更はスレッド間で表示されます。

第二に、同期メソッドが終了すると、同じオブジェクトに対する同期メソッドの後続の呼び出しと発生前の関係を自動的に確立します。 これにより、オブジェクトの状態の変更がすべてのスレッドに表示されることが保証されます。

私たちの最も単純な実装は並列テストに合格しますが、価格はどうですか？ 異なるキーを操作する場合でも、各メソッドは一度に1つのスレッドしか持つことができないため、マルチスレッドの負荷の下で高いパフォーマンスを期待することはできません。 それを測定しましょう。

性能試験

パフォーマンステストでは、 jmhライブラリを使用します。

 @State(Scope.Thread) @Warmup(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS) @Measurement(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS) @Fork(3) @BenchmarkMode(Mode.AverageTime) @OutputTimeUnit(MICROSECONDS) public class ConcurrentMapBenchmark { private Map<Integer, Integer> map; @Param({"concurrenthashmap", "hashtable", "synchronizedhashmap"}) private String type; @Param({"1", "10"}) private Integer writersNum; @Param({"1", "10"}) private Integer readersNum; private final static int NUM = 1000; @Setup public void setup() { switch (type) { case "hashtable": map = new Hashtable<>(); break; case "concurrenthashmap": map = new ConcurrentHashMap<>(); break; case "synchronizedhashmap": map = new SynchronizedHashMap<>(new HashMap<>()); break; } } @Benchmark public void test(Blackhole bh) throws ExecutionException, InterruptedException { List<CompletableFuture> futures = new ArrayList<>(); for (int i = 0; i < writersNum; i++) { futures.add(CompletableFuture.runAsync(() -> { for (int j = 0; j < NUM; j++) { map.put(j, j); } })); } for (int i = 0; i < readersNum; i++) { futures.add(CompletableFuture.runAsync(() -> { for (int j = 0; j < NUM; j++) { bh.consume(map.get(j)); } })); } CompletableFuture.allOf(futures.toArray(new CompletableFuture[1])).get(); } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

SynchronizedHashMapのパフォーマンスはjava-s HashTableとほぼ同等であり、ConcurrentHashMapよりも2倍悪いことを確認しました。 パフォーマンスを改善してみましょう。

ロックストライピングConcurrentHashMapの試行

最初の改善は、マップ全体へのアクセスをブロックするのではなく、スレッドが同じバケットにアクセスする場合にのみアクセスを同期する方がよいという考えに基づいている可能性があります（バケットインデックス= key.hashCode（）％array.length）。 この方法は、ロックストライピングまたはファイングレイン同期と呼ばれます。TheArt of Multiprocessor Programmingを参照してください。

バケットの配列の場合、ロックの配列が必要です。起動時には、ロックの配列のサイズは配列の内部サイズと等しくなければなりません。これは、2つのロッカが配列の1つのバケットを担当する状況が望ましくないため重要です。

簡単にするために、不変のバケット配列を持つマップを考えます-これは、初期容量を拡張できないことを意味します（N >> initialCapacityの場合、O（1）マップを失うと、取得要素の挿入が保証されます。また、loadFactorも必要ありません） 拡張可能な並行マップは、別の大きなトピックです。

 public class LockStripingArrayConcurrentHashMap<K, V> extends BaseMap<K, V> implements Map<K, V> { private final AtomicInteger count = new AtomicInteger(0); private final Node<K, V>[] buckets; private final Object[] locks; @SuppressWarnings({"rawtypes", "unchecked"}) public LockStripingArrayConcurrentHashMap(int capacity) { locks = new Object[capacity]; for (int i = 0; i < locks.length; i++) { locks[i] = new Object(); } buckets = (Node<K, V>[]) new Node[capacity]; } @Override public int size() { return count.get(); } @Override public V get(Object key) { if (key == null) throw new IllegalArgumentException(); int hash = hash(key); synchronized (getLockFor(hash)) { Node<K, V> node = buckets[getBucketIndex(hash)]; while (node != null) { if (isKeyEquals(key, hash, node)) { return node.value; } node = node.next; } return null; } } @Override public V put(K key, V value) { if (key == null || value == null) throw new IllegalArgumentException(); int hash = hash(key); synchronized (getLockFor(hash)) { int bucketIndex = getBucketIndex(hash); Node<K, V> node = buckets[bucketIndex]; if (node == null) { buckets[bucketIndex] = new Node<>(hash, key, value, null); count.incrementAndGet(); return null; } else { Node<K, V> prevNode = node; while (node != null) { if (isKeyEquals(key, hash, node)) { V prevValue = node.value; node.value = value; return prevValue; } prevNode = node; node = node.next; } prevNode.next = new Node<>(hash, key, value, null); count.incrementAndGet(); return null; } ... } } private boolean isKeyEquals(Object key, int hash, Node<K, V> node) { return node.hash == hash && node.key == key || (node.key != null && node.key.equals(key)); } private int hash(Object key) { return key.hashCode(); } private int getBucketIndex(int hash) { return hash % buckets.length; } private Object getLockFor(int hash) { return locks[hash % locks.length]; } private static class Node<K, V> { final int hash; K key; V value; Node<K, V> next; Node(int hash, K key, V value, Node<K, V> next) { this.hash = hash; this.key = key; this.value = value; this.next = next; } } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

クラスのすべてのフィールドがfinalであることが重要です-これは安全な公開を保証し、オブジェクトの最終的な作成まで誰もメソッドを呼び出さないことです-これはコンストラクタで初期化されているため重要です。

ソースコードはこちらにあります 。

テスト結果：





ファイングレイン同期の実装は、一般的なロックよりも優れていることがわかります。 結果は、1つのリーダーと1つのライターで、ConcurrentHashMapと比較してほぼ同じですが、スレッドの数が増えると、特にリーダーが多い場合は違いが大きくなります。

ロックなしの同時ハッシュマップ試行

正直に言うと、同期はスレッドを順次キューに入れ、別のスレッドが終了するのを待つため、並列プログラミングの方法ではありません。 また、システムコンテキストを同期するための追加コストは、待機中のスレッドの数とともに増加しますが、必要なのは、マップキーの値を変更するための少数の命令だけです。

新しいハッシュマップの実装に関するいくつかの要件を定義しますが、理論的には実装を改善するはずです。 また、要件は次のとおりです。

1. 異なるキー（書き込みまたは読み取り）で動作する2つのスレッドがある場合、それらの間の同期は必要ありません（Javaではワードティアリングは許可されません-配列の2つの異なるフィールドへのアクセスはスレッドセーフです）
2. 複数のスレッドが同じキー（書き込みおよび読み取り）で動作している場合、操作を並べ替える必要はありません（ 最新のキャッシュの構造の問題の原因について ）。スレッド間の時々前の保証が必要です。フロー。 しかし、読み取りストリームをブロックして、書き込みストリームが完了するのを待ちたくはありません。
3. 書き込みストリームが1つもない場合、複数のリーダーを1つのキーでブロックすることは望ましくありません。

ポイント2と3に集中しましょう。実際、（1）バケットの揮発性読み取り配列を作成し、（2）次のノードの揮発性読み取りでバケット内に移動できる場合、マップ読み取り操作を完全にロックフリーにすることができます。ノード自体の目的の揮発性の読み取り値が見つかるまでリンクリスト。

（2）の場合、次のフィールドと値フィールドをNodeでvolatileとしてマークするだけです。

（1）揮発性配列のようなものはありません、配列が揮発性として宣言されていても、これは要素の読み取りまたは書き込み時に揮発性のセマンティクスを提供しませんが、配列のk番目の要素にアクセスするには外部同期が必要で、揮発性はそれ自体です配列参照。 この目的でAtomicReferenceArrayを使用できますが、Object []配列のみを受け入れます。 別の方法として、揮発性配列の読み取りとロックフリーの書き込みにUnsafeを使用することを検討してください。 同じメソッドがAtomicReferenceArrayとConcurrentHashMapで使用されます。

 @SuppressWarnings("unchecked") // read array value by index private <K, V> Node<K, V> volatileGetNode(int i) { return (Node<K, V>) U.getObjectVolatile(buckets, ((long) i << ASHIFT) + ABASE); } // cas set array value by index private <K, V> boolean compareAndSwapNode(int i, Node<K, V> expectedNode, Node<K, V> setNode) { return U.compareAndSwapObject(buckets, ((long) i << ASHIFT) + ABASE, expectedNode, setNode); } private static final sun.misc.Unsafe U; // Node[] header shift private static final long ABASE; // Node.class size shift private static final int ASHIFT; static { try { // get unsafe by reflection - it is illegal to use not in java lib Constructor<Unsafe> unsafeConstructor = Unsafe.class.getDeclaredConstructor(); unsafeConstructor.setAccessible(true); U = unsafeConstructor.newInstance(); } catch (NoSuchMethodException | InstantiationException | InvocationTargetException | IllegalAccessException e) { throw new RuntimeException(e); } Class<?> ak = Node[].class; ABASE = U.arrayBaseOffset(ak); int scale = U.arrayIndexScale(ak); ASHIFT = 31 - Integer.numberOfLeadingZeros(scale); }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

volatile getNodeでは、ロックなしで安全に値を読み取ることができます。

ロックフリーのV get（オブジェクトキー）を書きましょう：

 public V get(Object key) { if (key == null) throw new IllegalArgumentException(); int hash = hash(key); Node<K, V> node; // volatile read of bucket head at hash index if ((node = volatileGetNode(getBucketIndex(hash))) != null) { // check first node if (isKeyEquals(key, hash, node)) { return node.value; } // walk through the rest to find target node while ((node = node.next) != null) { if (isKeyEquals(key, hash, node)) return node.value; } } return null; }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

最初の試みは、ロックプールを使用した大きなメモリオーバーヘッドでした。実際、追加のメモリなしで同じきめの細かいアプローチを使用できます。バケット内の最初のノードが存在する場合は、ロックするだけです。 存在しない場合-存在しない要素でブロックすることはできず、ヘッダーノードを設定するためのロックフリーメソッドが必要です-既にこのメソッドを記述しています-compareAndSwapNodeメソッド。

 @Override public V put(K key, V value) { if (key == null || value == null) throw new IllegalArgumentException(); int hash = hash(key); // no resize in this implementation - so the index will not change int bucketIndex = getBucketIndex(hash); // cas loop trying not to miss while (true) { Node<K, V> node; // if bucket is empty try to set new head with cas if ((node = volatileGetNode(bucketIndex)) == null) { if (compareAndSwapNode(bucketIndex, null, new Node<>(hash, key, value, null))) { // if we succeed to set head - then break and return null count.increment(); break; } } else { // head is not null - try to find place to insert or update under lock synchronized (node) { // check if node have not been changed since we got it // otherwise let's go to another loop iteration if (volatileGetNode(bucketIndex) == node) { V prevValue = null; Node<K, V> n = node; while (true) { ... simply walk through list under lock and update or insert value... } return prevValue; } } } } return null; }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

完全なソースコードはこちら 。

そのパフォーマンスをテストしましょう：

場合によっては、ConcurrentHashMapよりも優れていますが、これは完全に正直な比較ではありません。 ConcurrentHashMapは起動時にテーブルの遅延初期化を行い、境界要素threshold = initialCapacity * loadFactorで少なくとも1回サイズを変更するためです。 初期化された要素initialCapacityを使用してテストを再度実行すると、 = N（= N / 6）、結果はわずかに異なります：

これは、ConcurrentHashMapでバケット配列の初期サイズが増加し、バケット内のリンクリストの長さが減少するため、キーによる要素の取得に費やされる時間が短くなるために発生しました。

ConcurrentHashMapのように、完全な非ブロックデータ構造を取得していないことに注意する必要がありますが、必要なのはロックのないリンクリストだけですが、データのサイズ変更と変更を同時に行うことで、このタスクはそれほど単純ではありません- ここを読んでください 。

オリジナルのJava 8 ConcurrentHashMapには、言及していない多くの改善点があります。例えば：

1. 初めて使用する前のメモリフットプリントを最小化するバケットテーブルの遅延初期化
2. 同時サイズ変更バケット配列
3. LongAdderを使用して要素をカウントします。
4. 特殊なタイプのノード（1.8以降）-TreeBins、バケット内のリストがTREEIFY_THRESHOLD = 8より長くなる場合-バケットは、キーによる最悪の検索でバランスツリーになります（O（log（Nbucket_size）））

Java 1.8でのConcurrentHashMapの実装が1.7から大幅に変更されたことに注意してください。 1.7では、セグメントの数が並列性のレベルに等しいセグメントのアイデアでした。 Java 8では、バケット配列は単一の配列です。