// 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;
}