feat: add rate limiter

conf/pika.conf

@@ -157,3 +157,6 @@ max-bytes-for-level-multiplier : 10
 # optimize-filters-for-hits: no
 # https://github.com/facebook/rocksdb/wiki/Leveled-Compaction#levels-target-size
 # level-compaction-dynamic-level-bytes: no
+
+# rate limiter bandwidth, default 200MB
+#rate-limiter-bandwidth : 209715200

include/pika_conf.h

@@ -93,6 +93,7 @@ class PikaConf : public slash::BaseConf {
   int max_conn_rbuf_size()                          { return max_conn_rbuf_size_.load(); }
   int consensus_level()                             { return consensus_level_.load(); }
   int replication_num()                             { return replication_num_.load(); }
+  int64_t rate_limiter_bandwidth()                  { RWLock l(&rwlock_, false); return rate_limiter_bandwidth_; }
 
   // Immutable config items, we don't use lock.
   bool daemonize()                                  { return daemonize_; }
@@ -349,6 +350,7 @@ class PikaConf : public slash::BaseConf {
   bool pin_l0_filter_and_index_blocks_in_cache_;
   bool optimize_filters_for_hits_;
   bool level_compaction_dynamic_level_bytes_;
+  int64_t rate_limiter_bandwidth_ = 200 * 1024 * 1024; // 200M
   std::atomic<int> sync_window_size_;
   std::atomic<int> max_conn_rbuf_size_;
   std::atomic<int> consensus_level_;

src/blackwidow/include/blackwidow/blackwidow.h

@@ -15,6 +15,7 @@
 
 #include "rocksdb/status.h"
 #include "rocksdb/options.h"
+#include "rocksdb/rate_limiter.h"
 #include "rocksdb/slice.h"
 #include "rocksdb/table.h"
 #include "rocksdb/filter_policy.h"
@@ -62,17 +63,10 @@ class LRUCache;
 struct BlackwidowOptions {
   rocksdb::Options options;
   rocksdb::BlockBasedTableOptions table_options;
-  size_t block_cache_size;
-  bool share_block_cache;
-  size_t statistics_max_size;
-  size_t small_compaction_threshold;
-
-  explicit BlackwidowOptions()
-      : block_cache_size(0),
-        share_block_cache(false),
-        statistics_max_size(0),
-        small_compaction_threshold(5000) {}
-
+  size_t block_cache_size = 0;
+  bool share_block_cache = false;
+  size_t statistics_max_size = 0;
+  size_t small_compaction_threshold = 5000;
   Status ResetOptions(const OptionType& option_type,
                       const std::unordered_map<std::string, std::string>& options_map);
 };

src/pika_admin.cc

@@ -1739,6 +1739,12 @@ void ConfigCmd::ConfigGet(std::string& ret) {
     EncodeInt32(&config_body, g_pika_conf->consensus_level());
   }
 
+  if (slash::stringmatch(pattern.data(), "rate-limiter-bandwidth", 1)) {
+    elements += 2;
+    EncodeString(&config_body, "rate-limiter-bandwidth");
+    EncodeInt64(&config_body, g_pika_conf->rate_limiter_bandwidth());
+  }
+
   std::stringstream resp;
   resp << "*" << std::to_string(elements) << "\r\n" << config_body;
   ret = resp.str();

src/pika_conf.cc

@@ -375,6 +375,12 @@ int PikaConf::Load()
     max_write_buffer_size_ = 10737418240;  // 10Gb
   }
 
+  // rate-limiter-bandwidth
+  GetConfInt64("rate-limiter-bandwidth", &rate_limiter_bandwidth_);
+  if (rate_limiter_bandwidth_ <= 0 ) {
+    rate_limiter_bandwidth_ = 200 * 1024 * 1024;       // 200MB
+  }
+
   // max_write_buffer_num
   max_write_buffer_num_ = 2;
   GetConfInt("max-write-buffer-num", &max_write_buffer_num_);

src/pika_server.cc

@@ -1660,6 +1660,9 @@ void PikaServer::InitBlackwidowOptions() {
       rocksdb::NewLRUCache(bw_options_.block_cache_size, g_pika_conf->num_shard_bits());
   }
 
+  bw_options_.options.rate_limiter = 
+    std::shared_ptr<rocksdb::RateLimiter>(rocksdb::NewGenericRateLimiter(g_pika_conf->rate_limiter_bandwidth()));
+
   // For Blackwidow small compaction
   bw_options_.statistics_max_size = g_pika_conf->max_cache_statistic_keys();
   bw_options_.small_compaction_threshold =