这个请求频率限制器线程安全吗？

Posted 2023-02-16

【中文标题】这个请求频率限制器线程安全吗？

【英文标题】：Is this request frequency limiter thread safe?

【发布时间】：2021-09-10 12:42:36

【问题描述】：

为了防止服务器压力过大，我用滑动窗口算法实现了一个请求频率限制器，可以根据参数判断当前请求是否允许通过。为了实现算法的线程安全，我使用原子类型来控制窗口的滑动步数，并使用unique_lock来实现当前窗口请求总数的正确总和。 但是我不确定我的实现是否是线程安全的，如果是安全的，是否会影响服务性能。有没有更好的方法来实现它？

class SlideWindowLimiter

public:
  bool TryAcquire();
  void SlideWindow(int64_t window_number);

private:
  int32_t limit_;   // maximum number of window requests
  int32_t split_num_;   // subwindow number
  int32_t window_size_; // the big window
  int32_t sub_window_size_;  // size of subwindow = window_size_ / split_number
  
  int16_t index_0;  //the index of window vector
  std::mutex mtx_;
  std::vector<int32_t> sub_windows_;  // window vector 
  std::atomic<int64_t> start_time_0;  //start time of limiter


bool SlideWindowLimiter::TryAcquire() 
   int64_t cur_time = std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::system_clock::now().time_since_epoch()).count();
   auto time_stamp = start_time_.load();
   int64_t window_num = std::max(cur_time - window_size_ - start_time_, int64_t(0)) / sub_window_size_;
    
   std::unique_lock<std::mutex> guard(mtx_, std::defer_lock);
   if (window_num > 0  && start_time_.compare_exchange_strong(time_stamp, start_time_.load() + window_num*sub_window_size_)) 
     guard.lock();
     SlideWindow(window_num);
     guard.unlock();
   
 
   monitor_->TotalRequestQps();
   
     guard.lock();
     int32_t total_req = 0;
     std::cout<<" "<<std::endl;
     for(auto &p : sub_windows_) 
       std::cout<<p<<" "<<std::this_thread::get_id()<<std::endl;
       total_req += p;
     
 
     if(total_req >= limit_) 
       monitor_->RejectedRequestQps();
       return false;
      else 
       monitor_->PassedRequestQps();
       sub_windows_[index_] += 1;
       return true;
     
     guard.unlock();
   
 

void SlideWindowLimiter::SlideWindow(int64_t window_num) 
   int64_t slide_num = std::min(window_num, int64_t(split_num_));  
   for(int i = 0; i < slide_num; i++)
     index_ += 1;
     index_ = index_ % split_num_;
     sub_windows_[index_] = 0;
   
 

【参考方案1】：

首先，线程安全是一个相对属性。两个操作序列相对于彼此是线程安全的。单个代码本身不能是线程安全的。

我将改为回答“我处理线程的方式是否可以使用其他合理的代码做出合理的线程安全保证”。

答案是“不”。

我发现了一个具体问题；您对 atomic 和 compare_exchange_strong 的使用不在循环中，并且您在没有适当照顾的情况下在多个位置以原子方式访问 start_time_。如果start_time_ 在周期中与您读取和写入的 3 个点发生了变化，则返回 false，并且无法调用 SlideWindow，那么...继续进行。

我想不出为什么这是对争用的合理回应，所以这是“不，这段代码不是为了在使用它的多个线程下合理运行而编写的”。

---

您的代码中有很多难闻的气味。您将并发代码与一大堆状态混合在一起，这意味着不清楚哪些互斥锁在保护哪些数据。

您的代码中有一个从未定义过的指针。也许它应该是一个全局变量？

您在一行上使用多个<< 写信给cout。在多线程环境中这是一个糟糕的计划。即使你的cout 是并发强化的，你也会得到加扰的输出。构建一个缓冲区字符串并执行一个<<。

您正在通过后门在函数之间传递数据。 index_ 例如。一个函数设置一个成员变量，另一个函数读取它。有没有可能被另一个线程编辑？难以审计，但似乎有合理的可能性；你将它设置在一个.lock()，然后是.unlock()，然后在稍后的lock() 中读取它，就好像它处于合理状态一样。更重要的是，你用它来访问一个向量；如果向量或索引以计划外的方式更改，则可能会崩溃或导致内存损坏。

...

如果这段代码在生产中没有一堆竞争条件、崩溃等情况，我会感到震惊。我没有看到任何试图证明此代码是并发安全的，或将其简化到易于草拟证明的程度的迹象。

在实际的实际实践中，任何您尚未证明并发安全的代码在并发使用时都是不安全的。所以复杂的并发代码几乎可以保证并发使用是不安全的。

...

从一个非常非常简单的模型开始。如果您有一个互斥锁和一些数据，则将该互斥锁和数据放入一个结构中，这样您就可以确切地知道该互斥锁在保护什么。

如果您正在使用原子，请不要在与其他变量混合的其他代码中间使用它。把它放在它自己的类中。给那个类起一个名字，代表一些具体的语义，最好是你在别处找到的那些。描述它应该做什么，以及方法前后保证什么。然后使用它。

在其他地方，避免任何类型的全局状态。这包括用于传递状态的类成员变量。将数据从一个函数显式传递到另一个函数。避免指向任何可变对象的指针。

如果您的数据是自动存储中的所有值类型和指向不可变（在线程的生命周期内永远不会改变）数据的指针，则该数据不能直接参与竞争条件。

剩余的数据被捆绑起来，并在一个尽可能小的地方设置防火墙，您可以查看您如何与之交互并确定您是否搞砸了。

...

多线程编程很难，尤其是在具有可变数据的环境中。如果你不努力证明你的代码是正确的，你就会产生不正确的代码，而且你不会知道。

嗯，根据我的经验，我知道；所有没有明显试图以容易表明它是正确的方式运行的代码都是不正确的。如果代码很旧并且有十多年的大量补丁，那么错误可能不太可能并且更难找到，但它可能仍然不正确。如果是新代码，可能更容易发现错误。

【讨论】：

感谢您详细的cmets和建议，让我受益匪浅。我没有发布完整的代码，所以有些地方看起来很奇怪。我会再次编辑它。而关于“如果start_time_ 的周期与你从中读取和写入的3 个点发生变化，你返回false，并且调用SlideWindow 失败，那么……就好像你有一样继续”，我有点困惑。

我使用CAS来保证算法在新的时间子窗口中只能滑动一次。因为可能存在这样的场景：几乎同时有多个请求到达，而此时的时间戳足够长导致子窗口滑动。正确的算法是在窗口滑动并更新开始时间后，此时或以后的其他请求只需要计入当前子窗口的请求数，直到时间戳减去当前的@ 987654334@ 足够长，可以生成新的滑动。

CAS其实是用来正确更新start_time_的。

@salmon 仅当您不关心不准确性或假设代码与时钟相比速度更快时。两个线程具有相同的开始时间和不同的窗口编号，它们在 CAS 上竞争，无论哪个最先更新，另一个被忽略。然后两者都登录到获胜的窗口。试图写出前置/后置条件，你无法得到合理的条件。如果您不关心有时登录到错误的窗口，它会起作用。我可能是错的，可能会出现更严重的问题；如果没有简单有力且合理的保证，有时很难判断它会出错。

当两个线程启动时间相同时，一触CAS更新start_time_，CAS返回true，if的body中的index_会同时更新，之后，它会被计入新的子窗口（@98​​7654340@控制当前线程所属的窗口），虽然另一个线程接触到CAS并返回false，但它会被计入与第一个相同的子窗口（现在，两个线程可能有相同的start_time和index_，它们是类SlideWindowLimiter的变量）