/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You 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 org.apache.jackrabbit.oak.plugins.blob;

import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.collect.Lists.newArrayList;
import static java.io.File.createTempFile;
import static org.apache.commons.io.FileUtils.copyFile;
import static org.apache.jackrabbit.oak.commons.FileIOUtils.copy;
import static org.apache.jackrabbit.oak.commons.FileIOUtils.merge;
import static org.apache.jackrabbit.oak.commons.FileIOUtils.sort;
import static org.apache.jackrabbit.oak.commons.IOUtils.closeQuietly;

import java.io.BufferedWriter;
import java.io.ByteArrayInputStream;
import java.io.Closeable;
import java.io.File;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.LineNumberReader;
import java.sql.Timestamp;
import java.util.ArrayDeque;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import com.google.common.base.Charsets;
import com.google.common.base.Function;
import com.google.common.base.Joiner;
import com.google.common.base.StandardSystemProperty;
import com.google.common.base.Stopwatch;
import com.google.common.collect.FluentIterable;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableListMultimap;
import com.google.common.collect.Iterators;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.io.Closeables;
import com.google.common.io.Files;
import com.google.common.util.concurrent.ListenableFutureTask;
import org.apache.commons.io.FileUtils;
import org.apache.commons.io.LineIterator;
import org.apache.jackrabbit.core.data.DataRecord;
import org.apache.jackrabbit.core.data.DataStoreException;
import org.apache.jackrabbit.oak.api.jmx.CheckpointMBean;
import org.apache.jackrabbit.oak.commons.FileIOUtils;
import org.apache.jackrabbit.oak.commons.FileIOUtils.FileLineDifferenceIterator;
import org.apache.jackrabbit.oak.plugins.blob.datastore.BlobIdTracker;
import org.apache.jackrabbit.oak.plugins.blob.datastore.BlobTracker;
import org.apache.jackrabbit.oak.plugins.blob.datastore.DataStoreBlobStore;
import org.apache.jackrabbit.oak.plugins.blob.datastore.SharedDataStoreUtils;
import org.apache.jackrabbit.oak.plugins.blob.datastore.SharedDataStoreUtils.SharedStoreRecordType;
import org.apache.jackrabbit.oak.spi.blob.GarbageCollectableBlobStore;
import org.apache.jackrabbit.oak.spi.whiteboard.Whiteboard;
import org.apache.jackrabbit.oak.spi.whiteboard.WhiteboardUtils;
import org.apache.jackrabbit.oak.stats.CounterStats;
import org.apache.jackrabbit.oak.stats.StatisticsProvider;
import org.apache.jackrabbit.oak.stats.StatsOptions;
import org.apache.jackrabbit.oak.stats.TimerStats;
import org.jetbrains.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Mark and sweep garbage collector.
 * 
 * Uses the file system to store internal state while in process to account for huge data.
 * This class is not thread safe.
 * 
 */
public class MarkSweepGarbageCollector implements BlobGarbageCollector {

    public static final Logger LOG = LoggerFactory.getLogger(MarkSweepGarbageCollector.class);

    public static final String TEMP_DIR = StandardSystemProperty.JAVA_IO_TMPDIR.value();

    public static final int DEFAULT_BATCH_COUNT = 1024;
    
    public static final String DELIM = ",";

    private static final Function<String, String> transformer = new Function<String, String>() {
        @Nullable
        @Override
        public String apply(@Nullable String input) {
            if (input != null) {
                return input.split(DELIM)[0];
            }
            return "";
        }};

    /** The last modified time before current time of blobs to consider for garbage collection. */
    private final long maxLastModifiedInterval;

    /** The blob store to be garbage collected. */
    private final GarbageCollectableBlobStore blobStore;

    /** Helper class to mark blob references which **/
    private final BlobReferenceRetriever marker;

    private final Executor executor;

    /** The batch count. */
    private final int batchCount;

    private final String repoId;

    private final String root;

    private final Whiteboard whiteboard;

    private CheckpointMBean checkpointMbean;

    /** Operation stats object **/
    private final GarbageCollectionOperationStats stats;

    private final OperationStatsCollector statsCollector;

    /** Operation consistency stats object **/
    private final GarbageCollectionOperationStats consistencyStats;

    private final OperationStatsCollector consistencyStatsCollector;

    private boolean traceOutput;

    /**
     * Creates an instance of MarkSweepGarbageCollector
     *
     * @param marker BlobReferenceRetriever instanced used to fetch refereed blob entries
     * @param blobStore the blob store instance
     * @param executor executor
     * @param root the root absolute path of directory under which temporary
     *             files would be created
     * @param batchCount batch sized used for saving intermediate state
     * @param maxLastModifiedInterval lastModifiedTime in millis. Only files with time
     *                                less than this time would be considered for GC
     * @param repositoryId unique repository id for this node
     * @param whiteboard whiteboard instance
     * @param statisticsProvider statistics provider instance
     * @throws IOException
     */
    public MarkSweepGarbageCollector(
            BlobReferenceRetriever marker,
            GarbageCollectableBlobStore blobStore,
            Executor executor,
            String root,
            int batchCount,
            long maxLastModifiedInterval,
            @Nullable String repositoryId,
            @Nullable Whiteboard whiteboard,
            @Nullable StatisticsProvider statisticsProvider)
            throws IOException {
        this.executor = executor;
        this.blobStore = blobStore;
        checkNotNull(blobStore, "BlobStore cannot be null");
        this.marker = marker;
        this.batchCount = batchCount;
        this.maxLastModifiedInterval = maxLastModifiedInterval;
        this.repoId = repositoryId;
        this.root = root;
        this.whiteboard = whiteboard;
        if (whiteboard != null) {
            this.checkpointMbean = WhiteboardUtils.getService(whiteboard, CheckpointMBean.class);
        }

        // re-initialize the statsProvider if passed as parameter
        if (statisticsProvider == null) {
            statisticsProvider = StatisticsProvider.NOOP;
        }
        this.stats = new GarbageCollectionOperationStats(statisticsProvider);
        this.statsCollector = stats.getCollector();
        this.consistencyStats =
            new GarbageCollectionOperationStats(statisticsProvider, GarbageCollectionOperationStats.CONSISTENCY_NAME);
        this.consistencyStatsCollector = consistencyStats.getCollector();
    }

    public MarkSweepGarbageCollector(
            BlobReferenceRetriever marker,
            GarbageCollectableBlobStore blobStore,
            Executor executor,
            String root,
            int batchCount,
            long maxLastModifiedInterval,
            @Nullable String repositoryId)
            throws IOException {
        this(marker, blobStore, executor, root, batchCount, maxLastModifiedInterval, repositoryId, null, null);
    }

    /**
     * Instantiates a new blob garbage collector.
     */
    public MarkSweepGarbageCollector(
            BlobReferenceRetriever marker,
            GarbageCollectableBlobStore blobStore,
            Executor executor,
            long maxLastModifiedInterval,
            @Nullable String repositoryId,
            @Nullable Whiteboard whiteboard,
            @Nullable StatisticsProvider statisticsProvider)
            throws IOException {
        this(marker, blobStore, executor, TEMP_DIR, DEFAULT_BATCH_COUNT, maxLastModifiedInterval, repositoryId, whiteboard, statisticsProvider);
    }

    @Override
    public void collectGarbage(boolean markOnly) throws Exception {
        markAndSweep(markOnly, false);
    }

    @Override
    public void collectGarbage(boolean markOnly, boolean forceBlobRetrieve) throws Exception {
        markAndSweep(markOnly, forceBlobRetrieve);
    }

    /**
     * Returns the stats related to GC for all repos
     * 
     * @return a list of GarbageCollectionRepoStats objects
     * @throws Exception
     */
    @Override
    public List<GarbageCollectionRepoStats> getStats() throws Exception {
        List<GarbageCollectionRepoStats> stats = newArrayList();
        if (SharedDataStoreUtils.isShared(blobStore)) {
            // Get all the references available
            List<DataRecord> refFiles =
                ((SharedDataStore) blobStore).getAllMetadataRecords(SharedStoreRecordType.REFERENCES.getType());
            ImmutableListMultimap<String, DataRecord> references =
                FluentIterable.from(refFiles).index(new Function<DataRecord, String>() {
                    @Override public String apply(DataRecord input) {
                        return SharedStoreRecordType.REFERENCES.getIdFromName(input.getIdentifier().toString());

                    }
                });

            // Get all the markers available
            List<DataRecord> markerFiles =
                ((SharedDataStore) blobStore).getAllMetadataRecords(SharedStoreRecordType.MARKED_START_MARKER.getType());
            Map<String, DataRecord> markers = Maps.uniqueIndex(markerFiles, new Function<DataRecord, String>() {
                @Override
                public String apply(DataRecord input) {
                    return input.getIdentifier().toString().substring(SharedStoreRecordType.MARKED_START_MARKER.getType().length() + 1);
                }
            });

            // Get all the repositories registered
            List<DataRecord> repoFiles =
                ((SharedDataStore) blobStore).getAllMetadataRecords(SharedStoreRecordType.REPOSITORY.getType());

            for (DataRecord repoRec : repoFiles) {
                String id = SharedStoreRecordType.REPOSITORY.getIdFromName(repoRec.getIdentifier().toString());

                GarbageCollectionRepoStats stat = new GarbageCollectionRepoStats();
                stats.add(stat);

                stat.setRepositoryId(id);
                if (id != null && id.equals(repoId)) {
                    stat.setLocal(true);
                }

                if (references.containsKey(id)) {
                    ImmutableList<DataRecord> refRecs = references.get(id);
                    for(DataRecord refRec : refRecs) {
                        String uniqueSessionId = refRec.getIdentifier().toString()
                            .substring(SharedStoreRecordType.REFERENCES.getType().length() + 1);

                        stat.setEndTime(refRec.getLastModified());
                        stat.setLength(refRec.getLength());

                        if (markers.containsKey(uniqueSessionId)) {
                            stat.setStartTime(markers.get(uniqueSessionId).getLastModified());
                        }

                        LineNumberReader reader = null;
                        try {
                            reader = new LineNumberReader(new InputStreamReader(refRec.getStream()));
                            while (reader.readLine() != null) {
                            }
                            stat.setNumLines(reader.getLineNumber());
                        } finally {
                            Closeables.close(reader, true);
                        }
                    }
                }
            }
        }
        return stats;
    }

    @Override
    public OperationsStatsMBean getOperationStats() {
        return stats;
    }

    @Override
    public OperationsStatsMBean getConsistencyOperationStats() {
        return consistencyStats;
    }

    /**
     * Mark and sweep. Main entry method for GC.
     *
     * @param markOnly whether to mark only
     * @param forceBlobRetrieve force retrieve blob ids
     * @throws Exception the exception
     */
    protected void markAndSweep(boolean markOnly, boolean forceBlobRetrieve) throws Exception {
        statsCollector.start();
        boolean threw = true;
        GarbageCollectorFileState fs = new GarbageCollectorFileState(root);
        Stopwatch sw = Stopwatch.createStarted();

        try {
            LOG.info("Starting Blob garbage collection with markOnly [{}]", markOnly);

            long markStart = System.currentTimeMillis();
            long markFinish;
            try {
                mark(fs);
            } finally {
                markFinish = sw.elapsed(TimeUnit.MILLISECONDS);
                statsCollector.updateMarkDuration(markFinish, TimeUnit.MILLISECONDS);
                LOG.info("Blob garbage collection Mark completed in {} ({} ms).",
                    sw.toString(), sw.elapsed(TimeUnit.MILLISECONDS));
            }

            if (!markOnly) {
                long deleteCount;
                try {
                    deleteCount = sweep(fs, markStart, forceBlobRetrieve);
                    threw = false;
                } finally {
                    sw.stop();
                    statsCollector.updateSweepDuration(sw.elapsed(TimeUnit.MILLISECONDS) - markFinish, TimeUnit.MILLISECONDS);
                }

                long maxTime = getMaxModifiedTime(markStart) > 0 ? getMaxModifiedTime(markStart) : markStart;

                LOG.info("Blob garbage collection completed in {} ({} ms). Number of blobs deleted [{}] with max modification time of [{}]",
                        sw.toString(), sw.elapsed(TimeUnit.MILLISECONDS), deleteCount, timestampToString(maxTime));
            }
        } catch (Exception e) {
            statsCollector.finishFailure();
            LOG.error("Blob garbage collection error", e);
            throw e;
        } finally {
            statsCollector.updateDuration(sw.elapsed(TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS);
            if (!LOG.isTraceEnabled() && !traceOutput) {
                Closeables.close(fs, threw);
            }
        }
    }

    /**
     * Mark phase of the GC.
     * @param fs the garbage collector file state
     */
    protected void mark(GarbageCollectorFileState fs) throws IOException, DataStoreException {
        LOG.debug("Starting mark phase of the garbage collector");

        String uniqueSuffix = UUID.randomUUID().toString();

        // Create a time marker in the data store if applicable
        GarbageCollectionType.get(blobStore).addMarkedStartMarker(blobStore, repoId, uniqueSuffix);

        // Mark all used references
        iterateNodeTree(fs, false);

        // Move the marked references file to the data store meta area if applicable
        GarbageCollectionType.get(blobStore).addMarked(blobStore, fs, repoId, uniqueSuffix);

        LOG.debug("Ending mark phase of the garbage collector");
    }

    /**
     * Difference phase where the GC candidates are identified.
     * 
     * @param fs the garbage collector file state
     * @throws IOException
     *             Signals that an I/O exception has occurred.
     */
    private void difference(GarbageCollectorFileState fs) throws IOException {
        LOG.debug("Starting difference phase of the garbage collector");

        FileLineDifferenceIterator iter = new FileLineDifferenceIterator(
                fs.getMarkedRefs(),
                fs.getAvailableRefs(),
                transformer);
        int candidates = FileIOUtils.writeStrings(iter, fs.getGcCandidates(), true);
        LOG.debug("Found candidates - " + candidates);

        LOG.debug("Ending difference phase of the garbage collector");
    }

    /**
     * Sweep phase of gc candidate deletion.
     * <p>
     * Performs the following steps depending upon the type of the blob store refer
     * {@link org.apache.jackrabbit.oak.plugins.blob.SharedDataStore.Type}:
     *
     * <ul>
     *     <li>Shared</li>
     *     <li>
     *     <ul>
     *      <li> Merge all marked references (from the mark phase run independently) available in the data store meta
     *          store (from all configured independent repositories).
     *      <li> Retrieve all blob ids available.
     *      <li> Diffs the 2 sets above to retrieve list of blob ids not used.
     *      <li> Deletes only blobs created after
     *          (earliest time stamp of the marked references - #maxLastModifiedInterval) from the above set.
     *     </ul>
     *     </li>
     *
     *     <li>Default</li>
     *     <li>
     *     <ul>
     *      <li> Mark phase already run.
     *      <li> Retrieve all blob ids available.
     *      <li> Diffs the 2 sets above to retrieve list of blob ids not used.
     *      <li> Deletes only blobs created after
     *          (time stamp of the marked references - #maxLastModifiedInterval).
     *     </ul>
     *     </li>
     * </ul>
     *
     * @return the number of blobs deleted
     * @throws Exception the exception
     * @param fs the garbage collector file state
     * @param markStart the start time of mark to take as reference for deletion
     * @param forceBlobRetrieve
     */
    protected long sweep(GarbageCollectorFileState fs, long markStart, boolean forceBlobRetrieve) throws Exception {
        long earliestRefAvailTime;
        // Merge all the blob references available from all the reference files in the data store meta store
        // Only go ahead if merge succeeded
        try {
            earliestRefAvailTime =
                    GarbageCollectionType.get(blobStore).mergeAllMarkedReferences(blobStore, fs);
            LOG.debug("Earliest reference available for timestamp [{}]", earliestRefAvailTime);
            earliestRefAvailTime = (earliestRefAvailTime < markStart ? earliestRefAvailTime : markStart);
        } catch (Exception e) {
            return 0;
        }

        // Find all blob references after iterating over the whole repository
        (new BlobIdRetriever(fs, forceBlobRetrieve)).call();

        // Calculate the references not used
        difference(fs);
        long count = 0;
        long deleted = 0;
        
        long maxModifiedTime = getMaxModifiedTime(earliestRefAvailTime);
        LOG.debug("Starting sweep phase of the garbage collector");
        LOG.debug("Sweeping blobs with modified time > than the configured max deleted time ({}). ",
                timestampToString(maxModifiedTime));

        BufferedWriter removesWriter = null;
        LineIterator iterator = null;
        long deletedSize = 0;
        int numDeletedSizeAvailable = 0;
        try {
            removesWriter = Files.newWriter(fs.getGarbage(), Charsets.UTF_8);
            ArrayDeque<String> removesQueue = new ArrayDeque<String>();
            iterator =
                    FileUtils.lineIterator(fs.getGcCandidates(), Charsets.UTF_8.name());

            Iterator<List<String>> partitions = Iterators.partition(iterator, getBatchCount());
            while (partitions.hasNext()) {
                List<String> ids = partitions.next();
                count += ids.size();
                deleted += BlobCollectionType.get(blobStore)
                    .sweepInternal(blobStore, ids, removesQueue, maxModifiedTime);
                saveBatchToFile(newArrayList(removesQueue), removesWriter);

                for(String deletedId : removesQueue) {
                    // Estimate the size of the blob
                    long length = DataStoreBlobStore.BlobId.of(deletedId).getLength();
                    if (length != -1) {
                        deletedSize += length;
                        numDeletedSizeAvailable += 1;
                    }
                }
                removesQueue.clear();
            }
        } finally {
            LineIterator.closeQuietly(iterator);
            closeQuietly(removesWriter);
        }

        BlobCollectionType.get(blobStore).handleRemoves(blobStore, fs.getGarbage(), fs.getMarkedRefs());

        if(count != deleted) {
            LOG.warn("Deleted only [{}] blobs entries from the [{}] candidates identified. This may happen if blob " 
                         + "modified time is > "
                         + "than the max deleted time ({})", deleted, count,
                        timestampToString(maxModifiedTime));
        }

        if (deletedSize > 0) {
            LOG.info("Estimated size recovered for {} deleted blobs is {} ({} bytes)",
                numDeletedSizeAvailable,
                org.apache.jackrabbit.oak.commons.IOUtils.humanReadableByteCount(deletedSize), deletedSize);
        }

        statsCollector.updateNumCandidates(count);
        statsCollector.updateNumDeleted(deleted);
        statsCollector.updateTotalSizeDeleted(deletedSize);

        // Remove all the merged marked references
        GarbageCollectionType.get(blobStore).removeAllMarkedReferences(blobStore);
        LOG.debug("Ending sweep phase of the garbage collector");
        return deleted;
    }

    private int getBatchCount() {
        return batchCount;
    }

    /**
     * 3 possibilities
     *  - If maxLastModifiedInterval <= 0 then return 0 which is interpreted as current by delete call
     *      (For testing purposes only)
     *  - If oldest checkpoint creation date > 0 then reference time is the earliest of that and the parameter
     *      maxModificationReferenceTime
     *  - Else the parameter maxModificationReferenceTime is used as the reference time
     *
     * @param maxModificationReferenceTime typically the mark phase start time (could be 0 for tests)
     * @return max modified time of blobs to be considered for deletion
     */
    private long getMaxModifiedTime(long maxModificationReferenceTime) {
        if (maxLastModifiedInterval <= 0) {
            return 0;
        }

        long oldestCheckpoint = -1;
        if (checkpointMbean != null) {
            oldestCheckpoint = checkpointMbean.getOldestCheckpointCreationDate().getTime();
            LOG.debug("Oldest checkpoint data retrieved {} ", oldestCheckpoint);
        }
        LOG.debug("maxModificationReferenceTime {} ", maxModificationReferenceTime);

        maxModificationReferenceTime = maxModificationReferenceTime <= 0 ?
                                            System.currentTimeMillis() : maxModificationReferenceTime;
        long calculatedReferenceTime = (oldestCheckpoint <= 0 ? maxModificationReferenceTime :
                Math.min(maxModificationReferenceTime, oldestCheckpoint));
        LOG.debug("Calculated reference time {} ", calculatedReferenceTime);

        return (calculatedReferenceTime - maxLastModifiedInterval);
    }

    /**
     * Save batch to file.
     */
    static void saveBatchToFile(List<String> ids, BufferedWriter writer) throws IOException {
        for (String id : ids) {
            FileIOUtils.writeAsLine(writer, id, true);
        }
        writer.flush();
    }

    /**
     * Iterates the complete node tree and collect all blob references
     * @param fs the garbage collector file state
     * @param logPath whether to log path in the file or not
     */
    protected void iterateNodeTree(GarbageCollectorFileState fs, final boolean logPath) throws IOException {
        final BufferedWriter writer = Files.newWriter(fs.getMarkedRefs(), Charsets.UTF_8);
        final AtomicInteger count = new AtomicInteger();
        try {
            marker.collectReferences(
                    new ReferenceCollector() {
                        private final boolean debugMode = LOG.isTraceEnabled();

                        @Override
                        public void addReference(String blobId, final String nodeId) {
                            if (debugMode) {
                                LOG.trace("BlobId : {}, NodeId : {}", blobId, nodeId);
                            }

                            try {
                                Iterator<String> idIter = blobStore.resolveChunks(blobId);
                                final Joiner delimJoiner = Joiner.on(DELIM).skipNulls();
                                Iterator<List<String>> partitions = Iterators.partition(idIter, getBatchCount());
                                while (partitions.hasNext()) {
                                    List<String> idBatch = Lists.transform(partitions.next(), new Function<String,
                                        String>() {
                                        @Nullable @Override
                                        public String apply(@Nullable String id) {
                                            if (logPath) {
                                                return delimJoiner.join(id, nodeId);
                                            }
                                            return id;
                                        }
                                    });
                                    if (debugMode) {
                                        LOG.trace("chunkIds : {}", idBatch);
                                    }
                                    count.getAndAdd(idBatch.size());
                                    saveBatchToFile(idBatch, writer);
                                }

                                if (count.get() % getBatchCount() == 0) {
                                    LOG.info("Collected ({}) blob references", count.get());
                                }
                            } catch (Exception e) {
                                throw new RuntimeException("Error in retrieving references", e);
                            }
                        }
                    }
            );
            LOG.info("Number of valid blob references marked under mark phase of " +
                    "Blob garbage collection [{}]", count.get());
            // sort the marked references with the first part of the key
            sort(fs.getMarkedRefs(),
                new Comparator<String>() {
                    @Override
                    public int compare(String s1, String s2) {
                        return s1.split(DELIM)[0].compareTo(s2.split(DELIM)[0]);
                    }
                });
        } finally {
            closeQuietly(writer);
        }
    }
    
    /**
     * Checks for the DataStore consistency and reports the number of missing blobs still referenced.
     * 
     * @return the missing blobs
     * @throws Exception
     */
    @Override
    public long checkConsistency() throws Exception {
        consistencyStatsCollector.start();
        Stopwatch sw = Stopwatch.createStarted();

        boolean threw = true;
        GarbageCollectorFileState fs = new GarbageCollectorFileState(root);
        long candidates = 0;

        try {
            LOG.info("Starting blob consistency check");
    
            // Find all blobs available in the blob store
            ListenableFutureTask<Integer> blobIdRetriever = ListenableFutureTask.create(new BlobIdRetriever(fs,
                true));
            executor.execute(blobIdRetriever);
    
            // Mark all used blob references
            iterateNodeTree(fs, true);
            consistencyStatsCollector.updateMarkDuration(sw.elapsed(TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS);

            try {
                blobIdRetriever.get();
            } catch (ExecutionException e) {
                LOG.warn("Error occurred while fetching all the blobIds from the BlobStore");
                threw = false;
                throw e;
            }
            
            LOG.trace("Starting difference phase of the consistency check");
            FileLineDifferenceIterator iter = new FileLineDifferenceIterator(
                fs.getAvailableRefs(),
                fs.getMarkedRefs(),
                transformer);
            // If tracking then also filter ids being tracked which are active deletions for lucene
            candidates = BlobCollectionType.get(blobStore).filter(blobStore, iter, fs);

            LOG.trace("Ending difference phase of the consistency check");
            LOG.info("Consistency check found [{}] missing blobs", candidates);

            if (candidates > 0) {
                LOG.warn("Consistency check failure in the the blob store : {}, check missing candidates in file {}",
                            blobStore, fs.getGcCandidates().getAbsolutePath());
                consistencyStatsCollector.finishFailure();
                consistencyStatsCollector.updateNumDeleted(candidates);
            }
        } finally {
            if (!traceOutput && (!LOG.isTraceEnabled() && candidates == 0)) {
                Closeables.close(fs, threw);
            }
            sw.stop();
            consistencyStatsCollector.updateDuration(sw.elapsed(TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS);
        }
        return candidates;
    }

    public void setTraceOutput(boolean trace) {
        traceOutput = trace;
    }

    /**
     * BlobIdRetriever class to retrieve all blob ids.
     */
    private class BlobIdRetriever implements Callable<Integer> {
        private final GarbageCollectorFileState fs;
        private final boolean forceRetrieve;

        public BlobIdRetriever(GarbageCollectorFileState fs, boolean forceBlobRetrieve) {
            this.fs = fs;
            this.forceRetrieve = forceBlobRetrieve;
        }
    
        @Override
        public Integer call() throws Exception {
            if (!forceRetrieve) {
                BlobCollectionType.get(blobStore).retrieve(blobStore, fs, getBatchCount());
                LOG.info("Length of blob ids file retrieved from tracker {}", fs.getAvailableRefs().length());
            }

            // If the length is 0 then references not available from the tracker
            // retrieve from the data store
            if (fs.getAvailableRefs().length() <= 0) {
                BlobCollectionType.DEFAULT.retrieve(blobStore, fs, getBatchCount());
                LOG.info("Length of blob ids file retrieved {}", fs.getAvailableRefs().length());

                BlobCollectionType.get(blobStore).track(blobStore, fs);
            }
            return 0;
        }
    }

    /**
     * Provides a readable string for given timestamp
     */
    private static String timestampToString(long timestamp){
        return (new Timestamp(timestamp) + "00").substring(0, 23);
    }

    /**
     * Defines different data store types from the garbage collection perspective and encodes the divergent behavior.
     * <ul></ul>
     */
    enum GarbageCollectionType {
        SHARED {
            /**
             * Remove the maked references and the marked markers from the blob store root. Default NOOP.
             * 
             * @param blobStore the blobStore instance
             */
            @Override
            void removeAllMarkedReferences(GarbageCollectableBlobStore blobStore) {
                ((SharedDataStore) blobStore).deleteAllMetadataRecords(SharedStoreRecordType.REFERENCES.getType());
                ((SharedDataStore) blobStore).deleteAllMetadataRecords(SharedStoreRecordType.MARKED_START_MARKER.getType());
            }

            /**
             * Merge all marked references available from all repositories and return the earliest time of the references.
             * 
             * @param blobStore the blob store
             * @param fs the fs
             * @return the long the earliest time of the available references
             * @throws IOException Signals that an I/O exception has occurred.
             * @throws DataStoreException the data store exception
             */
            @Override
            long mergeAllMarkedReferences(GarbageCollectableBlobStore blobStore,
                    GarbageCollectorFileState fs)
                    throws IOException, DataStoreException {

                List<DataRecord> refFiles =
                    ((SharedDataStore) blobStore).getAllMetadataRecords(SharedStoreRecordType.REFERENCES.getType());

                // Get all the repositories registered
                List<DataRecord> repoFiles =
                    ((SharedDataStore) blobStore).getAllMetadataRecords(SharedStoreRecordType.REPOSITORY.getType());

                // Retrieve repos for which reference files have not been created
                Set<String> unAvailRepos =
                        SharedDataStoreUtils.refsNotAvailableFromRepos(repoFiles, refFiles);
                if (unAvailRepos.isEmpty()) {
                    // List of files to be merged
                    List<File> files = newArrayList();
                    for (DataRecord refFile : refFiles) {
                        File file = copy(refFile.getStream());
                        files.add(file);
                    }

                    merge(files, fs.getMarkedRefs());
                    
                    // Get the timestamp to indicate the earliest mark phase start
                    List<DataRecord> markerFiles =
                        ((SharedDataStore) blobStore).getAllMetadataRecords(
                                                        SharedStoreRecordType.MARKED_START_MARKER.getType());
                    long earliestMarker = SharedDataStoreUtils.getEarliestRecord(markerFiles).getLastModified();
                    LOG.trace("Earliest marker timestamp {}", earliestMarker);

                    long earliestRef = SharedDataStoreUtils.getEarliestRecord(refFiles).getLastModified();
                    LOG.trace("Earliest ref timestamp {}", earliestRef);

                    return (earliestMarker < earliestRef ? earliestMarker : earliestRef);
                } else {
                    LOG.error("Not all repositories have marked references available : {}", unAvailRepos);
                    throw new IOException("Not all repositories have marked references available");
                }
            }

            /**
             * Adds the marked references to the blob store root. Default NOOP
             * 
             * @param blobStore the blob store
             * @param fs the fs
             * @param repoId the repo id
             * @param uniqueSuffix the unique session suffix
             * @throws DataStoreException the data store exception
             * @throws IOException Signals that an I/O exception has occurred.
             */
            @Override
            void addMarked(GarbageCollectableBlobStore blobStore, GarbageCollectorFileState fs, String repoId,
                String uniqueSuffix) throws DataStoreException, IOException {
                boolean exists = ((SharedDataStore) blobStore)
                    .metadataRecordExists(SharedStoreRecordType.REFERENCES.getNameFromId(repoId));
                if (exists) {
                    LOG.info("References for repository id {} already exists. Creating a duplicate one. "
                        + "Please check for inadvertent sharing of repository id by different repositories", repoId);
                }

                ((SharedDataStore) blobStore).addMetadataRecord(fs.getMarkedRefs(), SharedStoreRecordType.REFERENCES
                    .getNameFromIdPrefix(repoId, uniqueSuffix));
            }
            
            @Override
            public void addMarkedStartMarker(GarbageCollectableBlobStore blobStore, String repoId,
                String uniqueSuffix) {
                try {
                    ((SharedDataStore) blobStore).addMetadataRecord(new ByteArrayInputStream(new byte[0]),
                        SharedStoreRecordType.MARKED_START_MARKER
                            .getNameFromIdPrefix(repoId, uniqueSuffix));
                } catch (DataStoreException e) {
                    LOG.debug("Error creating marked time marker for repo : {}", repoId);
                }
            }
        },
        DEFAULT;

        void removeAllMarkedReferences(GarbageCollectableBlobStore blobStore) {}

        void addMarked(GarbageCollectableBlobStore blobStore, GarbageCollectorFileState fs, String repoId,
            String uniqueSuffix) throws DataStoreException, IOException {}

        long mergeAllMarkedReferences(GarbageCollectableBlobStore blobStore,
                GarbageCollectorFileState fs)
                throws IOException, DataStoreException {
            // throw id the marked refs not available.
            if (!fs.getMarkedRefs().exists() || fs.getMarkedRefs().length() == 0) {
                throw new IOException("Marked references not available");
            }
            return fs.getMarkedRefs().lastModified();
        }

        public static GarbageCollectionType get(GarbageCollectableBlobStore blobStore) {
            if (SharedDataStoreUtils.isShared(blobStore)) {
                return SHARED;
            }
            return DEFAULT;
        }
    
        public void addMarkedStartMarker(GarbageCollectableBlobStore blobStore, String repoId, String uniqueSuffix) {}
    }

    /**
     * Defines different blob collection types and encodes the divergent behavior.
     * <ul></ul>
     */
    private enum BlobCollectionType {
        TRACKER {
            @Override
            void retrieve(GarbageCollectableBlobStore blobStore,
                    GarbageCollectorFileState fs, int batchCount) throws Exception {
                ((BlobTrackingStore) blobStore).getTracker()
                    .get(fs.getAvailableRefs().getAbsolutePath());
            }

            @Override
            void handleRemoves(GarbageCollectableBlobStore blobStore, File removedIds, File markedRefs) throws IOException {
                BlobTrackingStore store = (BlobTrackingStore) blobStore;
                BlobIdTracker tracker = (BlobIdTracker) store.getTracker();
                tracker.remove(removedIds);
                tracker.getDeleteTracker().reconcile(markedRefs);
            }

            @Override
            void track(GarbageCollectableBlobStore blobStore,
                GarbageCollectorFileState fs) {
                try {
                    File f = File.createTempFile("blobiddownload", null);
                    copyFile(fs.getAvailableRefs(), f);
                    ((BlobTrackingStore) blobStore).getTracker().add(f);
                } catch (IOException e) {
                    LOG.warn("Unable to track blob ids locally");
                }
            }

            @Override
            public int filter(GarbageCollectableBlobStore blobStore, FileLineDifferenceIterator iter,
                GarbageCollectorFileState fs) throws IOException {
                // Write the original candidates
                FileIOUtils.writeStrings(iter, fs.getGcCandidates(), true);

                // Filter the ids actively deleted
                BlobTrackingStore store = (BlobTrackingStore) blobStore;
                BlobIdTracker tracker = (BlobIdTracker) store.getTracker();

                // Move the candidates identified to a temp file
                File candTemp = createTempFile("candTemp", null);
                copyFile(fs.getGcCandidates(), candTemp);

                Iterator<String> filter = tracker.getDeleteTracker().filter(candTemp);
                try {
                    return FileIOUtils.writeStrings(filter, fs.getGcCandidates(), true);
                } finally {
                    if (filter != null && filter instanceof FileLineDifferenceIterator) {
                        ((FileLineDifferenceIterator) filter).close();
                    }

                    if (candTemp != null) {
                        candTemp.delete();
                    }
                }
            }
        },
        DEFAULT;

        /**
         * Deletes the given batch by deleting individually to exactly know the actual deletes.
         */
        long sweepInternal(GarbageCollectableBlobStore blobStore, List<String> ids,
            ArrayDeque<String> exceptionQueue, long maxModified) {
            long totalDeleted = 0;
            LOG.trace("Blob ids to be deleted {}", ids);
            for (String id : ids) {
                try {
                    long deleted = blobStore.countDeleteChunks(newArrayList(id), maxModified);
                    if (deleted != 1) {
                        LOG.debug("Blob [{}] not deleted", id);
                    } else {
                        exceptionQueue.add(id);
                        totalDeleted += 1;
                    }
                } catch (Exception e) {
                    LOG.warn("Error occurred while deleting blob with id [{}]", id, e);
                }
            }
            return totalDeleted;
        }

        /**
         * Retrieve the put the list of available blobs in the file.
         *
         * @param blobStore
         * @param fs
         * @param batchCount
         * @throws Exception
         */
        void retrieve(GarbageCollectableBlobStore blobStore,
                GarbageCollectorFileState fs, int batchCount) throws Exception {
            LOG.debug("Starting retrieve of all blobs");
            int blobsCount = 0;
            Iterator<String> idsIter = null;
            try {
                idsIter = blobStore.getAllChunkIds(0);
                blobsCount = FileIOUtils.writeStrings(idsIter, fs.getAvailableRefs(), true, LOG, "Retrieved blobs - ");

                // sort the file
                sort(fs.getAvailableRefs());
                LOG.info("Number of blobs present in BlobStore : [{}] ", blobsCount);
            } finally {
                if (idsIter instanceof Closeable) {
                    try {
                        Closeables.close((Closeable) idsIter, false);
                    } catch (Exception e) {
                        LOG.debug("Error closing iterator");
                    }
                }
            }
        }

        /**
         * Hook to handle all the removed ids.
         *
         * @param blobStore
         * @param removedIds
         * @param markedRefs
         * @throws IOException
         */
        void handleRemoves(GarbageCollectableBlobStore blobStore, File removedIds, File markedRefs) throws IOException {
            FileUtils.forceDelete(removedIds);
        }

        /**
         * Tracker may want to track this file
         *
         * @param blobStore
         * @param fs
         */
        void track(GarbageCollectableBlobStore blobStore, GarbageCollectorFileState fs) {
        }

        public static BlobCollectionType get(GarbageCollectableBlobStore blobStore) {
            if (blobStore instanceof BlobTrackingStore) {
                BlobTracker tracker = ((BlobTrackingStore) blobStore).getTracker();
                if (tracker != null) {
                    return TRACKER;
                }
            }
            return DEFAULT;
        }

        public int filter(GarbageCollectableBlobStore blobStore, FileLineDifferenceIterator iter,
            GarbageCollectorFileState fs) throws IOException {
            return FileIOUtils.writeStrings(iter, fs.getGcCandidates(), true);
        }
    }

    class GarbageCollectionOperationStats implements OperationsStatsMBean {
        static final String NAME = "DataStoreGarbageCollection";
        static final String CONSISTENCY_NAME = "DataStoreConsistencyCheck";
        static final String START = "COUNTER";
        static final String FINISH_FAILURE = "FAILURE";
        static final String DURATION = "ACTIVE_TIMER";
        static final String MARK_DURATION = "MARK_TIMER";
        static final String SWEEP_DURATION = "SWEEP_TIMER";
        static final String NUM_BLOBS_DELETED = "NUM_BLOBS_DELETED";
        static final String TOTAL_SIZE_DELETED = "TOTAL_SIZE_DELETED";
        static final String NUM_CANDIDATES = "NUM_CANDIDATES";

        private final String typeName;

        private CounterStats startCounter;
        private CounterStats finishFailureCounter;
        private CounterStats numDeletedCounter;
        private CounterStats totalSizeDeletedCounter;
        private CounterStats numCandidatesCounter;
        private TimerStats duration;
        private final TimerStats markDuration;
        private final TimerStats sweepDuration;
        private final OperationStatsCollector collector;

        GarbageCollectionOperationStats(StatisticsProvider sp, String typeName) {
            this.typeName = typeName;

            this.startCounter = sp.getCounterStats(getMetricName(START), StatsOptions.METRICS_ONLY);
            this.finishFailureCounter = sp.getCounterStats(getMetricName(FINISH_FAILURE), StatsOptions.METRICS_ONLY);
            this.numDeletedCounter = sp.getCounterStats(getMetricName(NUM_BLOBS_DELETED), StatsOptions.METRICS_ONLY);
            this.totalSizeDeletedCounter = sp.getCounterStats(getMetricName(TOTAL_SIZE_DELETED), StatsOptions.METRICS_ONLY);
            this.numCandidatesCounter = sp.getCounterStats(getMetricName(NUM_CANDIDATES), StatsOptions.METRICS_ONLY);
            this.duration = sp.getTimer(getMetricName(DURATION), StatsOptions.METRICS_ONLY);
            this.markDuration = sp.getTimer(getMetricName(MARK_DURATION), StatsOptions.METRICS_ONLY);
            this.sweepDuration = sp.getTimer(getMetricName(SWEEP_DURATION), StatsOptions.METRICS_ONLY);

            this.collector = new OperationStatsCollector() {
                @Override
                public void start() {
                    startCounter.inc();
                }

                @Override
                public void finishFailure() {
                    finishFailureCounter.inc();
                }

                @Override
                public void updateNumDeleted(long num) {
                    numDeletedCounter.inc(num);
                }

                @Override
                public void updateNumCandidates(long num) {
                    numCandidatesCounter.inc(num);
                }

                @Override
                public void updateTotalSizeDeleted(long size) {
                    totalSizeDeletedCounter.inc(size);
                }

                @Override
                public void updateDuration(long time, TimeUnit timeUnit) {
                    duration.update(time, timeUnit);
                }

                @Override public void updateMarkDuration(long time, TimeUnit timeUnit) {
                    markDuration.update(time, timeUnit);
                }

                @Override public void updateSweepDuration(long time, TimeUnit timeUnit) {
                    sweepDuration.update(time, timeUnit);
                }
            };
        }


        GarbageCollectionOperationStats(StatisticsProvider sp) {
            this(sp, NAME);
        }

        private String getMetricName(String name) {
            return getName() + "." + name;
        }

        protected OperationStatsCollector getCollector() {
            return collector;
        }

        @Override public String getName() {
            return TYPE + "." + typeName;
        }

        @Override public long getStartCount() {
            return startCounter.getCount();
        }

        @Override public long getFailureCount() {
            return finishFailureCounter.getCount();
        }

        @Override public long duration() {
            return duration.getCount();
        }

        @Override public long markDuration() {
            return markDuration.getCount();
        }

        @Override public long numDeleted() {
            return numDeletedCounter.getCount();
        }

        @Override public long sizeDeleted() {
            return totalSizeDeletedCounter.getCount();
        }
    }
}