class 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); // find chunk matching key
            c.lookUp(ctx, key);
            // If there is a matching value reference for the given key, and it is not marked as deleted,
            // then this put changes the slice pointed by this value reference.
            if (ctx.isValueValid()) {
                // there is a value and it is not deleted
                Result res = valueOperator.exchange(c, ctx, value, transformer, valueSerializer,
                    valuesMemoryManager,
                        this);
                if (res.operationResult == ValueUtils.ValueResult.TRUE) {
                    return (V) res.value;
                }
                // Exchange failed because the value was deleted/moved between lookup and exchange. Continue with
                // insertion.
                continue;
            }

            if (isAfterRebalanceOrValueUpdate(c, ctx)) {
                continue;
            }

            // AT THIS POINT EITHER (in all cases context is updated):
            // (1) Key wasn't found (key and value not valid)
            // (2) Key was found and it's value is deleted/invalid (key valid value invalid)
            if (!ctx.isKeyValid()) {
                if (!allocateAndLinkEntry(c, ctx, key, false)) {
                    continue; // allocation wasn't successfull and resulted in rebalance - retry
                }
            }

            c.writeValue(ctx, value, false); // write value in place

            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; // null can be returned only in zero-copy case
            }
        }
        throw new RuntimeException("put failed: reached retry limit (1024).");
    }

}

class InternalOakMapTest {

    private InternalOakMap<Integer, Integer> testMap;
    private static final long OPERATION_DELAY;
    private static final long LONG_TRANSFORMATION_DELAY;

    @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]);
    }

}