探索LRU算法

有太多能力可以提升，不要只看到技术

开始之前，先上一道经典的面试题，LRU【🐕】

热身

设计和构建一个“最近最少使用”缓存，该缓存会删除最近最少使用的项目。缓存应该从键映射到值(允许你插入和检索特定键对应的值)，并在初始化时指定最大容量。当缓存被填满时，它应该删除最近最少使用的项目。

它应该支持以下操作：获取数据 get 和写入数据 put 。

获取数据 get(key) - 如果密钥 (key) 存在于缓存中，则获取密钥的值（总是正数），否则返回 -1。
写入数据 put(key, value) - 如果密钥不存在，则写入其数据值。当缓存容量达到上限时，它应该在写入新数据之前删除最近最少使用的数据值，从而为新的数据值留出空间。

示例:

LRUCache cache = new LRUCache( 2 /* 缓存容量 */ );

cache.put(1, 1);
cache.put(2, 2);
cache.get(1);       // 返回  1
cache.put(3, 3);    // 该操作会使得密钥 2 作废
cache.get(2);       // 返回 -1 (未找到)
cache.put(4, 4);    // 该操作会使得密钥 1 作废
cache.get(1);       // 返回 -1 (未找到)
cache.get(3);       // 返回  3
cache.get(4);       // 返回  4

class Node{
public:
    int key,value;
    Node *pre,*next;
    Node(int k=0,int v=0):key(k),value(v){}
};

class LRUCache {
private:
    int len;
    Node* head;
    map<int,Node*> mp;
    void remove(Node *node){
        node->next->pre=node->pre;
        node->pre->next=node->next;
    }
    void push_front(Node* node){
        node->pre=head;
        node->next=head->next;
        node->pre->next=node;
        node->next->pre=node;
    }
    Node* getNode(int key){
        auto it=mp.find(key);
        if(it==mp.end()){
            return nullptr;
        }
        auto node=mp[key];
        remove(node);
        push_front(node);
        return node;
    }
public:
    LRUCache(int capacity) {
        len=capacity;
        mp.clear();
        head=new Node();
        head->next=head;
        head->pre=head;
    }
    int get(int key) {
        auto node=getNode(key);
        if(node==nullptr){
            return -1;
        }
        return node->value;
    }
    
    void put(int key, int value) {
        auto node=getNode(key);
        if(node){
            node->value=value;
            return ;
        }
        Node* now=new Node(key,value);
        mp[key]=now;
        push_front(now);
        if(mp.size()>len){
            auto last=head->pre;
            mp.erase(last->key);
            remove(last);
            delete last;
        }
    }
};

/**
 * Your LRUCache object will be instantiated and called as such:
 * LRUCache* obj = new LRUCache(capacity);
 * int param_1 = obj->get(key);
 * obj->put(key,value);
 */

存在问题

学习自：探索 LRU 算法的缺陷与解决方案

学习完了LRU算法，接下来看看是否存在什么问题呢？

传统的 LRU 算法无法避免下面这两个问题：

预读失效导致缓存命中率下降；
缓存污染导致缓存命中率下降；

预读失效

在 Linux 操作系统中，为了优化基于 Page Cache 的读缓存机制，提供了预读（read-ahead）机制。这个机制的工作原理可以通过以下例子来说明：

假设应用程序需要读取磁盘上文件 A 中 offset 为 0-3KB 的数据。由于磁盘的基本读写单位是块（通常为 4KB），操作系统会至少读取 0-4KB 范围内的内容，将其加载到内存中的一个 page 中。

然而，操作系统会基于空间局部性原则（即与当前访问的数据相邻的数据在未来有很大可能会被访问到），进一步优化读操作。为了提高后续读操作的效率，操作系统会预先将文件 A 的其他部分也加载到内存中，例如 offset [4KB,8KB)、[8KB,12KB)、以及 [12KB,16KB) 范围内的数据。这意味着操作系统会在内存中额外分配 3 个 page 来存储这些数据块，以便在应用程序需要访问这些范围的数据时，可以直接从内存中获取，而无需再次访问磁盘。

通过这种预读机制，操作系统能显著提高数据访问的效率，减少磁盘 I/O 的次数，提升整体性能。

问题：

预读失效会带来什么问题？

如果这些被提前加载进来的页，并没有被访问，相当于这个预读工作是白做了，这个就是预读失效。

如果使用传统的 LRU 算法，就会把「预读页」放到 LRU 链表头部，而当内存空间不够的时候，还需要把末尾的页淘汰掉。不会被访问的预读页却占用了 LRU 链表前排的位置，而末尾淘汰的页，可能是热点数据，这样就大大降低了缓存命中率。

Linux方案

Linux 操作系统实现两个了 LRU 链表：活跃 LRU 链表和非活跃 LRU 链表；

有了这两个 LRU 链表后，预读页就只需要加入到 inactive list 区域的头部，当页被真正访问的时候，才将页插入 active list 的头部。如果预读的页一直没有被访问，就会从 inactive list 移除，这样不会影响 active list 中的热点数据。active list 淘汰的数据就会被降级到 inactive list ，作为 inactive list 头部。

缓冲污染

当我们在批量读取数据的时候，由于数据被访问了一次，这些大量数据都会被加入到「活跃 LRU 链表」里，然后之前缓存在活跃 LRU 链表（或者 young 区域）里的热点数据全部都被淘汰了，如果这些大量的数据在很长一段时间都不会被访问的话，那么整个活跃 LRU 链表（或者 young 区域）就被污染了。

以 MySQL 举例子

某一个 SQL 语句扫描了大量的数据时，在 Buffer Pool 空间比较有限的情况下，可能会将 Buffer Pool 里的所有页都替换出去，导致大量热数据被淘汰了，等这些热数据又被再次访问的时候，由于缓存未命中，就会产生大量的磁盘 I/O，MySQL 性能就会急剧下降。

LRU-K

针对上述问题，之前在数据库实践专门针对这种情况使用另外的LRU-K算法。

LRU-K 算法的核心是根据每个页面的最近 K 次访问记录来决定哪个页面应该被淘汰。访问次数越多，页面在缓存中存活的时间就越长。简单来说，LRU-K 提供了一种更精细的页面替换策略，与普通的 LRU 不同，它考虑了更深层次的访问历史。

#include "buffer/lru_k_replacer.h"

namespace bustub {

// 构造函数，初始化替换器大小和K值
LRUKReplacer::LRUKReplacer(size_t num_frames, size_t k) : replacer_size_(num_frames), k_(k) {}

// 驱逐一个frame，选择最久未被访问的
auto LRUKReplacer::Evict(frame_id_t *frame_id) -> bool {
  std::lock_guard<std::mutex> lock_guard(latch_);

  // 1. 先从历史队列(hist_)中选择可驱逐的frame
  if (!hist_.empty()) {
    for (auto it : hist_) {
      if (mp_[it]) { // 判断是否可以驱逐
        frame_id_t num = it;
        frame_num_[it] = 0; // 重置frame的访问计数
        mp_[it] = false; // 标记为不可驱逐
        hist_.remove(num); // 从历史队列中移除
        *frame_id = num;
        return true;
      }
    }
  }

  // 2. 如果历史队列中没有可驱逐的frame，则从当前缓存队列(cur_)中选择
  if (!cur_.empty()) {
    for (auto it : cur_) {
      if (mp_[it]) { // 判断是否可以驱逐
        frame_id_t num = it;
        frame_num_[it] = 0; // 重置frame的访问计数
        mp_[it] = false; // 标记为不可驱逐
        cur_.remove(num); // 从当前缓存队列中移除
        *frame_id = num;
        return true;
      }
    }
  }
  
  return false; // 如果没有可驱逐的frame，则返回false
}

// 记录一个frame的访问，更新其在LRU-K队列中的位置
void LRUKReplacer::RecordAccess(frame_id_t frame_id) {
  std::lock_guard<std::mutex> lock_guard(latch_);
  
  // 忽略无效的frame_id
  if (frame_id > static_cast<int>(replacer_size_)) {
    return;
  }

  // 如果frame是第一次被访问或者计数为0
  if (frame_num_.find(frame_id) == frame_num_.end() || frame_num_[frame_id] == 0) {
    frame_num_[frame_id]++; // 增加访问计数
    if (frame_num_[frame_id] >= k_) {
      cur_.push_back(frame_id); // 如果访问次数达到k，加入当前缓存队列
    } else {
      hist_.push_back(frame_id); // 否则加入历史队列
    }
  } else { // 如果frame已经被访问过
    frame_num_[frame_id]++; // 增加访问计数
    if (frame_num_[frame_id] < k_) {
      // 如果还未达到k次访问，保持在历史队列中
    } else if (frame_num_[frame_id] == k_) {
      hist_.remove(frame_id); // 从历史队列中移除
      cur_.push_back(frame_id); // 加入当前缓存队列
    } else {
      cur_.remove(frame_id); // 从当前缓存队列中移除
      cur_.push_back(frame_id); // 再次加入当前缓存队列，保持LRU顺序
    }
  }
}

// 设置一个frame是否可以被驱逐
void LRUKReplacer::SetEvictable(frame_id_t frame_id, bool set_evictable) {
  std::lock_guard<std::mutex> lock_guard(latch_);
  
  // 忽略无效的frame_id
  if (frame_id > static_cast<int>(replacer_size_)) {
    return;
  }
  
  mp_[frame_id] = set_evictable; // 设置可驱逐标记
}

// 从替换器中移除一个frame
void LRUKReplacer::Remove(frame_id_t frame_id) {
  std::lock_guard<std::mutex> lock_guard(latch_);
  
  // 忽略无效的frame_id
  if (frame_id > static_cast<int>(replacer_size_)) {
    return;
  }

  // 从历史队列中移除
  if (!hist_.empty()) {
    for (auto it : hist_) {
      if (it == frame_id) {
        frame_num_[it] = 0; // 重置访问计数
        mp_[it] = false; // 标记为不可驱逐
        break;
      }
    }
    hist_.remove(frame_id);
  }
  
  // 从当前缓存队列中移除
  if (!cur_.empty()) {
    for (auto it : cur_) {
      if (it == frame_id) {
        frame_num_[it] = 0; // 重置访问计数
        mp_[it] = false; // 标记为不可驱逐
        break;
      }
    }
    cur_.remove(frame_id);
  }
}

// 返回当前可驱逐的frame数量
auto LRUKReplacer::Size() -> size_t {
  std::lock_guard<std::mutex> lock_guard(latch_);
  size_t num = 0;
  
  // 统计历史队列中的可驱逐frame数量
  for (auto i : hist_) {
    if (mp_[i]) {
      num++;
    }
  }
  
  // 统计当前缓存队列中的可驱逐frame数量
  for (auto i : cur_) {
    if (mp_[i]) {
      num++;
    }
  }
  
  return num;
}

}  // namespace bustub

示例场景

假设 k=2，即我们在使用 LRU-2 算法，num_frames=3，即缓存最多可以存储3个页面。

访问顺序：A, B, C, A, B, D, A
过程说明：
- 初始状态：缓存为空。
- 访问A：将 A 添加到历史队列 hist_，因为 A 的访问次数为1。
- 访问B：将 B 添加到历史队列 hist_，因为 B 的访问次数为1。
- 访问C：将 C 添加到历史队列 hist_，因为 C 的访问次数为1。
- 再次访问A：A 的访问次数增加到2，将 A 从历史队列 hist_ 移动到当前缓存队列 cur_。
- 再次访问B：B 的访问次数增加到2，将 B 从历史队列 hist_ 移动到当前缓存队列 cur_。
- 访问D：此时缓存已满（cur_ 中有 A 和 B），需要驱逐一个页面。算法会先检查历史队列中的页面，发现 C 只被访问过1次，可以被驱逐。C 被移除，将 D 加入历史队列 hist_。
- 再次访问A：A 的访问次数增加到3，仍然保持在当前缓存队列 cur_ 中。
最终状态：
- 历史队列 hist_：D
- 当前缓存队列 cur_：A, B

LRU-K 算法有效地平衡了频繁访问页面和较久未访问页面之间的替换策略，使得那些在较长时间内多次被访问的页面能够留在缓存中，而那些仅在短时间内频繁被访问的页面则容易被淘汰。

总结：

历史队列 (hist_) 存储那些访问次数未达到 K 次的缓存块。
当前队列 (cur_) 存储那些访问次数达到 K 次的缓存块。
当缓存块的访问次数达到 K 次时，会从历史队列移动到当前队列。
当缓存空间不足时，优先淘汰历史队列中的缓存块，如果历史队列为空，则淘汰当前队列中的缓存块。

数据库中的Buffer pool Manage选用什么算法比较好呢？

LRU-K

LRU的考量：

存在两个用户，一个用户需要进行一次扫描操作，而另一个用户进行正常的随机访问操作。扫描操作其实很简单，例如遍历一张数据表等，但其带来的后果却是灾难的。扫描操作会快速引用大量页面，这对于传统 LRU 而言将迅速驱逐所有随机访问页面，但是扫描操作引用的页面几乎在之后很长一段时间内不再被引用，而随机访问页面才是真正需要被缓存的，这几乎使传统的 LRU 失效。对于这个问题 LRU-2 就能很好的解决。