/*
 * Copyright 2012-2018 Chronicle Map Contributors
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package net.openhft.chronicle.map.impl.stage.entry;

import net.openhft.chronicle.bytes.Bytes;
import net.openhft.chronicle.hash.Data;
import net.openhft.chronicle.hash.impl.CompactOffHeapLinearHashTable;
import net.openhft.chronicle.hash.impl.stage.entry.AllocatedChunks;
import net.openhft.chronicle.hash.impl.stage.entry.HashEntryStages;
import net.openhft.chronicle.hash.impl.stage.query.KeySearch;
import net.openhft.chronicle.map.MapEntry;
import net.openhft.chronicle.map.VanillaChronicleMap;
import net.openhft.chronicle.map.impl.VanillaChronicleMapHolder;
import net.openhft.chronicle.map.impl.stage.data.bytes.EntryValueBytesData;
import net.openhft.sg.Stage;
import net.openhft.sg.StageRef;
import net.openhft.sg.Staged;
import org.jetbrains.annotations.NotNull;

import static net.openhft.chronicle.map.VanillaChronicleMap.alignAddr;

@Staged
public abstract class MapEntryStages<K, V> extends HashEntryStages<K>
        implements MapEntry<K, V> {

    @StageRef
    public VanillaChronicleMapHolder<?, ?, ?> mh;
    @StageRef
    public AllocatedChunks allocatedChunks;
    public long valueSizeOffset = -1;
    @Stage("ValueSize")
    public long valueSize = -1;
    @Stage("ValueSize")
    public long valueOffset;
    @StageRef
    public EntryValueBytesData<V> entryValue;
    @StageRef
    KeySearch<K> ks;

    long countValueSizeOffset() {
        return keyEnd();
    }

    @SuppressWarnings("unused")
    void initValueSizeOffset() {
        valueSizeOffset = countValueSizeOffset();
    }

    void initValueSize(long valueSize) {
        this.valueSize = valueSize;
        Bytes segmentBytes = s.segmentBytesForWrite();
        segmentBytes.writePosition(valueSizeOffset);
        mh.m().valueSizeMarshaller.writeSize(segmentBytes, valueSize);
        long currentPosition = segmentBytes.writePosition();
        long currentAddr = segmentBytes.addressForRead(currentPosition);
        long skip = alignAddr(currentAddr, mh.m().alignment) - currentAddr;
        if (skip > 0)
            segmentBytes.writeSkip(skip);
        valueOffset = segmentBytes.writePosition();
    }

    @SuppressWarnings("unused")
    void initValueSize() {
        Bytes segmentBytes = s.segmentBytesForRead();
        segmentBytes.readPosition(valueSizeOffset);
        valueSize = mh.m().readValueSize(segmentBytes);
        valueOffset = segmentBytes.readPosition();
    }

    public void initValue(Data<?> value) {
        initValueSize(value.size());
        writeValue(value);
    }

    public void writeValue(Data<?> value) {
        initDelayedUpdateChecksum(true);
        // In acquireContext(), replaceValue() is called for
        // 1) executing custom replaceValue() logic, if defined, from configured MapEntryOperations
        // 2) Update replication status (bits, timestamp)
        // 3) update entry checksum
        // but the actual entry replacement is not needed, if the value object is a flyweight over
        // the off-heap bytes. This condition avoids in-place data copy.
        // TODO would be nice to reduce scope of this check, i. e. check only when it could be
        // true, and avoid when it surely false (fresh value put, relocating put etc.)
        // TODO this optimization is now disabled, because it calls value.bytes() that forces double
        // data copy, if sizedReader/Writer configured for the value.
//        RandomDataInput valueBytes = value.bytes();
//        if (valueBytes instanceof NativeBytesStore &&
//                valueBytes.address(value.offset()) == s.segmentBS.address(valueOffset)) {
//            return;
//        }
        value.writeTo(s.segmentBS, valueOffset);
    }

    @Override
    public long entryEnd() {
        return valueOffset + valueSize;
    }

    @NotNull
    @Override
    public Data<V> value() {
        checkOnEachPublicOperation.checkOnEachPublicOperation();
        return entryValue;
    }

    public long newSizeOfEverythingBeforeValue(Data<V> newValue) {
        return valueSizeOffset + mh.m().valueSizeMarshaller.storingLength(newValue.size()) -
                keySizeOffset;
    }

    public void innerDefaultReplaceValue(Data<V> newValue) {
        assert s.innerUpdateLock.isHeldByCurrentThread();

        boolean newValueSizeIsDifferent = newValue.size() != this.valueSize;
        if (newValueSizeIsDifferent) {
            long newSizeOfEverythingBeforeValue = newSizeOfEverythingBeforeValue(newValue);
            long entryStartOffset = keySizeOffset;
            VanillaChronicleMap<?, ?, ?> m = mh.m();
            long newValueOffset =
                    alignAddr(entryStartOffset + newSizeOfEverythingBeforeValue, mh.m().alignment);
            long newEntrySize = newEntrySize(newValue, entryStartOffset, newValueOffset);
            int newSizeInChunks = m.inChunks(newEntrySize);
            newValueDoesNotFit:
            if (newSizeInChunks > entrySizeInChunks) {
                if (newSizeInChunks > m.maxChunksPerEntry) {
                    throw new IllegalArgumentException(m.toIdentityString() +
                            ": Value too large: entry takes " + newSizeInChunks + " chunks, " +
                            m.maxChunksPerEntry + " is maximum.");
                }
                if (s.realloc(pos, entrySizeInChunks, newSizeInChunks)) {
                    break newValueDoesNotFit;
                }
                relocation(newValue, newEntrySize);
                return;
            } else if (newSizeInChunks < entrySizeInChunks) {
                s.freeExtra(pos, entrySizeInChunks, newSizeInChunks);
            }
            // new size != old size => size is not constant => size is actually written =>
            // to prevent (at least) this execution:
            // 1. concurrent reader thread reads the size
            // 2. this thread updates the size and the value
            // 3. concurrent reader reads the value
            // We MUST upgrade to exclusive lock
        } else {
            // TODO to turn the following block on, JLANG-46 is required. Also unclear what happens
            // if the value is DataValue generated with 2, 4 or 8 distinct bytes, putting on-heap
            // implementation of such value is also not atomic currently, however there is a way
            // to make it atomic, we should identify such cases and make a single write:
            // state = UNSAFE.getLong(onHeapValueObject, offsetToTheFirstField);
            // bytes.writeLong(state);
//            boolean newValueSizeIsPowerOf2 = ((newValueSize - 1L) & newValueSize) != 0;
//            if (!newValueSizeIsPowerOf2 || newValueSize > 8L) {
//                 if the new value size is 1, 2, 4, or 8, it is written not atomically only if
//                 the user provided own marshaller and writes value byte-by-byte, that is very
//                 unlikely. in this case the user should update acquire write lock before write
//                 updates himself
//                upgradeToWriteLock();
//            }
        }
        s.innerWriteLock.lock();

        if (newValueSizeIsDifferent) {
            initValue(newValue);
        } else {
            writeValue(newValue);
        }
    }

    public long newEntrySize(Data<V> newValue, long entryStartOffset, long newValueOffset) {
        return checksumStrategy.extraEntryBytes() +
                newValueOffset + newValue.size() - entryStartOffset;
    }

    protected void relocation(Data<V> newValue, long newEntrySize) {
        // need to copy, because in initEntryAndKeyCopying(), in alloc(), nextTier() called ->
        // hashLookupPos cleared, as a dependant
        long oldHashLookupPos = hlp.hashLookupPos;
        long oldHashLookupAddr = s.tierBaseAddr;

        boolean tierHasChanged = allocatedChunks.initEntryAndKeyCopying(
                newEntrySize, valueSizeOffset - keySizeOffset, pos, entrySizeInChunks);

        if (tierHasChanged) {
            // implicitly inits key search, locating hashLookupPos on the empty slot
            if (!ks.searchStateAbsent())
                throw new AssertionError();
        }

        initValue(newValue);

        freeExtraAllocatedChunks();

        CompactOffHeapLinearHashTable hl = hh.h().hashLookup;
        long hashLookupKey = hl.key(hl.readEntry(oldHashLookupAddr, oldHashLookupPos));
        hl.checkValueForPut(pos);
        hl.writeEntryVolatile(s.tierBaseAddr, hlp.hashLookupPos, hashLookupKey, pos);
        // write lock is needed anyway (see testPutShouldBeWriteLocked()) but the scope is reduced
        // as much as possible
        s.innerWriteLock.lock();
        if (tierHasChanged)
            hl.remove(oldHashLookupAddr, oldHashLookupPos);
    }

    public final long entrySize(long keySize, long valueSize) {
        long sizeOfEverythingBeforeValue = sizeOfEverythingBeforeValue(keySize, valueSize);
        return innerEntrySize(sizeOfEverythingBeforeValue, valueSize);
    }

    public long innerEntrySize(long sizeOfEverythingBeforeValue, long valueSize) {
        if (!mh.m().constantlySizedEntry && mh.m().couldNotDetermineAlignmentBeforeAllocation)
            sizeOfEverythingBeforeValue += mh.m().worstAlignment;
        int alignment = mh.m().alignment;
        return alignAddr(sizeOfEverythingBeforeValue, alignment) +
                alignAddr(valueSize, alignment);
    }

    long sizeOfEverythingBeforeValue(long keySize, long valueSize) {
        return mh.m().keySizeMarshaller.storingLength(keySize) + keySize +
                checksumStrategy.extraEntryBytes() +
                mh.m().valueSizeMarshaller.storingLength(valueSize);
    }

    public final void freeExtraAllocatedChunks() {
        // fast path
        if (!mh.m().constantlySizedEntry && mh.m().couldNotDetermineAlignmentBeforeAllocation &&
                entrySizeInChunks < allocatedChunks.allocatedChunks) {
            s.freeExtra(pos, allocatedChunks.allocatedChunks, entrySizeInChunks);
        } else {
            initEntrySizeInChunks(allocatedChunks.allocatedChunks);
        }
    }

    public boolean entryDeleted() {
        return false;
    }
}
