// SPDX-License-Identifier: GPL-2.0-only
/*
 * Copyright 2023 Red Hat
 */

#include "sparse-cache.h"

#include <linux/cache.h>
#include <linux/delay.h>
#include <linux/dm-bufio.h>

#include "logger.h"
#include "memory-alloc.h"
#include "permassert.h"

#include "chapter-index.h"
#include "config.h"
#include "index.h"

/*
 * Since the cache is small, it is implemented as a simple array of cache entries. Searching for a
 * specific virtual chapter is implemented as a linear search. The cache replacement policy is
 * least-recently-used (LRU). Again, the small size of the cache allows the LRU order to be
 * maintained by shifting entries in an array list.
 *
 * Changing the contents of the cache requires the coordinated participation of all zone threads
 * via the careful use of barrier messages sent to all the index zones by the triage queue worker
 * thread. The critical invariant for coordination is that the cache membership must not change
 * between updates, so that all calls to uds_sparse_cache_contains() from the zone threads must all
 * receive the same results for every virtual chapter number. To ensure that critical invariant,
 * state changes such as "that virtual chapter is no longer in the volume" and "skip searching that
 * chapter because it has had too many cache misses" are represented separately from the cache
 * membership information (the virtual chapter number).
 *
 * As a result of this invariant, we have the guarantee that every zone thread will call
 * uds_update_sparse_cache() once and exactly once to request a chapter that is not in the cache,
 * and the serialization of the barrier requests from the triage queue ensures they will all
 * request the same chapter number. This means the only synchronization we need can be provided by
 * a pair of thread barriers used only in the uds_update_sparse_cache() call, providing a critical
 * section where a single zone thread can drive the cache update while all the other zone threads
 * are known to be blocked, waiting in the second barrier. Outside that critical section, all the
 * zone threads implicitly hold a shared lock. Inside it, the thread for zone zero holds an
 * exclusive lock. No other threads may access or modify the cache entries.
 *
 * Chapter statistics must only be modified by a single thread, which is also the zone zero thread.
 * All fields that might be frequently updated by that thread are kept in separate cache-aligned
 * structures so they will not cause cache contention via "false sharing" with the fields that are
 * frequently accessed by all of the zone threads.
 *
 * The LRU order is managed independently by each zone thread, and each zone uses its own list for
 * searching and cache membership queries. The zone zero list is used to decide which chapter to
 * evict when the cache is updated, and its search list is copied to the other threads at that
 * time.
 *
 * The virtual chapter number field of the cache entry is the single field indicating whether a
 * chapter is a member of the cache or not. The value NO_CHAPTER is used to represent a null or
 * undefined chapter number. When present in the virtual chapter number field of a
 * cached_chapter_index, it indicates that the cache entry is dead, and all the other fields of
 * that entry (other than immutable pointers to cache memory) are undefined and irrelevant. Any
 * cache entry that is not marked as dead is fully defined and a member of the cache, and
 * uds_sparse_cache_contains() will always return true for any virtual chapter number that appears
 * in any of the cache entries.
 *
 * A chapter index that is a member of the cache may be excluded from searches between calls to
 * uds_update_sparse_cache() in two different ways. First, when a chapter falls off the end of the
 * volume, its virtual chapter number will be less that the oldest virtual chapter number. Since
 * that chapter is no longer part of the volume, there's no point in continuing to search that
 * chapter index. Once invalidated, that virtual chapter will still be considered a member of the
 * cache, but it will no longer be searched for matching names.
 *
 * The second mechanism is a heuristic based on keeping track of the number of consecutive search
 * misses in a given chapter index. Once that count exceeds a threshold, the skip_search flag will
 * be set to true, causing the chapter to be skipped when searching the entire cache, but still
 * allowing it to be found when searching for a hook in that specific chapter. Finding a hook will
 * clear the skip_search flag, once again allowing the non-hook searches to use that cache entry.
 * Again, regardless of the state of the skip_search flag, the virtual chapter must still
 * considered to be a member of the cache for uds_sparse_cache_contains().
 */

#define SKIP_SEARCH_THRESHOLD 20000
#define ZONE_ZERO 0

/*
 * These counters are essentially fields of the struct cached_chapter_index, but are segregated
 * into this structure because they are frequently modified. They are grouped and aligned to keep
 * them on different cache lines from the chapter fields that are accessed far more often than they
 * are updated.
 */
struct __aligned(L1_CACHE_BYTES) cached_index_counters {
	u64 consecutive_misses;
};

struct __aligned(L1_CACHE_BYTES) cached_chapter_index {
	/*
	 * The virtual chapter number of the cached chapter index. NO_CHAPTER means this cache
	 * entry is unused. This field must only be modified in the critical section in
	 * uds_update_sparse_cache().
	 */
	u64 virtual_chapter;

	u32 index_pages_count;

	/*
	 * These pointers are immutable during the life of the cache. The contents of the arrays
	 * change when the cache entry is replaced.
	 */
	struct delta_index_page *index_pages;
	struct dm_buffer **page_buffers;

	/*
	 * If set, skip the chapter when searching the entire cache. This flag is just a
	 * performance optimization. This flag is mutable between cache updates, but it rarely
	 * changes and is frequently accessed, so it groups with the immutable fields.
	 */
	bool skip_search;

	/*
	 * The cache-aligned counters change often and are placed at the end of the structure to
	 * prevent false sharing with the more stable fields above.
	 */
	struct cached_index_counters counters;
};

/*
 * A search_list represents an ordering of the sparse chapter index cache entry array, from most
 * recently accessed to least recently accessed, which is the order in which the indexes should be
 * searched and the reverse order in which they should be evicted from the cache.
 *
 * Cache entries that are dead or empty are kept at the end of the list, avoiding the need to even
 * iterate over them to search, and ensuring that dead entries are replaced before any live entries
 * are evicted.
 *
 * The search list is instantiated for each zone thread, avoiding any need for synchronization. The
 * structure is allocated on a cache boundary to avoid false sharing of memory cache lines between
 * zone threads.
 */
struct search_list {
	u8 capacity;
	u8 first_dead_entry;
	struct cached_chapter_index *entries[];
};

struct threads_barrier {
	/* Lock for this barrier object */
	struct semaphore lock;
	/* Semaphore for threads waiting at this barrier */
	struct semaphore wait;
	/* Number of threads which have arrived */
	int arrived;
	/* Total number of threads using this barrier */
	int thread_count;
};

struct sparse_cache {
	const struct index_geometry *geometry;
	unsigned int capacity;
	unsigned int zone_count;

	unsigned int skip_threshold;
	struct search_list *search_lists[MAX_ZONES];
	struct cached_chapter_index **scratch_entries;

	struct threads_barrier begin_update_barrier;
	struct threads_barrier end_update_barrier;

	struct cached_chapter_index chapters[];
};

static void initialize_threads_barrier(struct threads_barrier *barrier,
				       unsigned int thread_count)
{
	sema_init(&barrier->lock, 1);
	barrier->arrived = 0;
	barrier->thread_count = thread_count;
	sema_init(&barrier->wait, 0);
}

static inline void __down(struct semaphore *semaphore)
{
	/*
	 * Do not use down(semaphore). Instead use down_interruptible so that
	 * we do not get 120 second stall messages in kern.log.
	 */
	while (down_interruptible(semaphore) != 0) {
		/*
		 * If we're called from a user-mode process (e.g., "dmsetup
		 * remove") while waiting for an operation that may take a
		 * while (e.g., UDS index save), and a signal is sent (SIGINT,
		 * SIGUSR2), then down_interruptible will not block. If that
		 * happens, sleep briefly to avoid keeping the CPU locked up in
		 * this loop. We could just call cond_resched, but then we'd
		 * still keep consuming CPU time slices and swamp other threads
		 * trying to do computational work.
		 */
		fsleep(1000);
	}
}

static void enter_threads_barrier(struct threads_barrier *barrier)
{
	__down(&barrier->lock);
	if (++barrier->arrived == barrier->thread_count) {
		/* last thread */
		int i;

		for (i = 1; i < barrier->thread_count; i++)
			up(&barrier->wait);

		barrier->arrived = 0;
		up(&barrier->lock);
	} else {
		up(&barrier->lock);
		__down(&barrier->wait);
	}
}

static int __must_check initialize_cached_chapter_index(struct cached_chapter_index *chapter,
							const struct index_geometry *geometry)
{
	int result;

	chapter->virtual_chapter = NO_CHAPTER;
	chapter->index_pages_count = geometry->index_pages_per_chapter;

	result = vdo_allocate(chapter->index_pages_count, struct delta_index_page,
			      __func__, &chapter->index_pages);
	if (result != VDO_SUCCESS)
		return result;

	return vdo_allocate(chapter->index_pages_count, struct dm_buffer *,
			    "sparse index volume pages", &chapter->page_buffers);
}

static int __must_check make_search_list(struct sparse_cache *cache,
					 struct search_list **list_ptr)
{
	struct search_list *list;
	unsigned int bytes;
	u8 i;
	int result;

	bytes = (sizeof(struct search_list) +
		 (cache->capacity * sizeof(struct cached_chapter_index *)));
	result = vdo_allocate_cache_aligned(bytes, "search list", &list);
	if (result != VDO_SUCCESS)
		return result;

	list->capacity = cache->capacity;
	list->first_dead_entry = 0;

	for (i = 0; i < list->capacity; i++)
		list->entries[i] = &cache->chapters[i];

	*list_ptr = list;
	return UDS_SUCCESS;
}

int uds_make_sparse_cache(const struct index_geometry *geometry, unsigned int capacity,
			  unsigned int zone_count, struct sparse_cache **cache_ptr)
{
	int result;
	unsigned int i;
	struct sparse_cache *cache;
	unsigned int bytes;

	bytes = (sizeof(struct sparse_cache) + (capacity * sizeof(struct cached_chapter_index)));
	result = vdo_allocate_cache_aligned(bytes, "sparse cache", &cache);
	if (result != VDO_SUCCESS)
		return result;

	cache->geometry = geometry;
	cache->capacity = capacity;
	cache->zone_count = zone_count;

	/*
	 * Scale down the skip threshold since the cache only counts cache misses in zone zero, but
	 * requests are being handled in all zones.
	 */
	cache->skip_threshold = (SKIP_SEARCH_THRESHOLD / zone_count);

	initialize_threads_barrier(&cache->begin_update_barrier, zone_count);
	initialize_threads_barrier(&cache->end_update_barrier, zone_count);

	for (i = 0; i < capacity; i++) {
		result = initialize_cached_chapter_index(&cache->chapters[i], geometry);
		if (result != UDS_SUCCESS)
			goto out;
	}

	for (i = 0; i < zone_count; i++) {
		result = make_search_list(cache, &cache->search_lists[i]);
		if (result != UDS_SUCCESS)
			goto out;
	}

	/* purge_search_list() needs some temporary lists for sorting. */
	result = vdo_allocate(capacity * 2, struct cached_chapter_index *,
			      "scratch entries", &cache->scratch_entries);
	if (result != VDO_SUCCESS)
		goto out;

	*cache_ptr = cache;
	return UDS_SUCCESS;
out:
	uds_free_sparse_cache(cache);
	return result;
}

static inline void set_skip_search(struct cached_chapter_index *chapter,
				   bool skip_search)
{
	/* Check before setting to reduce cache line contention. */
	if (READ_ONCE(chapter->skip_search) != skip_search)
		WRITE_ONCE(chapter->skip_search, skip_search);
}

static void score_search_hit(struct cached_chapter_index *chapter)
{
	chapter->counters.consecutive_misses = 0;
	set_skip_search(chapter, false);
}

static void score_search_miss(struct sparse_cache *cache,
			      struct cached_chapter_index *chapter)
{
	chapter->counters.consecutive_misses++;
	if (chapter->counters.consecutive_misses > cache->skip_threshold)
		set_skip_search(chapter, true);
}

static void release_cached_chapter_index(struct cached_chapter_index *chapter)
{
	unsigned int i;

	chapter->virtual_chapter = NO_CHAPTER;
	if (chapter->page_buffers == NULL)
		return;

	for (i = 0; i < chapter->index_pages_count; i++) {
		if (chapter->page_buffers[i] != NULL)
			dm_bufio_release(vdo_forget(chapter->page_buffers[i]));
	}
}

void uds_free_sparse_cache(struct sparse_cache *cache)
{
	unsigned int i;

	if (cache == NULL)
		return;

	vdo_free(cache->scratch_entries);

	for (i = 0; i < cache->zone_count; i++)
		vdo_free(cache->search_lists[i]);

	for (i = 0; i < cache->capacity; i++) {
		release_cached_chapter_index(&cache->chapters[i]);
		vdo_free(cache->chapters[i].index_pages);
		vdo_free(cache->chapters[i].page_buffers);
	}

	vdo_free(cache);
}

/*
 * Take the indicated element of the search list and move it to the start, pushing the pointers
 * previously before it back down the list.
 */
static inline void set_newest_entry(struct search_list *search_list, u8 index)
{
	struct cached_chapter_index *newest;

	if (index > 0) {
		newest = search_list->entries[index];
		memmove(&search_list->entries[1], &search_list->entries[0],
			index * sizeof(struct cached_chapter_index *));
		search_list->entries[0] = newest;
	}

	/*
	 * This function may have moved a dead chapter to the front of the list for reuse, in which
	 * case the set of dead chapters becomes smaller.
	 */
	if (search_list->first_dead_entry <= index)
		search_list->first_dead_entry++;
}

bool uds_sparse_cache_contains(struct sparse_cache *cache, u64 virtual_chapter,
			       unsigned int zone_number)
{
	struct search_list *search_list;
	struct cached_chapter_index *chapter;
	u8 i;

	/*
	 * The correctness of the barriers depends on the invariant that between calls to
	 * uds_update_sparse_cache(), the answers this function returns must never vary: the result
	 * for a given chapter must be identical across zones. That invariant must be maintained
	 * even if the chapter falls off the end of the volume, or if searching it is disabled
	 * because of too many search misses.
	 */
	search_list = cache->search_lists[zone_number];
	for (i = 0; i < search_list->first_dead_entry; i++) {
		chapter = search_list->entries[i];

		if (virtual_chapter == chapter->virtual_chapter) {
			if (zone_number == ZONE_ZERO)
				score_search_hit(chapter);

			set_newest_entry(search_list, i);
			return true;
		}
	}

	return false;
}

/*
 * Re-sort cache entries into three sets (active, skippable, and dead) while maintaining the LRU
 * ordering that already existed. This operation must only be called during the critical section in
 * uds_update_sparse_cache().
 */
static void purge_search_list(struct search_list *search_list,
			      struct sparse_cache *cache, u64 oldest_virtual_chapter)
{
	struct cached_chapter_index **entries;
	struct cached_chapter_index **skipped;
	struct cached_chapter_index **dead;
	struct cached_chapter_index *chapter;
	unsigned int next_alive = 0;
	unsigned int next_skipped = 0;
	unsigned int next_dead = 0;
	unsigned int i;

	entries = &search_list->entries[0];
	skipped = &cache->scratch_entries[0];
	dead = &cache->scratch_entries[search_list->capacity];

	for (i = 0; i < search_list->first_dead_entry; i++) {
		chapter = search_list->entries[i];
		if ((chapter->virtual_chapter < oldest_virtual_chapter) ||
		    (chapter->virtual_chapter == NO_CHAPTER))
			dead[next_dead++] = chapter;
		else if (chapter->skip_search)
			skipped[next_skipped++] = chapter;
		else
			entries[next_alive++] = chapter;
	}

	memcpy(&entries[next_alive], skipped,
	       next_skipped * sizeof(struct cached_chapter_index *));
	memcpy(&entries[next_alive + next_skipped], dead,
	       next_dead * sizeof(struct cached_chapter_index *));
	search_list->first_dead_entry = next_alive + next_skipped;
}

static int __must_check cache_chapter_index(struct cached_chapter_index *chapter,
					    u64 virtual_chapter,
					    const struct volume *volume)
{
	int result;

	release_cached_chapter_index(chapter);

	result = uds_read_chapter_index_from_volume(volume, virtual_chapter,
						    chapter->page_buffers,
						    chapter->index_pages);
	if (result != UDS_SUCCESS)
		return result;

	chapter->counters.consecutive_misses = 0;
	chapter->virtual_chapter = virtual_chapter;
	chapter->skip_search = false;

	return UDS_SUCCESS;
}

static inline void copy_search_list(const struct search_list *source,
				    struct search_list *target)
{
	*target = *source;
	memcpy(target->entries, source->entries,
	       source->capacity * sizeof(struct cached_chapter_index *));
}

/*
 * Update the sparse cache to contain a chapter index. This function must be called by all the zone
 * threads with the same chapter number to correctly enter the thread barriers used to synchronize
 * the cache updates.
 */
int uds_update_sparse_cache(struct index_zone *zone, u64 virtual_chapter)
{
	int result = UDS_SUCCESS;
	const struct uds_index *index = zone->index;
	struct sparse_cache *cache = index->volume->sparse_cache;

	if (uds_sparse_cache_contains(cache, virtual_chapter, zone->id))
		return UDS_SUCCESS;

	/*
	 * Wait for every zone thread to reach its corresponding barrier request and invoke this
	 * function before starting to modify the cache.
	 */
	enter_threads_barrier(&cache->begin_update_barrier);

	/*
	 * This is the start of the critical section: the zone zero thread is captain, effectively
	 * holding an exclusive lock on the sparse cache. All the other zone threads must do
	 * nothing between the two barriers. They will wait at the end_update_barrier again for the
	 * captain to finish the update.
	 */

	if (zone->id == ZONE_ZERO) {
		unsigned int z;
		struct search_list *list = cache->search_lists[ZONE_ZERO];

		purge_search_list(list, cache, zone->oldest_virtual_chapter);

		if (virtual_chapter >= index->oldest_virtual_chapter) {
			set_newest_entry(list, list->capacity - 1);
			result = cache_chapter_index(list->entries[0], virtual_chapter,
						     index->volume);
		}

		for (z = 1; z < cache->zone_count; z++)
			copy_search_list(list, cache->search_lists[z]);
	}

	/*
	 * This is the end of the critical section. All cache invariants must have been restored.
	 */
	enter_threads_barrier(&cache->end_update_barrier);
	return result;
}

void uds_invalidate_sparse_cache(struct sparse_cache *cache)
{
	unsigned int i;

	for (i = 0; i < cache->capacity; i++)
		release_cached_chapter_index(&cache->chapters[i]);
}

static inline bool should_skip_chapter(struct cached_chapter_index *chapter,
				       u64 oldest_chapter, u64 requested_chapter)
{
	if ((chapter->virtual_chapter == NO_CHAPTER) ||
	    (chapter->virtual_chapter < oldest_chapter))
		return true;

	if (requested_chapter != NO_CHAPTER)
		return requested_chapter != chapter->virtual_chapter;
	else
		return READ_ONCE(chapter->skip_search);
}

static int __must_check search_cached_chapter_index(struct cached_chapter_index *chapter,
						    const struct index_geometry *geometry,
						    const struct index_page_map *index_page_map,
						    const struct uds_record_name *name,
						    u16 *record_page_ptr)
{
	u32 physical_chapter =
		uds_map_to_physical_chapter(geometry, chapter->virtual_chapter);
	u32 index_page_number =
		uds_find_index_page_number(index_page_map, name, physical_chapter);
	struct delta_index_page *index_page =
		&chapter->index_pages[index_page_number];

	return uds_search_chapter_index_page(index_page, geometry, name,
					     record_page_ptr);
}

int uds_search_sparse_cache(struct index_zone *zone, const struct uds_record_name *name,
			    u64 *virtual_chapter_ptr, u16 *record_page_ptr)
{
	int result;
	struct volume *volume = zone->index->volume;
	struct sparse_cache *cache = volume->sparse_cache;
	struct cached_chapter_index *chapter;
	struct search_list *search_list;
	u8 i;
	/* Search the entire cache unless a specific chapter was requested. */
	bool search_one = (*virtual_chapter_ptr != NO_CHAPTER);

	*record_page_ptr = NO_CHAPTER_INDEX_ENTRY;
	search_list = cache->search_lists[zone->id];
	for (i = 0; i < search_list->first_dead_entry; i++) {
		chapter = search_list->entries[i];

		if (should_skip_chapter(chapter, zone->oldest_virtual_chapter,
					*virtual_chapter_ptr))
			continue;

		result = search_cached_chapter_index(chapter, cache->geometry,
						     volume->index_page_map, name,
						     record_page_ptr);
		if (result != UDS_SUCCESS)
			return result;

		if (*record_page_ptr != NO_CHAPTER_INDEX_ENTRY) {
			/*
			 * In theory, this might be a false match while a true match exists in
			 * another chapter, but that's a very rare case and not worth the extra
			 * search complexity.
			 */
			set_newest_entry(search_list, i);
			if (zone->id == ZONE_ZERO)
				score_search_hit(chapter);

			*virtual_chapter_ptr = chapter->virtual_chapter;
			return UDS_SUCCESS;
		}

		if (zone->id == ZONE_ZERO)
			score_search_miss(cache, chapter);

		if (search_one)
			break;
	}

	return UDS_SUCCESS;
}