/*
    * Copyright (C) 2008-2011, Google Inc.
    * Copyright (C) 2008, Shawn O. Pearce <spearce@spearce.org> and others
    *
    * This program and the accompanying materials are made available under the
    * terms of the Eclipse Distribution License v. 1.0 which is available at
    * https://www.eclipse.org/org/documents/edl-v10.php.
    *
    * SPDX-License-Identifier: BSD-3-Clause
   */
  
  package org.eclipse.jgit.internal.storage.dfs;
  
  import java.io.IOException;
  import java.util.concurrent.atomic.AtomicLong;
  import java.util.concurrent.atomic.AtomicReference;
  import java.util.concurrent.atomic.AtomicReferenceArray;
  import java.util.concurrent.locks.ReentrantLock;
  import java.util.function.Consumer;
  import java.util.stream.LongStream;
  
  import org.eclipse.jgit.internal.JGitText;
  import org.eclipse.jgit.internal.storage.pack.PackExt;
  
  /**
   * Caches slices of a
   * {@link org.eclipse.jgit.internal.storage.dfs.BlockBasedFile} in memory for
   * faster read access.
   * <p>
   * The DfsBlockCache serves as a Java based "buffer cache", loading segments of
   * a BlockBasedFile into the JVM heap prior to use. As JGit often wants to do
   * reads of only tiny slices of a file, the DfsBlockCache tries to smooth out
   * these tiny reads into larger block-sized IO operations.
   * <p>
   * Whenever a cache miss occurs, loading is invoked by exactly one thread for
   * the given <code>(DfsStreamKey,position)</code> key tuple. This is ensured by
   * an array of locks, with the tuple hashed to a lock instance.
   * <p>
   * Its too expensive during object access to be accurate with a least recently
   * used (LRU) algorithm. Strictly ordering every read is a lot of overhead that
   * typically doesn't yield a corresponding benefit to the application. This
   * cache implements a clock replacement algorithm, giving each block at least
   * one chance to have been accessed during a sweep of the cache to save itself
   * from eviction. The number of swipe chances is configurable per pack
   * extension.
   * <p>
   * Entities created by the cache are held under hard references, preventing the
   * Java VM from clearing anything. Blocks are discarded by the replacement
   * algorithm when adding a new block would cause the cache to exceed its
   * configured maximum size.
   * <p>
   * The key tuple is passed through to methods as a pair of parameters rather
   * than as a single Object, thus reducing the transient memory allocations of
   * callers. It is more efficient to avoid the allocation, as we can't be 100%
   * sure that a JIT would be able to stack-allocate a key tuple.
   * <p>
   * The internal hash table does not expand at runtime, instead it is fixed in
   * size at cache creation time. The internal lock table used to gate load
   * invocations is also fixed in size.
   */
  public final class DfsBlockCache {
  	private static volatile DfsBlockCache cache;
  
  	static {
  		reconfigure(new DfsBlockCacheConfig());
  	}
  
  	/**
  	 * Modify the configuration of the window cache.
  	 * <p>
  	 * The new configuration is applied immediately, and the existing cache is
  	 * cleared.
  	 *
  	 * @param cfg
  	 *            the new window cache configuration.
  	 * @throws java.lang.IllegalArgumentException
  	 *             the cache configuration contains one or more invalid
  	 *             settings, usually too low of a limit.
  	 */
  	public static void reconfigure(DfsBlockCacheConfig cfg) {
  		cache = new DfsBlockCache(cfg);
  	}
  
  	/**
  	 * Get the currently active DfsBlockCache.
  	 *
  	 * @return the currently active DfsBlockCache.
  	 */
  	public static DfsBlockCache getInstance() {
  		return cache;
  	}
  
  	/** Number of entries in {@link #table}. */
  	private final int tableSize;
  
  	/** Hash bucket directory; entries are chained below. */
  	private final AtomicReferenceArray<HashEntry> table;
  
  	/**
 	 * Locks to prevent concurrent loads for same (PackFile,position) block. The
 	 * number of locks is {@link DfsBlockCacheConfig#getConcurrencyLevel()} to
 	 * cap the overall concurrent block loads.
 	 */
 	private final ReentrantLock[] loadLocks;
 
 	/**
 	 * A separate pool of locks per pack extension to prevent concurrent loads
 	 * for same index or bitmap from PackFile.
 	 */
 	private final ReentrantLock[][] refLocks;
 
 	/** Maximum number of bytes the cache should hold. */
 	private final long maxBytes;
 
 	/** Pack files smaller than this size can be copied through the cache. */
 	private final long maxStreamThroughCache;
 
 	/**
 	 * Suggested block size to read from pack files in.
 	 * <p>
 	 * If a pack file does not have a native block size, this size will be used.
 	 * <p>
 	 * If a pack file has a native size, a whole multiple of the native size
 	 * will be used until it matches this size.
 	 * <p>
 	 * The value for blockSize must be a power of 2.
 	 */
 	private final int blockSize;
 
 	/** As {@link #blockSize} is a power of 2, bits to shift for a / blockSize. */
 	private final int blockSizeShift;
 
 	/**
 	 * Number of times a block was found in the cache, per pack file extension.
 	 */
 	private final AtomicReference<AtomicLong[]> statHit;
 
 	/**
 	 * Number of times a block was not found, and had to be loaded, per pack
 	 * file extension.
 	 */
 	private final AtomicReference<AtomicLong[]> statMiss;
 
 	/**
 	 * Number of blocks evicted due to cache being full, per pack file
 	 * extension.
 	 */
 	private final AtomicReference<AtomicLong[]> statEvict;
 
 	/**
 	 * Number of bytes currently loaded in the cache, per pack file extension.
 	 */
 	private final AtomicReference<AtomicLong[]> liveBytes;
 
 	/** Protects the clock and its related data. */
 	private final ReentrantLock clockLock;
 
 	/**
 	 * A consumer of object reference lock wait time milliseconds.  May be used to build a metric.
 	 */
 	private final Consumer<Long> refLockWaitTime;
 
 	/** Current position of the clock. */
 	private Ref clockHand;
 
 	/** Limits of cache hot count per pack file extension. */
 	private final int[] cacheHotLimits = new int[PackExt.values().length];
 
 	@SuppressWarnings("unchecked")
 	private DfsBlockCache(DfsBlockCacheConfig cfg) {
 		tableSize = tableSize(cfg);
 		if (tableSize < 1) {
 			throw new IllegalArgumentException(JGitText.get().tSizeMustBeGreaterOrEqual1);
 		}
 
 		table = new AtomicReferenceArray<>(tableSize);
 		int concurrencyLevel = cfg.getConcurrencyLevel();
 		loadLocks = new ReentrantLock[concurrencyLevel];
 		for (int i = 0; i < loadLocks.length; i++) {
 			loadLocks[i] = new ReentrantLock(true /* fair */);
 		}
 		refLocks = new ReentrantLock[PackExt.values().length][concurrencyLevel];
 		for (int i = 0; i < PackExt.values().length; i++) {
 			for (int j = 0; j < concurrencyLevel; ++j) {
 				refLocks[i][j] = new ReentrantLock(true /* fair */);
 			}
 		}
 
 		maxBytes = cfg.getBlockLimit();
 		maxStreamThroughCache = (long) (maxBytes * cfg.getStreamRatio());
 		blockSize = cfg.getBlockSize();
 		blockSizeShift = Integer.numberOfTrailingZeros(blockSize);
 
 		clockLock = new ReentrantLock(true /* fair */);
 		String none = ""; //$NON-NLS-1$
 		clockHand = new Ref<>(
 				DfsStreamKey.of(new DfsRepositoryDescription(none), none, null),
 				-1, 0, null);
 		clockHand.next = clockHand;
 
 		statHit = new AtomicReference<>(newCounters());
 		statMiss = new AtomicReference<>(newCounters());
 		statEvict = new AtomicReference<>(newCounters());
 		liveBytes = new AtomicReference<>(newCounters());
 
 		refLockWaitTime = cfg.getRefLockWaitTimeConsumer();
 
 		for (int i = 0; i < PackExt.values().length; ++i) {
 			Integer limit = cfg.getCacheHotMap().get(PackExt.values()[i]);
 			if (limit != null && limit.intValue() > 0) {
 				cacheHotLimits[i] = limit.intValue();
 			} else {
 				cacheHotLimits[i] = DfsBlockCacheConfig.DEFAULT_CACHE_HOT_MAX;
 			}
 		}
 	}
 
 	boolean shouldCopyThroughCache(long length) {
 		return length <= maxStreamThroughCache;
 	}
 
 	/**
 	 * Get total number of bytes in the cache, per pack file extension.
 	 *
 	 * @return total number of bytes in the cache, per pack file extension.
 	 */
 	public long[] getCurrentSize() {
 		return getStatVals(liveBytes);
 	}
 
 	/**
 	 * Get 0..100, defining how full the cache is.
 	 *
 	 * @return 0..100, defining how full the cache is.
 	 */
 	public long getFillPercentage() {
 		return LongStream.of(getCurrentSize()).sum() * 100 / maxBytes;
 	}
 
 	/**
 	 * Get number of requests for items in the cache, per pack file extension.
 	 *
 	 * @return number of requests for items in the cache, per pack file
 	 *         extension.
 	 */
 	public long[] getHitCount() {
 		return getStatVals(statHit);
 	}
 
 	/**
 	 * Get number of requests for items not in the cache, per pack file
 	 * extension.
 	 *
 	 * @return number of requests for items not in the cache, per pack file
 	 *         extension.
 	 */
 	public long[] getMissCount() {
 		return getStatVals(statMiss);
 	}
 
 	/**
 	 * Get total number of requests (hit + miss), per pack file extension.
 	 *
 	 * @return total number of requests (hit + miss), per pack file extension.
 	 */
 	public long[] getTotalRequestCount() {
 		AtomicLong[] hit = statHit.get();
 		AtomicLong[] miss = statMiss.get();
 		long[] cnt = new long[Math.max(hit.length, miss.length)];
 		for (int i = 0; i < hit.length; i++) {
 			cnt[i] += hit[i].get();
 		}
 		for (int i = 0; i < miss.length; i++) {
 			cnt[i] += miss[i].get();
 		}
 		return cnt;
 	}
 
 	/**
 	 * Get hit ratios
 	 *
 	 * @return hit ratios
 	 */
 	public long[] getHitRatio() {
 		AtomicLong[] hit = statHit.get();
 		AtomicLong[] miss = statMiss.get();
 		long[] ratio = new long[Math.max(hit.length, miss.length)];
 		for (int i = 0; i < ratio.length; i++) {
 			if (i >= hit.length) {
 				ratio[i] = 0;
 			} else if (i >= miss.length) {
 				ratio[i] = 100;
 			} else {
 				long hitVal = hit[i].get();
 				long missVal = miss[i].get();
 				long total = hitVal + missVal;
 				ratio[i] = total == 0 ? 0 : hitVal * 100 / total;
 			}
 		}
 		return ratio;
 	}
 
 	/**
 	 * Get number of evictions performed due to cache being full, per pack file
 	 * extension.
 	 *
 	 * @return number of evictions performed due to cache being full, per pack
 	 *         file extension.
 	 */
 	public long[] getEvictions() {
 		return getStatVals(statEvict);
 	}
 
 	/**
 	 * Quickly check if the cache contains block 0 of the given stream.
 	 * <p>
 	 * This can be useful for sophisticated pre-read algorithms to quickly
 	 * determine if a file is likely already in cache, especially small
 	 * reftables which may be smaller than a typical DFS block size.
 	 *
 	 * @param key
 	 *            the file to check.
 	 * @return true if block 0 (the first block) is in the cache.
 	 */
 	public boolean hasBlock0(DfsStreamKey key) {
 		HashEntry e1 = table.get(slot(key, 0));
 		DfsBlock v = scan(e1, key, 0);
 		return v != null && v.contains(key, 0);
 	}
 
 	private int hash(int packHash, long off) {
 		return packHash + (int) (off >>> blockSizeShift);
 	}
 
 	int getBlockSize() {
 		return blockSize;
 	}
 
 	private static int tableSize(DfsBlockCacheConfig cfg) {
 		final int wsz = cfg.getBlockSize();
 		final long limit = cfg.getBlockLimit();
 		if (wsz <= 0) {
 			throw new IllegalArgumentException(JGitText.get().invalidWindowSize);
 		}
 		if (limit < wsz) {
 			throw new IllegalArgumentException(JGitText.get().windowSizeMustBeLesserThanLimit);
 		}
 		return (int) Math.min(5 * (limit / wsz) / 2, Integer.MAX_VALUE);
 	}
 
 	/**
 	 * Look up a cached object, creating and loading it if it doesn't exist.
 	 *
 	 * @param file
 	 *            the pack that "contains" the cached object.
 	 * @param position
 	 *            offset within <code>pack</code> of the object.
 	 * @param ctx
 	 *            current thread's reader.
 	 * @param fileChannel
 	 *            supplier for channel to read {@code pack}.
 	 * @return the object reference.
 	 * @throws IOException
 	 *             the reference was not in the cache and could not be loaded.
 	 */
 	DfsBlock getOrLoad(BlockBasedFile file, long position, DfsReader ctx,
 			ReadableChannelSupplier fileChannel) throws IOException {
 		final long requestedPosition = position;
 		position = file.alignToBlock(position);
 
 		DfsStreamKey key = file.key;
 		int slot = slot(key, position);
 		HashEntry e1 = table.get(slot);
 		DfsBlock v = scan(e1, key, position);
 		if (v != null && v.contains(key, requestedPosition)) {
 			ctx.stats.blockCacheHit++;
 			getStat(statHit, key).incrementAndGet();
 			return v;
 		}
 
 		reserveSpace(blockSize, key);
 		ReentrantLock regionLock = lockFor(key, position);
 		regionLock.lock();
 		try {
 			HashEntry e2 = table.get(slot);
 			if (e2 != e1) {
 				v = scan(e2, key, position);
 				if (v != null) {
 					ctx.stats.blockCacheHit++;
 					getStat(statHit, key).incrementAndGet();
 					creditSpace(blockSize, key);
 					return v;
 				}
 			}
 
 			getStat(statMiss, key).incrementAndGet();
 			boolean credit = true;
 			try {
 				v = file.readOneBlock(position, ctx, fileChannel.get());
 				credit = false;
 			} finally {
 				if (credit) {
 					creditSpace(blockSize, key);
 				}
 			}
 			if (position != v.start) {
 				// The file discovered its blockSize and adjusted.
 				position = v.start;
 				slot = slot(key, position);
 				e2 = table.get(slot);
 			}
 
 			Ref<DfsBlock> ref = new Ref<>(key, position, v.size(), v);
 			ref.markHotter();
 			for (;;) {
 				HashEntry n = new HashEntry(clean(e2), ref);
 				if (table.compareAndSet(slot, e2, n)) {
 					break;
 				}
 				e2 = table.get(slot);
 			}
 			addToClock(ref, blockSize - v.size());
 		} finally {
 			regionLock.unlock();
 		}
 
 		// If the block size changed from the default, it is possible the block
 		// that was loaded is the wrong block for the requested position.
 		if (v.contains(file.key, requestedPosition)) {
 			return v;
 		}
 		return getOrLoad(file, requestedPosition, ctx, fileChannel);
 	}
 
 	@SuppressWarnings("unchecked")
 	private void reserveSpace(long reserve, DfsStreamKey key) {
 		clockLock.lock();
 		try {
 			long live = LongStream.of(getCurrentSize()).sum() + reserve;
 			if (maxBytes < live) {
 				Ref prev = clockHand;
 				Ref hand = clockHand.next;
 				do {
 					if (hand.isHot()) {
 						// Value was recently touched. Cache is still hot so
 						// give it another chance, but cool it down a bit.
 						hand.markColder();
 						prev = hand;
 						hand = hand.next;
 						continue;
 					} else if (prev == hand)
 						break;
 
 					// No recent access since last scan, kill
 					// value and remove from clock.
 					Ref dead = hand;
 					hand = hand.next;
 					prev.next = hand;
 					dead.next = null;
 					dead.value = null;
 					live -= dead.size;
 					getStat(liveBytes, dead.key).addAndGet(-dead.size);
 					getStat(statEvict, dead.key).incrementAndGet();
 				} while (maxBytes < live);
 				clockHand = prev;
 			}
 			getStat(liveBytes, key).addAndGet(reserve);
 		} finally {
 			clockLock.unlock();
 		}
 	}
 
 	private void creditSpace(long credit, DfsStreamKey key) {
 		clockLock.lock();
 		try {
 			getStat(liveBytes, key).addAndGet(-credit);
 		} finally {
 			clockLock.unlock();
 		}
 	}
 
 	@SuppressWarnings("unchecked")
 	private void addToClock(Ref ref, long credit) {
 		clockLock.lock();
 		try {
 			if (credit != 0) {
 				getStat(liveBytes, ref.key).addAndGet(-credit);
 			}
 			Ref ptr = clockHand;
 			ref.next = ptr.next;
 			ptr.next = ref;
 			clockHand = ref;
 		} finally {
 			clockLock.unlock();
 		}
 	}
 
 	void put(DfsBlock v) {
 		put(v.stream, v.start, v.size(), v);
 	}
 
 	/**
 	 * Look up a cached object, creating and loading it if it doesn't exist.
 	 *
 	 * @param key
 	 *            the stream key of the pack.
 	 * @param position
 	 *            the position in the key. The default should be 0.
 	 * @param loader
 	 *            the function to load the reference.
 	 * @return the object reference.
 	 * @throws IOException
 	 *             the reference was not in the cache and could not be loaded.
 	 */
 	<T> Ref<T> getOrLoadRef(
 			DfsStreamKey key, long position, RefLoader<T> loader)
 			throws IOException {
 		int slot = slot(key, position);
 		HashEntry e1 = table.get(slot);
 		Ref<T> ref = scanRef(e1, key, position);
 		if (ref != null) {
 			getStat(statHit, key).incrementAndGet();
 			return ref;
 		}
 
 		ReentrantLock regionLock = lockForRef(key);
 		long lockStart = System.currentTimeMillis();
 		regionLock.lock();
 		try {
 			HashEntry e2 = table.get(slot);
 			if (e2 != e1) {
 				ref = scanRef(e2, key, position);
 				if (ref != null) {
 					getStat(statHit, key).incrementAndGet();
 					return ref;
 				}
 			}
 
 			if (refLockWaitTime != null) {
 				refLockWaitTime.accept(
 						Long.valueOf(System.currentTimeMillis() - lockStart));
 			}
 			getStat(statMiss, key).incrementAndGet();
 			ref = loader.load();
 			ref.markHotter();
 			// Reserve after loading to get the size of the object
 			reserveSpace(ref.size, key);
 			for (;;) {
 				HashEntry n = new HashEntry(clean(e2), ref);
 				if (table.compareAndSet(slot, e2, n)) {
 					break;
 				}
 				e2 = table.get(slot);
 			}
 			addToClock(ref, 0);
 		} finally {
 			regionLock.unlock();
 		}
 		return ref;
 	}
 
 	<T> Ref<T> putRef(DfsStreamKey key, long size, T v) {
 		return put(key, 0, size, v);
 	}
 
 	<T> Ref<T> put(DfsStreamKey key, long pos, long size, T v) {
 		int slot = slot(key, pos);
 		HashEntry e1 = table.get(slot);
 		Ref<T> ref = scanRef(e1, key, pos);
 		if (ref != null) {
 			return ref;
 		}
 
 		reserveSpace(size, key);
 		ReentrantLock regionLock = lockFor(key, pos);
 		regionLock.lock();
 		try {
 			HashEntry e2 = table.get(slot);
 			if (e2 != e1) {
 				ref = scanRef(e2, key, pos);
 				if (ref != null) {
 					creditSpace(size, key);
 					return ref;
 				}
 			}
 
 			ref = new Ref<>(key, pos, size, v);
 			ref.markHotter();
 			for (;;) {
 				HashEntry n = new HashEntry(clean(e2), ref);
 				if (table.compareAndSet(slot, e2, n)) {
 					break;
 				}
 				e2 = table.get(slot);
 			}
 			addToClock(ref, 0);
 		} finally {
 			regionLock.unlock();
 		}
 		return ref;
 	}
 
 	boolean contains(DfsStreamKey key, long position) {
 		return scan(table.get(slot(key, position)), key, position) != null;
 	}
 
 	@SuppressWarnings("unchecked")
 	<T> T get(DfsStreamKey key, long position) {
 		T val = (T) scan(table.get(slot(key, position)), key, position);
 		if (val == null) {
 			getStat(statMiss, key).incrementAndGet();
 		} else {
 			getStat(statHit, key).incrementAndGet();
 		}
 		return val;
 	}
 
 	private <T> T scan(HashEntry n, DfsStreamKey key, long position) {
 		Ref<T> r = scanRef(n, key, position);
 		return r != null ? r.get() : null;
 	}
 
 	@SuppressWarnings("unchecked")
 	private <T> Ref<T> scanRef(HashEntry n, DfsStreamKey key, long position) {
 		for (; n != null; n = n.next) {
 			Ref<T> r = n.ref;
 			if (r.position == position && r.key.equals(key)) {
 				return r.get() != null ? r : null;
 			}
 		}
 		return null;
 	}
 
 	private int slot(DfsStreamKey key, long position) {
 		return (hash(key.hash, position) >>> 1) % tableSize;
 	}
 
 	private ReentrantLock lockFor(DfsStreamKey key, long position) {
 		return loadLocks[(hash(key.hash, position) >>> 1) % loadLocks.length];
 	}
 
 	private ReentrantLock lockForRef(DfsStreamKey key) {
 		int slot = (key.hash >>> 1) % refLocks[key.packExtPos].length;
 		return refLocks[key.packExtPos][slot];
 	}
 
 	private static AtomicLong[] newCounters() {
 		AtomicLong[] ret = new AtomicLong[PackExt.values().length];
 		for (int i = 0; i < ret.length; i++) {
 			ret[i] = new AtomicLong();
 		}
 		return ret;
 	}
 
 	private static AtomicLong getStat(AtomicReference<AtomicLong[]> stats,
 			DfsStreamKey key) {
 		int pos = key.packExtPos;
 		while (true) {
 			AtomicLong[] vals = stats.get();
 			if (pos < vals.length) {
 				return vals[pos];
 			}
 			AtomicLong[] expect = vals;
 			vals = new AtomicLong[Math.max(pos + 1, PackExt.values().length)];
 			System.arraycopy(expect, 0, vals, 0, expect.length);
 			for (int i = expect.length; i < vals.length; i++) {
 				vals[i] = new AtomicLong();
 			}
 			if (stats.compareAndSet(expect, vals)) {
 				return vals[pos];
 			}
 		}
 	}
 
 	private static long[] getStatVals(AtomicReference<AtomicLong[]> stat) {
 		AtomicLong[] stats = stat.get();
 		long[] cnt = new long[stats.length];
 		for (int i = 0; i < stats.length; i++) {
 			cnt[i] = stats[i].get();
 		}
 		return cnt;
 	}
 
 	private static HashEntry clean(HashEntry top) {
 		while (top != null && top.ref.next == null)
 			top = top.next;
 		if (top == null) {
 			return null;
 		}
 		HashEntry n = clean(top.next);
 		return n == top.next ? top : new HashEntry(n, top.ref);
 	}
 
 	private static final class HashEntry {
 		/** Next entry in the hash table's chain list. */
 		final HashEntry next;
 
 		/** The referenced object. */
 		final Ref ref;
 
 		HashEntry(HashEntry n, Ref r) {
 			next = n;
 			ref = r;
 		}
 	}
 
 	static final class Ref<T> {
 		final DfsStreamKey key;
 		final long position;
 		final long size;
 		volatile T value;
 		Ref next;
 
 		private volatile int hotCount;
 
 		Ref(DfsStreamKey key, long position, long size, T v) {
 			this.key = key;
 			this.position = position;
 			this.size = size;
 			this.value = v;
 		}
 
 		T get() {
 			T v = value;
 			if (v != null) {
 				markHotter();
 			}
 			return v;
 		}
 
 		boolean has() {
 			return value != null;
 		}
 
 		void markHotter() {
 			int cap = DfsBlockCache
 					.getInstance().cacheHotLimits[key.packExtPos];
 			hotCount = Math.min(cap, hotCount + 1);
 		}
 
 		void markColder() {
 			hotCount = Math.max(0, hotCount - 1);
 		}
 
 		boolean isHot() {
 			return hotCount > 0;
 		}
 	}
 
 	@FunctionalInterface
 	interface RefLoader<T> {
 		Ref<T> load() throws IOException;
 	}
 
 	/**
 	 * Supplier for readable channel
 	 */
 	@FunctionalInterface
 	interface ReadableChannelSupplier {
 		/**
 		 * @return ReadableChannel
 		 * @throws IOException
 		 */
 		ReadableChannel get() throws IOException;
 	}
 }