/*-------------------------------------------------------------------------
 *
 * fsmpage.c
 *   routines to search and manipulate one FSM page.
 *
 *
 * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 *   src/backend/storage/freespace/fsmpage.c
 *
 * NOTES:
 *
 * The public functions in this file form an API that hides the internal
 * structure of a FSM page. This allows freespace.c to treat each FSM page
 * as a black box with SlotsPerPage "slots". fsm_set_avail() and
 * fsm_get_avail() let you get/set the value of a slot, and
 * fsm_search_avail() lets you search for a slot with value >= X.
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "storage/bufmgr.h"
#include "storage/fsm_internals.h"

/* Macros to navigate the tree within a page. Root has index zero. */
#define leftchild(x) (2 * (x) + 1)
#define rightchild(x) (2 * (x) + 2)
#define parentof(x)  (((x) - 1) / 2)

/*
 * Find right neighbor of x, wrapping around within the level
 */
static int
rightneighbor(int x)
{
 /*
  * Move right. This might wrap around, stepping to the leftmost node at
  * the next level.
 */
 x++;

 /*
  * Check if we stepped to the leftmost node at next level, and correct if
  * so. The leftmost nodes at each level are numbered x = 2^level - 1, so
  * check if (x + 1) is a power of two, using a standard
  * twos-complement-arithmetic trick.
 */
 if (((x + 1) & x) == 0)
  x = parentof(x);

 return x;
}

/*
 * Sets the value of a slot on page. Returns true if the page was modified.
 *
 * The caller must hold an exclusive lock on the page.
 */
bool
fsm_set_avail(Page page, int slot, uint8 value)
{
 int   nodeno = NonLeafNodesPerPage + slot;
 FSMPage  fsmpage = (FSMPage) PageGetContents(page);
 uint8  oldvalue;

 Assert(slot < LeafNodesPerPage);

 oldvalue = fsmpage->fp_nodes[nodeno];

 /* If the value hasn't changed, we don't need to do anything */
 if (oldvalue == value && value <= fsmpage->fp_nodes[0])
  return false;

 fsmpage->fp_nodes[nodeno] = value;

 /*
  * Propagate up, until we hit the root or a node that doesn't need to be
  * updated.
 */
 do
 {
  uint8  newvalue = 0;
  int   lchild;
  int   rchild;

  nodeno = parentof(nodeno);
  lchild = leftchild(nodeno);
  rchild = lchild + 1;

  newvalue = fsmpage->fp_nodes[lchild];
  if (rchild < NodesPerPage)
   newvalue = Max(newvalue,
         fsmpage->fp_nodes[rchild]);

  oldvalue = fsmpage->fp_nodes[nodeno];
  if (oldvalue == newvalue)
   break;

  fsmpage->fp_nodes[nodeno] = newvalue;
 } while (nodeno > 0);

 /*
  * sanity check: if the new value is (still) higher than the value at the
  * top, the tree is corrupt.  If so, rebuild.
 */
 if (value > fsmpage->fp_nodes[0])
  fsm_rebuild_page(page);

 return true;
}

/*
 * Returns the value of given slot on page.
 *
 * Since this is just a read-only access of a single byte, the page doesn't
 * need to be locked.
 */
uint8
fsm_get_avail(Page page, int slot)
{
 FSMPage  fsmpage = (FSMPage) PageGetContents(page);

 Assert(slot < LeafNodesPerPage);

 return fsmpage->fp_nodes[NonLeafNodesPerPage + slot];
}

/*
 * Returns the value at the root of a page.
 *
 * Since this is just a read-only access of a single byte, the page doesn't
 * need to be locked.
 */
uint8
fsm_get_max_avail(Page page)
{
 FSMPage  fsmpage = (FSMPage) PageGetContents(page);

 return fsmpage->fp_nodes[0];
}

/*
 * Searches for a slot with category at least minvalue.
 * Returns slot number, or -1 if none found.
 *
 * The caller must hold at least a shared lock on the page, and this
 * function can unlock and lock the page again in exclusive mode if it
 * needs to be updated. exclusive_lock_held should be set to true if the
 * caller is already holding an exclusive lock, to avoid extra work.
 *
 * If advancenext is false, fp_next_slot is set to point to the returned
 * slot, and if it's true, to the slot after the returned slot.
 */
int
fsm_search_avail(Buffer buf, uint8 minvalue, bool advancenext,
     bool exclusive_lock_held)
{
 Page  page = BufferGetPage(buf);
 FSMPage  fsmpage = (FSMPage) PageGetContents(page);
 int   nodeno;
 int   target;
 uint16  slot;

restart:

 /*
  * Check the root first, and exit quickly if there's no leaf with enough
  * free space
 */
 if (fsmpage->fp_nodes[0] < minvalue)
  return -1;

 /*
  * Start search using fp_next_slot.  It's just a hint, so check that it's
  * sane.  (This also handles wrapping around when the prior call returned
  * the last slot on the page.)
 */
 target = fsmpage->fp_next_slot;
 if (target < 0 || target >= LeafNodesPerPage)
  target = 0;
 target += NonLeafNodesPerPage;

 /*----------
  * Start the search from the target slot.  At every step, move one
  * node to the right, then climb up to the parent.  Stop when we reach
  * a node with enough free space (as we must, since the root has enough
  * space).
  *
  * The idea is to gradually expand our "search triangle", that is, all
  * nodes covered by the current node, and to be sure we search to the
  * right from the start point.  At the first step, only the target slot
  * is examined.  When we move up from a left child to its parent, we are
  * adding the right-hand subtree of that parent to the search triangle.
  * When we move right then up from a right child, we are dropping the
  * current search triangle (which we know doesn't contain any suitable
  * page) and instead looking at the next-larger-size triangle to its
  * right.  So we never look left from our original start point, and at
  * each step the size of the search triangle doubles, ensuring it takes
  * only log2(N) work to search N pages.
  *
  * The "move right" operation will wrap around if it hits the right edge
  * of the tree, so the behavior is still good if we start near the right.
  * Note also that the move-and-climb behavior ensures that we can't end
  * up on one of the missing nodes at the right of the leaf level.
  *
  * For example, consider this tree:
  *
  *     7
  *    7    6
  *  5  7  6  5
  * 4 5 5 7 2 6 5 2
  *    T
  *
  * Assume that the target node is the node indicated by the letter T,
  * and we're searching for a node with value of 6 or higher. The search
  * begins at T. At the first iteration, we move to the right, then to the
  * parent, arriving at the rightmost 5. At the second iteration, we move
  * to the right, wrapping around, then climb up, arriving at the 7 on the
  * third level.  7 satisfies our search, so we descend down to the bottom,
  * following the path of sevens.  This is in fact the first suitable page
  * to the right of (allowing for wraparound) our start point.
  *----------
 */
 nodeno = target;
 while (nodeno > 0)
 {
  if (fsmpage->fp_nodes[nodeno] >= minvalue)
   break;

  /*
   * Move to the right, wrapping around on same level if necessary, then
   * climb up.
 */
  nodeno = parentof(rightneighbor(nodeno));
 }

 /*
  * We're now at a node with enough free space, somewhere in the middle of
  * the tree. Descend to the bottom, following a path with enough free
  * space, preferring to move left if there's a choice.
 */
 while (nodeno < NonLeafNodesPerPage)
 {
  int   childnodeno = leftchild(nodeno);

  if (childnodeno < NodesPerPage &&
   fsmpage->fp_nodes[childnodeno] >= minvalue)
  {
   nodeno = childnodeno;
   continue;
  }
  childnodeno++;   /* point to right child */
  if (childnodeno < NodesPerPage &&
   fsmpage->fp_nodes[childnodeno] >= minvalue)
  {
   nodeno = childnodeno;
  }
  else
  {
   /*
    * Oops. The parent node promised that either left or right child
    * has enough space, but neither actually did. This can happen in
    * case of a "torn page", IOW if we crashed earlier while writing
    * the page to disk, and only part of the page made it to disk.
    *
    * Fix the corruption and restart.
 */
   RelFileLocator rlocator;
   ForkNumber forknum;
   BlockNumber blknum;

   BufferGetTag(buf, &rlocator, &forknum, &blknum);
   elog(DEBUG1, "fixing corrupt FSM block %u, relation %u/%u/%u",
     blknum, rlocator.spcOid, rlocator.dbOid, rlocator.relNumber);

   /* make sure we hold an exclusive lock */
   if (!exclusive_lock_held)
   {
    LockBuffer(buf, BUFFER_LOCK_UNLOCK);
    LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
    exclusive_lock_held = true;
   }
   fsm_rebuild_page(page);
   MarkBufferDirtyHint(buf, false);
   goto restart;
  }
 }

 /* We're now at the bottom level, at a node with enough space. */
 slot = nodeno - NonLeafNodesPerPage;

 /*
  * Update the next-target pointer. Note that we do this even if we're only
  * holding a shared lock, on the grounds that it's better to use a shared
  * lock and get a garbled next pointer every now and then, than take the
  * concurrency hit of an exclusive lock.
  *
  * Wrap-around is handled at the beginning of this function.
 */
 fsmpage->fp_next_slot = slot + (advancenext ? 1 : 0);

 return slot;
}

/*
 * Sets the available space to zero for all slots numbered >= nslots.
 * Returns true if the page was modified.
 */
bool
fsm_truncate_avail(Page page, int nslots)
{
 FSMPage  fsmpage = (FSMPage) PageGetContents(page);
 uint8    *ptr;
 bool  changed = false;

 Assert(nslots >= 0 && nslots < LeafNodesPerPage);

 /* Clear all truncated leaf nodes */
 ptr = &fsmpage->fp_nodes[NonLeafNodesPerPage + nslots];
 for (; ptr < &fsmpage->fp_nodes[NodesPerPage]; ptr++)
 {
  if (*ptr != 0)
   changed = true;
  *ptr = 0;
 }

 /* Fix upper nodes. */
 if (changed)
  fsm_rebuild_page(page);

 return changed;
}

/*
 * Reconstructs the upper levels of a page. Returns true if the page
 * was modified.
 */
bool
fsm_rebuild_page(Page page)
{
 FSMPage  fsmpage = (FSMPage) PageGetContents(page);
 bool  changed = false;
 int   nodeno;

 /*
  * Start from the lowest non-leaf level, at last node, working our way
  * backwards, through all non-leaf nodes at all levels, up to the root.
 */
 for (nodeno = NonLeafNodesPerPage - 1; nodeno >= 0; nodeno--)
 {
  int   lchild = leftchild(nodeno);
  int   rchild = lchild + 1;
  uint8  newvalue = 0;

  /* The first few nodes we examine might have zero or one child. */
  if (lchild < NodesPerPage)
   newvalue = fsmpage->fp_nodes[lchild];

  if (rchild < NodesPerPage)
   newvalue = Max(newvalue,
         fsmpage->fp_nodes[rchild]);

  if (fsmpage->fp_nodes[nodeno] != newvalue)
  {
   fsmpage->fp_nodes[nodeno] = newvalue;
   changed = true;
  }
 }

 return changed;
}