src/main/java/net/onrc/onos/core/util/distributed/sharedlog/hazelcast/HazelcastRuntime.java

package net.onrc.onos.core.util.distributed.sharedlog.hazelcast;

import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Preconditions.checkArgument;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import net.onrc.onos.core.util.distributed.sharedlog.ByteValue;
import net.onrc.onos.core.util.distributed.sharedlog.LogEventListener;
import net.onrc.onos.core.util.distributed.sharedlog.SeqNum;
import net.onrc.onos.core.util.distributed.sharedlog.Sequencer;
import net.onrc.onos.core.util.distributed.sharedlog.SharedLogObject;
import net.onrc.onos.core.util.distributed.sharedlog.SharedLogObjectID;
import net.onrc.onos.core.util.distributed.sharedlog.exception.LogNotApplicable;
import net.onrc.onos.core.util.distributed.sharedlog.exception.LogWriteTimedOut;
import net.onrc.onos.core.util.distributed.sharedlog.internal.LogValue;
import net.onrc.onos.core.util.distributed.sharedlog.internal.SnapShotValue;
import net.onrc.onos.core.util.distributed.sharedlog.internal.StringID;
import net.onrc.onos.core.util.distributed.sharedlog.internal.NoOp;
import net.onrc.onos.core.util.distributed.sharedlog.runtime.LogBasedRuntime;
import net.onrc.onos.core.util.distributed.sharedlog.runtime.SequencerRuntime;

import org.apache.commons.lang3.concurrent.ConcurrentException;
import org.apache.commons.lang3.concurrent.ConcurrentInitializer;
import org.apache.commons.lang3.concurrent.ConcurrentUtils;
import org.apache.commons.lang3.tuple.ImmutablePair;
import org.joda.time.DateTime;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.annotations.Beta;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.IAtomicLong;
import com.hazelcast.core.IFunction;
import com.hazelcast.core.IMap;
import com.hazelcast.nio.ObjectDataInput;
import com.hazelcast.nio.ObjectDataOutput;
import com.hazelcast.nio.serialization.DataSerializable;


/**
 * LogBasedRuntime using Hazelcast.
 */
@Beta
public class HazelcastRuntime implements LogBasedRuntime {

    private static final Logger log = LoggerFactory
            .getLogger(HazelcastRuntime.class);

    // Hazelcast distributed object name prefixes
    private static final String LATEST_SNAPSHOT_PREFIX = "latestSS://";
    private static final String SNAPSHOTMAP_PREFIX = "ssmap://";
    private static final String LOGMAP_PREFIX = "logmap://";

    // FIXME set appropriate timeout & make it configurable
    private static final long LOG_READ_TIMEOUT_MS = 1000L;

    // FIXME make these configurable from file, etc.
    /**
     * Number of times to retry, if write to LogMap failed due to
     * other node invalidated the sequencer due to timeout.
     */
    private static final int UPDATE_RETRIES = 5;

    /**
     * Interval to check if we should create SnapShot.
     * <p>
     * If the sequence number of the log just written is multiple of this value,
     * SnapShot check and log cleanup will be triggered.
     */
    private static final int SNAPSHOT_CHECK_INTERVAL = 50;

    /**
     * Interval SnapShot should be created.
     * <p>
     * SnapShot will be created if latest SnapShot was more than
     * this interval apart from last written.
     */
    public static final int SNAPSHOT_INTERVAL = 500;

    /**
     * Maximum number of Snapshots to store in distributed Map.
     */
    public static final int MAX_SNAPSHOTS = 10;

    // TODO create ThreadFactory to assign thread name and lower priority
    // executor for SnapShot builder
    private static final ExecutorService EXECUTOR =
            new ThreadPoolExecutor(1, 5, 3L, TimeUnit.MINUTES,
                    new ArrayBlockingQueue<Runnable>(10),
                    new ThreadPoolExecutor.DiscardOldestPolicy());

    private final HazelcastInstance instance;
    private final SequencerRuntime sequencerRuntime;

    private final ConcurrentMap<SharedLogObjectID, LogMapManager>
                    listenerMap;

    /**
     * Initializes Hazelcast based LogRuntime.
     *
     * @param instance Hazelcast instance to use
     * @param sequencerRuntime {@link SequencerRuntime} to use
     */
    public HazelcastRuntime(HazelcastInstance instance,
                            SequencerRuntime sequencerRuntime) {

        this.instance = checkNotNull(instance);
        this.sequencerRuntime = checkNotNull(sequencerRuntime);
        this.listenerMap = new ConcurrentHashMap<>();
    }

    @Override
    public SharedLogObjectID getOid(String sharedObjectName) {

        // TODO Create specialized SharedLogObject, if getting IMap from
        //      HZ instance each time becomes a significant overhead.

        return new StringID(sharedObjectName);
    }

    @Override
    public SeqNum updateHelper(SharedLogObject sobj, ByteValue logValue,
                                boolean queryBeforeUpdate)
                                   throws LogNotApplicable, LogWriteTimedOut {

        return updateHelper(sobj, logValue, queryBeforeUpdate, UPDATE_RETRIES);
    }

    /**
     * Apply given logValue to SharedLogObject.
     *
     * @param sobj SharedLogObject to manipulate.
     * @param logValue LogValue to apply
     * @param queryBeforeUpdate true if log should be replayed to latest before updating
     * @param retries number of retries
     * @return sequence number of the new log entry
     *
     * @throws LogNotApplicable Thrown when {@code logValue} was not applicable to
     *          SharedLogObject.
     * @throws LogWriteTimedOut Thrown when write failed due to time out.
     */
    private SeqNum updateHelper(final SharedLogObject sobj, ByteValue logValue,
                                boolean queryBeforeUpdate, int retries)
                                   throws LogNotApplicable, LogWriteTimedOut {

        checkNotNull(sobj);
        checkNotNull(logValue);

        log.trace("updating {}@{} with {}", sobj.getObjectID(), sobj.getSeqNum(),
                    logValue);

        final Sequencer sequencer = sequencerRuntime.getSequencer(sobj.getObjectID());
        final IMap<SeqNum, LogValue> logMap = getLogMap(sobj);

        // allocate seq #
        final SeqNum allocated = sequencer.next();
        log.trace("allocated {}", allocated);

        // replay
        if (queryBeforeUpdate) {
            // FIXME How to handle initial case?
            queryHelper(sobj, allocated.prev());
        }

        // test if applicable
        sobj.acquireWriteLock();
        try {
            final boolean isApplicable = sobj.isApplicable(allocated, logValue);
            if (!isApplicable) {
                log.trace("log not applicable abondoning {}", allocated);
                logMap.putIfAbsent(allocated, NoOp.VALUE);
                throw new LogNotApplicable("Rejected by " + sobj.getObjectID());
            }

            // write to shared log
            LogValue existing = logMap.putIfAbsent(allocated, logValue);
            if (NoOp.VALUE.equals(existing)) {
                if (retries > 0) {
                    log.trace("write failed due to time out retrying {}", retries);
                    return updateHelper(sobj, logValue, queryBeforeUpdate, retries - 1);
                }
                throw new LogWriteTimedOut("Was timed out by other node by " + sobj.getObjectID());
            }

            // apply to local object
            sobj.apply(allocated, logValue);

            // Success.

            // FIXME Current design is that any SharedObject instance could
            // become SnapShot writer/Log cleaner.
            // We may need provide a way for a SharedObject instance
            // to declare that it does not want to become
            // a SnapShot writer (Don't wont unexpected ReadLock, etc.)

            // give hint to snapshot builder
            if (allocated.longValue() % SNAPSHOT_CHECK_INTERVAL == 0) {
                // check and create snapshot in background
                EXECUTOR.execute(new Runnable() {

                    @Override
                    public void run() {
                        createSnapShot(sobj);
                    }
                });
            }
            return allocated;

        } finally {
            sobj.releaseWriteLock();
        }
    }

    /**
     * Updates latest snapshot pointer.
     * <p>
     * IFunction to apply to IAtomicLong used as latest snapshot pointer.
     */
    public static final class UpdateLatestSnapshot implements
                                IFunction<Long, Long>, DataSerializable {

        private SeqNum ssCreated;

        /**
         * Updates latest snapshot pointer to {@code ssCreated}.
         *
         * @param ssCreated value to update to
         */
        public UpdateLatestSnapshot(SeqNum ssCreated) {
            this.ssCreated = ssCreated;
        }

        @Override
        public Long apply(Long input) {
            final SeqNum in = SeqNum.anyValueOf(input);

            if (ssCreated.compareTo(in) > 0) {
                // update if snapshot written is >= existing latest SS.
                return ssCreated.longValue();
            } else {
                return input;
            }
        }

        @Override
        public void writeData(ObjectDataOutput out) throws IOException {
            out.writeLong(ssCreated.longValue());
        }

        @Override
        public void readData(ObjectDataInput in) throws IOException {
            this.ssCreated = SeqNum.valueOf(in.readLong());
        }

        /**
         * Default constructor for deserialization.
         */
        public UpdateLatestSnapshot() {
            this.ssCreated = SeqNum.INITIAL;
        }
    }

    /**
     * Creates a snapshot of {@link SharedLogObject} if
     * latest snapshot is old.
     *
     * @param sobj {@link SharedLogObject} to create snapshot.
     */
    private void createSnapShot(SharedLogObject sobj) {
        if (sobj == null) {
            log.warn("Tried to create snapshot of null object. Ignoring.");
            return;
        }
        log.trace("Checking if {} needs a new snapshot.", sobj.getObjectID());

        // XXX lock can be more fine grained if necessary
        final IAtomicLong rawLatest = getLatestSnapshotStore(sobj.getObjectID());
        sobj.acquireReadLock();
        try {
            final SeqNum current = sobj.getSeqNum();
            final SeqNum latest = SeqNum.anyValueOf(rawLatest.get());

            // if distance(current - latest) > SNAPSHOT_INTERVAL
            if (latest.distance(current) < SNAPSHOT_INTERVAL) {
                // not the time to create snap shot yet
                log.trace("Skip creating snapshot. current:{}, latest:{}", current, latest);
                return;
            }
            // create SnapShot Bytes
            final ImmutablePair<SeqNum, ? extends SnapShotValue> snapshot
                = sobj.createSnapshot();

            // write to SnapShot Map
            IMap<SeqNum, SnapShotValue> ssMap = getSnapshotMap(sobj);
            final SeqNum ssCreated = snapshot.getKey();
            // XXX is there a way to do setIfAbsent
            ssMap.set(ssCreated, snapshot.getValue());
            log.info("Created snapshot. {}@{}", sobj.getObjectID(), ssCreated);

            // update latest snapshot pointer
            rawLatest.alter(new UpdateLatestSnapshot(ssCreated));

        } finally {
            sobj.releaseReadLock();
        }

        // XXX Should we be triggering log clean up in same context?
        IMap<SeqNum, SnapShotValue> ssMap = getSnapshotMap(sobj);

        // note: snapshots is a copy of keySet unlike usual Map
        Set<SeqNum> snapshots = ssMap.keySet();
        if (snapshots.size() > MAX_SNAPSHOTS) {
            // FIXME we should be able to avoid prev() walking if we can
            // always safely compare orders of SeqNums.

            SeqNum latest = SeqNum.valueOf(rawLatest.get());
            snapshots.remove(latest);
            SeqNum remove = latest.prev();
            while (snapshots.size() > MAX_SNAPSHOTS) {
                snapshots.remove(remove);
                remove = remove.prev();
            }

            // what's still left in `snapshots` is subject to removal.
            // delete snapshots and find log SeqNum to start removing
            SeqNum deleteBeforeThis = null;
            for (SeqNum ss : snapshots) {
                log.debug("Removed snapshot {}@{}", sobj.getObjectID(), ss);
                // XXX If we decide to persist snapshots for reference,
                //     evict() ing here is another option.
                ssMap.delete(ss);
                if (deleteBeforeThis == null) {
                    deleteBeforeThis = ss;
                } else {
                    if (deleteBeforeThis.compareTo(ss) > 0) {
                        deleteBeforeThis = ss;
                    }
                }
            }
            if (deleteBeforeThis == null) {
                // nothing to do
                return;
            }

            // start removing log
            IMap<SeqNum, LogValue> logMap = getLogMap(sobj);
            // naive log cleaning
            log.debug("Trimming log before {}@{}...", sobj.getObjectID(), deleteBeforeThis);
            while (logMap.remove(deleteBeforeThis) != null) {
                deleteBeforeThis = deleteBeforeThis.prev();
            }
            log.debug("Trimming log up to this{}@{}...", sobj.getObjectID(), deleteBeforeThis);
        }
    }

    /**
     * Gets the Log Map for specified shared object.
     *
     * @param sobj {@link SharedLogObject}
     * @return Log Map
     */
    private IMap<SeqNum, LogValue> getLogMap(SharedLogObject sobj) {
        return getLogMap(sobj.getObjectID());
    }

    /**
     * Gets the Log Map for specified shared object ID.
     *
     * @param oid {@link SharedLogObjectID}
     * @return Log Map
     */
    private IMap<SeqNum, LogValue> getLogMap(SharedLogObjectID oid) {
        return instance.getMap(LOGMAP_PREFIX + oid.getObjectName());
    }

    /**
     * Gets the Snapshot Map for specified shared object.
     *
     * @param sobj {@link SharedLogObject}
     * @return Snapshot Map
     */
    private IMap<SeqNum, SnapShotValue> getSnapshotMap(SharedLogObject sobj) {
        return instance.getMap(SNAPSHOTMAP_PREFIX + sobj.getObjectID().getObjectName());
    }

    /**
     * Gets the store holding latest snapshot version.
     *
     * @param oid {@link SharedLogObjectID}
     * @return IAtomicLong
     */
    private IAtomicLong getLatestSnapshotStore(SharedLogObjectID oid) {
        return instance.getAtomicLong(LATEST_SNAPSHOT_PREFIX + oid.getObjectName());
    }

    @Override
    public void queryHelper(SharedLogObject sobj) {
        queryHelper(checkNotNull(sobj), sequencerRuntime.getSequencer(sobj.getObjectID()).get());
    }

    @Override
    public void queryHelper(SharedLogObject sobj, SeqNum replayTo) {

        checkNotNull(sobj);
        checkNotNull(replayTo);
        // TODO check if sobj.getSeqNum() <= replayTo

        if (sobj.getSeqNum().equals(replayTo)) {
            // nothing to do.
            return;
        }

        log.trace("querying {}@{} to {}", sobj.getObjectID(), sobj.getSeqNum(),
                                            replayTo);

        IMap<SeqNum, LogValue> logMap = getLogMap(sobj);
        final LogMapManager logCache = getLogMapManager(sobj.getObjectID());

        // TODO Consider more fine grained lock if we ever need
        //   SharedLogObject to be exposed to heavy concurrent read/write.

        sobj.acquireWriteLock();
        try {
            final SeqNum current = sobj.getSeqNum();

            List<SeqNum> range = getSeqNumRange(current, replayTo);
            Map<SeqNum, Future<LogValue>> values = new HashMap<>(range.size());

            // FIXME use notification based cache

            // pre-request all range first to pre-populate near cache
            log.trace("Pre reading range: {}", range);
            for (SeqNum key : range) {
                Future<LogValue> value = logCache.getLogValue(key);
                values.put(key, value);
            }

            // walk and apply range 1 by 1
            for (SeqNum key : range) {
                Future<LogValue> future = values.get(key);
                LogValue value = null;

                // FIXME handle Recycled Snapshot log entry scenario
                // when the get result is null it could be waiting for writer
                // or log has been trimmed.

                final long deadline = System.currentTimeMillis()
                                        + LOG_READ_TIMEOUT_MS;
                if (log.isTraceEnabled()) {
                    log.trace("Try reading {} until {}", key, new DateTime(deadline));
                }
                while (System.currentTimeMillis() < deadline) {
                    try {
                        value = future.get(LOG_READ_TIMEOUT_MS,
                                                    TimeUnit.MILLISECONDS);
                        if (value == null) {
                            log.trace("{} was not there yet retrying.", key);
                            // value not set yet, retry;
                            future = logCache.getLogValue(key);
                            continue;
                        }
                    } catch (InterruptedException | ExecutionException e) {
                        log.error("queryError retrying", e);
                        future = logCache.getLogValue(key);
                        continue;
                    } catch (TimeoutException e) {
                        log.warn("query timed out", e);
                    }
                }
                if (value == null) {
                    log.trace("{} was not found, writer failed?", key);
                    // writer failure scenario handling
                    //  mark sequence number null
                    LogValue newVal = logMap.put(key, NoOp.VALUE);
                    if (newVal == null) {
                        // skip this sequence number.
                        continue;
                    }
                    log.trace("{} appeared while invalidating.", newVal);
                    value = newVal;
                }

                log.trace("applying log {} {}", key, value);
                // apply log value
                sobj.apply(key, value);
            }

        } finally {
            sobj.releaseWriteLock();
            log.trace("query done");
        }
    }

    /**
     * Creates a List containing (afterThis, uptoThis].
     *
     * @param afterThis beginning of range not including this sequence number
     * @param upToThis end of range including this sequence number
     * @return List of SeqNum in order
     */
    private List<SeqNum> getSeqNumRange(SeqNum afterThis, SeqNum upToThis) {
        checkArgument(!upToThis.equals(SeqNum.INITIAL),
                        "upToThis must not be INITIAL");

        // TODO Wasting heap, etc.
        // If iterable SeqNum is the only thing required,
        // implement SeqNum range class, etc.
        List<SeqNum> range = new ArrayList<>();
        for (SeqNum i = afterThis.next(); !i.equals(upToThis); i = i.next()) {
            range.add(i);
        }
        range.add(upToThis);
        return range;
    }

    @Override
    public void addListener(SharedLogObjectID oid, LogEventListener listener) {
        LogMapManager hzListener = getLogMapManager(oid);
        hzListener.addListener(listener);

        getLogMap(oid).addEntryListener(hzListener, true);
    }

    @Override
    public void removeListener(SharedLogObjectID oid, LogEventListener listener) {
        LogMapManager hzListener = getLogMapManager(oid);
        hzListener.removeListener(listener);
    }

    /**
     * Gets the LogMapManager for given {@link SharedLogObjectID}.
     * <p/>
     * If listener was not registered, it will create and register a listener.
     *
     * @param oid {@link SharedLogObjectID}
     * @return {@link LogMapManager}
     */
    private LogMapManager getLogMapManager(final SharedLogObjectID oid) {
        LogMapManager listener
            = ConcurrentUtils.createIfAbsentUnchecked(listenerMap, oid,
                new ConcurrentInitializer<LogMapManager>() {
                    @Override
                    public LogMapManager get() throws ConcurrentException {
                        IMap<SeqNum, LogValue> logMap = getLogMap(oid);
                        return new LogMapManager(oid, logMap);
                    }
                });
        return listener;
    }

    @Override
    public SeqNum getLatestSnapShotId(SharedLogObjectID oid) {
        IAtomicLong latest = getLatestSnapshotStore(oid);
        final long ssId = latest.get();
        return SeqNum.anyValueOf(ssId);
    }

    @Override
    public boolean resetToSnapShot(SharedLogObject sobj, SeqNum ssId) {
        checkNotNull(ssId);

        IMap<SeqNum, SnapShotValue> ssMap = getSnapshotMap(sobj);
        SnapShotValue snapshot = ssMap.get(ssId);
        if (snapshot == null) {
            if (ssId.equals(SeqNum.INITIAL)) {
                snapshot = NoOp.VALUE;
            } else {
                log.error("Invalid Snapshot version {}@{} specified",
                                    sobj.getObjectID(), ssId);
                return false;
            }
        }

        sobj.reset(ssId, snapshot);

        return true;
    }

    /**
     * Gets the LogValues in range ({@code after}, {@code upToThis}].
     *
     * @param oid ID of SharedLogObject
     * @param after sequence number before (typically current sequence number)
     * @param upToThis last sequence number you want to retrieve
     * @return List of {@link LogValue}s in specified range
     */
    @Override
    public List<LogValue> getLogRange(final SharedLogObjectID oid,
                                    final SeqNum after, final SeqNum upToThis) {

        List<LogValue> logs = new ArrayList<>();
        final SeqNum oneAfterLast = upToThis.next();
        LogMapManager logCache = getLogMapManager(oid);
        // may want to do async pre-fetching if RPC is triggered often
        for (SeqNum s = after.next(); !s.equals(oneAfterLast); s = s.next()) {
            try {
                logs.add(logCache.getLogValue(s).get());
                // FIXME Need to somehow detect log has been recycled case and fail
            } catch (InterruptedException | ExecutionException e) {
                log.error("Failed getting log " + oid + "@" + s, e);
                // Should we retry/[ignore]/throw?
            }
        }
        return logs;
    }
}