add limiter

.gitmodules

@@ -1,6 +1,3 @@
 [submodule "rdsn"]
   path = rdsn
   url = https://github.com/XiaoMi/rdsn.git
-[submodule "rocksdb"]
-  path = rocksdb
-  url = https://github.com/XiaoMi/pegasus-rocksdb.git

.travis.yml

@@ -1,10 +1,8 @@
 sudo: required
-dist: trusty
+dist: bionic
 
 language: cpp
 
-os: linux
-
 compiler:
     - gcc
 
@@ -14,30 +12,19 @@ addons:
       - libsnappy-dev
       - liblz4-dev
       - clang-format-3.9
+      - libboost-all-dev
+      - libaio-dev
+      - libzstd-dev
 
 cache:
     - ccache
     - apt
 
-before_install:
-    - wget https://github.com/XiaoMi/pegasus-common/releases/download/deps/build-depends.tar.gz
-    - tar xfz build-depends.tar.gz
-    - rm -f build-depends.tar.gz
-    - cd packages
-    - ls | xargs sudo dpkg -i --force-depends
-    - cd ..
-
-install:
-    - wget https://github.com/facebook/zstd/archive/v1.3.7.zip; unzip v1.3.7; cd zstd-1.3.7;
-    - mkdir cmake-build; cd cmake-build; cmake -DCMAKE_INSTALL_PREFIX=/usr -DCMAKE_INSTALL_LIBDIR=lib -DZSTD_BUILD_PROGRAMS=OFF ../build/cmake; sudo make install -j8;
-    - cd ../..
-    - rm -rf v1.3.7.zip zstd-1.3.7;
-
 before_script:
     - cd rdsn/thirdparty
-    - wget https://github.com/XiaoMi/pegasus-common/releases/download/deps/pegasus-thirdparty-prebuild.tar.gz
-    - tar xf pegasus-thirdparty-prebuild.tar.gz
-    - rm -f pegasus-thirdparty-prebuild.tar.gz
+    - wget https://github.com/XiaoMi/pegasus-common/releases/download/deps/pegasus-thirdparty-prebuild.zip
+    - unzip pegasus-thirdparty-prebuild.zip
+    - rm -f pegasus-thirdparty-prebuild.zip
     - cd ../..
     - ulimit -c unlimited -S
 

run.sh

@@ -58,9 +58,9 @@ function usage_build()
     echo "   -h|--help             print the help info"
     echo "   -t|--type             build type: debug|release, default is release"
     echo "   -s|--serialize        serialize type: dsn|thrift|proto, default is thrift"
-    echo "   -c|--clear            clear rdsn/rocksdb/pegasus before building, not clear thirdparty"
-    echo "   -cc|--half-clear      clear pegasus before building, not clear thirdparty/rdsn/rocksdb"
-    echo "   --clear_thirdparty    clear thirdparty/rdsn/rocksdb/pegasus before building"
+    echo "   -c|--clear            clear rdsn/pegasus before building, not clear thirdparty"
+    echo "   -cc|--half-clear      clear pegasus before building, not clear thirdparty/rdsn"
+    echo "   --clear_thirdparty    clear thirdparty/rdsn/pegasus before building"
     echo "   --compiler            specify c and cxx compiler, sperated by ','"
     echo "                         e.g., \"gcc,g++\" or \"clang-3.9,clang++-3.9\""
     echo "                         default is \"gcc,g++\""
@@ -226,79 +226,6 @@ function run_build()
         exit 1
     fi
 
-    echo "INFO: start build rocksdb..."
-    ROCKSDB_BUILD_DIR="$ROOT/rocksdb/build"
-    ROCKSDB_BUILD_OUTPUT="$ROCKSDB_BUILD_DIR/output"
-    CMAKE_OPTIONS="-DCMAKE_C_COMPILER=$C_COMPILER -DCMAKE_CXX_COMPILER=$CXX_COMPILER -DWITH_LZ4=ON -DWITH_ZSTD=ON -DWITH_SNAPPY=ON -DWITH_BZ2=OFF -DWITH_TESTS=OFF -DWITH_GFLAGS=OFF -DUSE_RTTI=ON -DCMAKE_BUILD_TYPE=Release -DCMAKE_CXX_FLAGS=-g"
-    if [ "$WARNING_ALL" == "YES" ]
-    then
-        echo "WARNING_ALL=YES"
-        CMAKE_OPTIONS="$CMAKE_OPTIONS -DWARNING_ALL=TRUE"
-    else
-        echo "WARNING_ALL=NO"
-    fi
-    if [ "$ENABLE_GCOV" == "YES" ]
-    then
-        echo "ENABLE_GCOV=YES"
-        CMAKE_OPTIONS="$CMAKE_OPTIONS -DENABLE_GCOV=TRUE"
-    else
-        echo "ENABLE_GCOV=NO"
-    fi
-    if [ "$BUILD_TYPE" == "debug" ]
-    then
-        echo "BUILD_TYPE=debug"
-        CMAKE_OPTIONS="$CMAKE_OPTIONS -DCMAKE_BUILD_TYPE=RelWithDebInfo"
-    else
-        echo "BUILD_TYPE=release"
-    fi
-
-    if [ -f $ROCKSDB_BUILD_DIR/CMAKE_OPTIONS ]
-    then
-        LAST_OPTIONS=`cat $ROCKSDB_BUILD_DIR/CMAKE_OPTIONS`
-        if [ "$CMAKE_OPTIONS" != "$LAST_OPTIONS" ]
-        then
-            echo "WARNING: CMAKE_OPTIONS has changed from last build, clear environment first"
-            CLEAR=YES
-        fi
-    fi
-
-    if [ "$CLEAR" == "YES" ] && [ -d "$ROCKSDB_BUILD_DIR" ]
-    then
-        echo "Clear $ROCKSDB_BUILD_DIR ..."
-        rm -rf $ROCKSDB_BUILD_DIR
-    fi
-
-    if [ ! -f $ROCKSDB_BUILD_DIR/Makefile ]; then
-        echo "Running cmake..."
-        mkdir -p $ROCKSDB_BUILD_DIR
-        cd $ROCKSDB_BUILD_DIR
-        echo "$CMAKE_OPTIONS" >CMAKE_OPTIONS
-        cmake .. -DCMAKE_INSTALL_PREFIX=$ROCKSDB_BUILD_OUTPUT $CMAKE_OPTIONS
-        if [ $? -ne 0 ]; then
-            echo "ERROR: cmake failed"
-            exit 1
-        fi
-    else
-        cd $ROCKSDB_BUILD_DIR
-    fi
-
-    echo "Building..."
-    if [ "$RUN_VERBOSE" == "YES" ]
-    then
-        echo "RUN_VERBOSE=YES"
-        MAKE_OPTIONS="$MAKE_OPTIONS VERBOSE=1"
-    else
-        echo "RUN_VERBOSE=NO"
-    fi
-    make install -j $JOB_NUM $MAKE_OPTIONS
-    if [ $? -ne 0 ]
-    then
-        echo "ERROR: build rocksdb failed"
-        exit 1
-    else
-        echo "Build rocksdb succeed"
-    fi
-
     echo "INFO: start build pegasus..."
     cd $ROOT/src
     C_COMPILER="$C_COMPILER" CXX_COMPILER="$CXX_COMPILER" BUILD_TYPE="$BUILD_TYPE" \

src/CMakeLists.txt

@@ -38,14 +38,6 @@ link_directories(${PEGASUS_ROOT}/lib)
 set(CMAKE_INSTALL_PREFIX ${PEGASUS_ROOT} CACHE STRING "" FORCE)
 message (STATUS "pegasus Installation directory: CMAKE_INSTALL_PREFIX = " ${CMAKE_INSTALL_PREFIX})
 
-# rocksdb
-list(APPEND CMAKE_MODULE_PATH "${PEGASUS_PROJECT_DIR}/rocksdb/cmake/modules/")
-find_package(snappy)
-find_package(zstd)
-find_package(lz4)
-set(CMAKE_PREFIX_PATH ${PEGASUS_PROJECT_DIR}/rocksdb/build/output;${CMAKE_PREFIX_PATH})
-find_package(RocksDB REQUIRED) # RocksDB::rocksdb means librocksdb.a
-
 add_subdirectory(base)
 add_subdirectory(reporter)
 add_subdirectory(base/test)

src/build.sh

@@ -119,10 +119,6 @@ fi
 
 echo "CMAKE_OPTIONS=$CMAKE_OPTIONS"
 
-#rocksdb enable jemalloc by default, but we use regular malloc.
-MAKE_OPTIONS="$MAKE_OPTIONS DISABLE_JEMALLOC=1"
-echo "MAKE_OPTIONS=$MAKE_OPTIONS"
-
 echo "#############################################################################"
 
 if [ -f $BUILD_DIR/CMAKE_OPTIONS ]
@@ -183,4 +179,3 @@ else
     echo "Build pegasus succeed"
 fi
 cd ..
-

src/server/CMakeLists.txt

@@ -16,6 +16,7 @@ set(MY_PROJ_LIBS
     dsn.replication.ddlclient
     dsn.block_service.local
     dsn.block_service.fds
+    dsn.block_service
     dsn.failure_detector
     dsn.failure_detector.multimaster
     dsn.replication.zookeeper_provider

src/server/pegasus_server_impl.cpp

@@ -42,12 +42,16 @@ static bool chkpt_init_from_dir(const char *name, int64_t &decree)
            std::string(name) == chkpt_get_dir_name(decree);
 }
 
+std::shared_ptr<rocksdb::RateLimiter> pegasus_server_impl::_s_rate_limiter;
+int64_t pegasus_server_impl::_rocksdb_limiter_last_total_through;
 std::shared_ptr<rocksdb::Cache> pegasus_server_impl::_s_block_cache;
 ::dsn::task_ptr pegasus_server_impl::_update_server_rdb_stat;
 ::dsn::perf_counter_wrapper pegasus_server_impl::_pfc_rdb_block_cache_mem_usage;
+::dsn::perf_counter_wrapper pegasus_server_impl::_pfc_rdb_write_limiter_rate_bytes;
 const std::string pegasus_server_impl::COMPRESSION_HEADER = "per_level:";
 const std::string pegasus_server_impl::DATA_COLUMN_FAMILY_NAME = "default";
 const std::string pegasus_server_impl::META_COLUMN_FAMILY_NAME = "pegasus_meta_cf";
+const std::chrono::seconds pegasus_server_impl::kServerStatUpdateTimeSec = std::chrono::seconds(10);
 
 void pegasus_server_impl::parse_checkpoints()
 {
@@ -1424,16 +1428,18 @@ ::dsn::error_code pegasus_server_impl::start(int argc, char **argv)
                                       [this]() { this->update_replica_rocksdb_statistics(); },
                                       _update_rdb_stat_interval);
 
-    // Block cache is a singleton on this server shared by all replicas, its metrics update task
+    // These counters are singletons on this server shared by all replicas, its metrics update task
     // should be scheduled once an interval on the server view.
     static std::once_flag flag;
     std::call_once(flag, [&]() {
         // The timer task will always running even though there is no replicas
+        dassert(kServerStatUpdateTimeSec.count() != 0,
+                "kServerStatUpdateTimeSec shouldn't be zero");
         _update_server_rdb_stat = ::dsn::tasking::enqueue_timer(
             LPC_REPLICATION_LONG_COMMON,
             nullptr, // TODO: the tracker is nullptr, we will fix it later
             [this]() { update_server_rocksdb_statistics(); },
-            _update_rdb_stat_interval);
+            kServerStatUpdateTimeSec);
     });
 
     // initialize cu calculator and write service after server being initialized.
@@ -2107,9 +2113,16 @@ void pegasus_server_impl::update_server_rocksdb_statistics()
     if (_s_block_cache) {
         uint64_t val = _s_block_cache->GetUsage();
         _pfc_rdb_block_cache_mem_usage->set(val);
-        dinfo_f("_pfc_rdb_block_cache_mem_usage: {} bytes", val);
-    } else {
-        dinfo("_pfc_rdb_block_cache_mem_usage: 0 bytes because block cache is disabled");
+    }
+
+    // Update _pfc_rdb_write_limiter_rate_bytes
+    if (_s_rate_limiter) {
+        uint64_t current_total_through = _s_rate_limiter->GetTotalBytesThrough();
+        uint64_t through_bytes_per_sec =
+            (current_total_through - _rocksdb_limiter_last_total_through) /
+            kServerStatUpdateTimeSec.count();
+        _pfc_rdb_write_limiter_rate_bytes->set(through_bytes_per_sec);
+        _rocksdb_limiter_last_total_through = current_total_through;
     }
 }
 

src/server/pegasus_server_impl.h

@@ -14,6 +14,7 @@
 #include <rrdb/rrdb_types.h>
 #include <rrdb/rrdb.server.h>
 #include <gtest/gtest_prod.h>
+#include <rocksdb/rate_limiter.h>
 
 #include "key_ttl_compaction_filter.h"
 #include "pegasus_scan_context.h"
@@ -309,6 +310,7 @@ class pegasus_server_impl : public ::dsn::apps::rrdb_service
     ::dsn::error_code flush_all_family_columns(bool wait);
 
 private:
+    static const std::chrono::seconds kServerStatUpdateTimeSec;
     static const std::string COMPRESSION_HEADER;
     // Column family names.
     static const std::string DATA_COLUMN_FAMILY_NAME;
@@ -338,6 +340,8 @@ class pegasus_server_impl : public ::dsn::apps::rrdb_service
     rocksdb::ColumnFamilyHandle *_data_cf;
     rocksdb::ColumnFamilyHandle *_meta_cf;
     static std::shared_ptr<rocksdb::Cache> _s_block_cache;
+    static std::shared_ptr<rocksdb::RateLimiter> _s_rate_limiter;
+    static int64_t _rocksdb_limiter_last_total_through;
     volatile bool _is_open;
     uint32_t _pegasus_data_version;
     std::atomic<int64_t> _last_durable_decree;
@@ -381,6 +385,7 @@ class pegasus_server_impl : public ::dsn::apps::rrdb_service
 
     // rocksdb internal statistics
     // server level
+    static ::dsn::perf_counter_wrapper _pfc_rdb_write_limiter_rate_bytes;
     static ::dsn::perf_counter_wrapper _pfc_rdb_block_cache_mem_usage;
     // replica level
     ::dsn::perf_counter_wrapper _pfc_rdb_sst_count;

src/server/pegasus_server_impl_init.cpp

@@ -4,6 +4,7 @@
 
 #include "pegasus_server_impl.h"
 
+#include <dsn/utility/flags.h>
 #include <rocksdb/filter_policy.h>
 
 #include "capacity_unit_calculator.h"
@@ -14,6 +15,18 @@
 
 namespace pegasus {
 namespace server {
+
+DSN_DEFINE_int64(
+    "pegasus.server",
+    rocksdb_limiter_max_write_megabytes_per_sec,
+    500,
+    "max rate of rocksdb flush and compaction(MB/s), if less than or equal to 0 means close limit");
+
+DSN_DEFINE_bool("pegasus.server",
+                rocksdb_limiter_auto_tune_enable,
+                false,
+                "whether to enable write rate auto tune when open rocksdb write limit");
+
 pegasus_server_impl::pegasus_server_impl(dsn::replication::replica *r)
     : dsn::apps::rrdb_service(r),
       _db(nullptr),
@@ -251,6 +264,22 @@ pegasus_server_impl::pegasus_server_impl(dsn::replication::replica *r)
         tbl_opts.block_cache = _s_block_cache;
     }
 
+    // FLAGS_rocksdb_limiter_max_write_megabytes_per_sec <= 0 means close the rate limit.
+    // For more detail arguments see
+    // https://github.com/facebook/rocksdb/blob/v6.6.4/include/rocksdb/rate_limiter.h#L111-L137
+    if (FLAGS_rocksdb_limiter_max_write_megabytes_per_sec > 0) {
+        static std::once_flag flag;
+        std::call_once(flag, [&]() {
+            _s_rate_limiter = std::shared_ptr<rocksdb::RateLimiter>(rocksdb::NewGenericRateLimiter(
+                FLAGS_rocksdb_limiter_max_write_megabytes_per_sec << 20,
+                100000,
+                10,
+                rocksdb::RateLimiter::Mode::kWritesOnly,
+                FLAGS_rocksdb_limiter_auto_tune_enable));
+        });
+        _db_opts.rate_limiter = _s_rate_limiter;
+    }
+
     // Bloom filter configurations.
     bool disable_bloom_filter = dsn_config_get_value_bool(
         "pegasus.server", "rocksdb_disable_bloom_filter", false, "Whether to disable bloom filter");
@@ -400,8 +429,8 @@ pegasus_server_impl::pegasus_server_impl(dsn::replication::replica *r)
         COUNTER_TYPE_NUMBER,
         "statistic the total count of rocksdb block cache");
 
-    // Block cache is a singleton on this server shared by all replicas, so we initialize
-    // `_pfc_rdb_block_cache_mem_usage` only once.
+    // These counters are singletons on this server shared by all replicas, so we initialize
+    // them only once.
     static std::once_flag flag;
     std::call_once(flag, [&]() {
         _pfc_rdb_block_cache_mem_usage.init_global_counter(
@@ -410,6 +439,13 @@ pegasus_server_impl::pegasus_server_impl(dsn::replication::replica *r)
             "rdb.block_cache.memory_usage",
             COUNTER_TYPE_NUMBER,
             "statistic the memory usage of rocksdb block cache");
+
+        _pfc_rdb_write_limiter_rate_bytes.init_global_counter(
+            "replica",
+            "app.pegasus",
+            "rdb.write_limiter_rate_bytes",
+            COUNTER_TYPE_NUMBER,
+            "statistic the through bytes of rocksdb write rate limiter");
     });
 
     snprintf(name, 255, "rdb.index_and_filter_blocks.memory_usage@%s", str_gpid.c_str());

src/server/test/CMakeLists.txt

@@ -21,6 +21,7 @@ set(MY_PROJ_LIBS
         dsn.replication.ddlclient
         dsn.block_service.local
         dsn.block_service.fds
+        dsn.block_service
         dsn.failure_detector
         dsn.failure_detector.multimaster
         dsn.replication.zookeeper_provider

src/shell/CMakeLists.txt

@@ -18,6 +18,7 @@ set(MY_PROJ_LIBS
     dsn.replication.ddlclient
     dsn.block_service.local
     dsn.block_service.fds
+    dsn.block_service
     dsn.failure_detector
     dsn.failure_detector.multimaster
     pegasus_client_static