@Test public void concurrentPuts() throws InterruptedException { Integer k = 1; Integer v1 = 1; Integer v2 = 2; Integer v3 = 3; final Integer[] results = new Integer[3]; List<Thread> threadList = new ArrayList<>(results.length); threadList.add(new Thread(() -> results[0] = testMap.put(k, v1, OakCommonBuildersFactory.DEFAULT_INT_SERIALIZER::deserialize))); threadList.add(new Thread(() -> results[1] = testMap.put(k, v2, InternalOakMapTest::slowDeserialize))); threadList.add(new Thread(() -> results[2] = testMap.put(k, v3, OakCommonBuildersFactory.DEFAULT_INT_SERIALIZER::deserialize))); runThreads(threadList); Assert.assertNull(results[0]); Assert.assertNotEquals(results[1], results[2]); }
V put(K key, V value, OakTransformer<V> transformer) { if (key == null || value == null) { throw new NullPointerException(); } ThreadContext ctx = getThreadContext(); for (int i = 0; i < MAX_RETRIES; i++) { Chunk<K, V> c = findChunk(key); c.lookUp(ctx, key); if (ctx.isValueValid()) { Result res = valueOperator.exchange(c, ctx, value, transformer, valueSerializer, valuesMemoryManager, this); if (res.operationResult == ValueUtils.ValueResult.TRUE) { return (V) res.value; } continue; } if (isAfterRebalanceOrValueUpdate(c, ctx)) { continue; } if (!ctx.isKeyValid()) { if (!allocateAndLinkEntry(c, ctx, key, false)) { continue; } } c.writeValue(ctx, value, false); if (!c.publish()) { c.releaseNewValue(ctx); rebalance(c); continue; } if (c.linkValue(ctx) != ValueUtils.ValueResult.TRUE) { c.releaseNewValue(ctx); c.unpublish(); } else { c.unpublish(); checkRebalance(c); return null; } } throw new RuntimeException("put failed: reached retry limit (1024)."); }
InternalOakMap { V put(K key, V value, OakTransformer<V> transformer) { if (key == null || value == null) { throw new NullPointerException(); } ThreadContext ctx = getThreadContext(); for (int i = 0; i < MAX_RETRIES; i++) { Chunk<K, V> c = findChunk(key); c.lookUp(ctx, key); if (ctx.isValueValid()) { Result res = valueOperator.exchange(c, ctx, value, transformer, valueSerializer, valuesMemoryManager, this); if (res.operationResult == ValueUtils.ValueResult.TRUE) { return (V) res.value; } continue; } if (isAfterRebalanceOrValueUpdate(c, ctx)) { continue; } if (!ctx.isKeyValid()) { if (!allocateAndLinkEntry(c, ctx, key, false)) { continue; } } c.writeValue(ctx, value, false); if (!c.publish()) { c.releaseNewValue(ctx); rebalance(c); continue; } if (c.linkValue(ctx) != ValueUtils.ValueResult.TRUE) { c.releaseNewValue(ctx); c.unpublish(); } else { c.unpublish(); checkRebalance(c); return null; } } throw new RuntimeException("put failed: reached retry limit (1024)."); } }
InternalOakMap { V put(K key, V value, OakTransformer<V> transformer) { if (key == null || value == null) { throw new NullPointerException(); } ThreadContext ctx = getThreadContext(); for (int i = 0; i < MAX_RETRIES; i++) { Chunk<K, V> c = findChunk(key); c.lookUp(ctx, key); if (ctx.isValueValid()) { Result res = valueOperator.exchange(c, ctx, value, transformer, valueSerializer, valuesMemoryManager, this); if (res.operationResult == ValueUtils.ValueResult.TRUE) { return (V) res.value; } continue; } if (isAfterRebalanceOrValueUpdate(c, ctx)) { continue; } if (!ctx.isKeyValid()) { if (!allocateAndLinkEntry(c, ctx, key, false)) { continue; } } c.writeValue(ctx, value, false); if (!c.publish()) { c.releaseNewValue(ctx); rebalance(c); continue; } if (c.linkValue(ctx) != ValueUtils.ValueResult.TRUE) { c.releaseNewValue(ctx); c.unpublish(); } else { c.unpublish(); checkRebalance(c); return null; } } throw new RuntimeException("put failed: reached retry limit (1024)."); } InternalOakMap(K minKey, OakSerializer<K> keySerializer, OakSerializer<V> valueSerializer,
        OakComparator<K> oakComparator, MemoryManager vMM, MemoryManager kMM, int chunkMaxItems,
        ValueUtils valueOperator); }
InternalOakMap { V put(K key, V value, OakTransformer<V> transformer) { if (key == null || value == null) { throw new NullPointerException(); } ThreadContext ctx = getThreadContext(); for (int i = 0; i < MAX_RETRIES; i++) { Chunk<K, V> c = findChunk(key); c.lookUp(ctx, key); if (ctx.isValueValid()) { Result res = valueOperator.exchange(c, ctx, value, transformer, valueSerializer, valuesMemoryManager, this); if (res.operationResult == ValueUtils.ValueResult.TRUE) { return (V) res.value; } continue; } if (isAfterRebalanceOrValueUpdate(c, ctx)) { continue; } if (!ctx.isKeyValid()) { if (!allocateAndLinkEntry(c, ctx, key, false)) { continue; } } c.writeValue(ctx, value, false); if (!c.publish()) { c.releaseNewValue(ctx); rebalance(c); continue; } if (c.linkValue(ctx) != ValueUtils.ValueResult.TRUE) { c.releaseNewValue(ctx); c.unpublish(); } else { c.unpublish(); checkRebalance(c); return null; } } throw new RuntimeException("put failed: reached retry limit (1024)."); } InternalOakMap(K minKey, OakSerializer<K> keySerializer, OakSerializer<V> valueSerializer,
        OakComparator<K> oakComparator, MemoryManager vMM, MemoryManager kMM, int chunkMaxItems,
        ValueUtils valueOperator);  }
InternalOakMap { V put(K key, V value, OakTransformer<V> transformer) { if (key == null || value == null) { throw new NullPointerException(); } ThreadContext ctx = getThreadContext(); for (int i = 0; i < MAX_RETRIES; i++) { Chunk<K, V> c = findChunk(key); c.lookUp(ctx, key); if (ctx.isValueValid()) { Result res = valueOperator.exchange(c, ctx, value, transformer, valueSerializer, valuesMemoryManager, this); if (res.operationResult == ValueUtils.ValueResult.TRUE) { return (V) res.value; } continue; } if (isAfterRebalanceOrValueUpdate(c, ctx)) { continue; } if (!ctx.isKeyValid()) { if (!allocateAndLinkEntry(c, ctx, key, false)) { continue; } } c.writeValue(ctx, value, false); if (!c.publish()) { c.releaseNewValue(ctx); rebalance(c); continue; } if (c.linkValue(ctx) != ValueUtils.ValueResult.TRUE) { c.releaseNewValue(ctx); c.unpublish(); } else { c.unpublish(); checkRebalance(c); return null; } } throw new RuntimeException("put failed: reached retry limit (1024)."); } InternalOakMap(K minKey, OakSerializer<K> keySerializer, OakSerializer<V> valueSerializer,
        OakComparator<K> oakComparator, MemoryManager vMM, MemoryManager kMM, int chunkMaxItems,
        ValueUtils valueOperator);   }