curvefs/client: add throttle for disk cache

Cyber-SiKu · Jan 14, 2022 · 34de287 · 34de287
1 parent 29f51ac
commit 34de287
Show file tree

Hide file tree

Showing 6 changed files with 50 additions and 1 deletion.
diff --git a/curvefs/conf/client.conf b/curvefs/conf/client.conf
@@ -1,4 +1,4 @@
-#### mdsOpt
+##### mdsOpt
 # RPC total retry time with MDS
 mdsOpt.mdsMaxRetryMS=16000
 # The maximum timeout of RPC communicating with MDS.
@@ -144,6 +144,14 @@ diskCache.maxUsableSpaceBytes=107374182400
 diskCache.cmdTimeoutSec=300
 # directory of disk cache
 diskCache.cacheDir=/mnt/curvefs_cache  # __CURVEADM_TEMPLATE__ /curvefs/client/data/cache __CURVEADM_TEMPLATE__  __ANSIBLE_TEMPLATE__ /mnt/curvefs_disk_cache/{{ 99999999 | random | to_uuid | upper }} __ANSIBLE_TEMPLATE__
+# the write bps of disk cache
+diskCache.maxFlushBytes=83886080
+# the write iops of disk cache
+diskCache.maxFlushIops=100000000
+# the read bps of disk cache
+diskCache.maxReadFileBytes=83886080
+# the read iops of disk cache
+diskCache.maxReadFileIops=100000000
 
 #### common
 client.common.logDir=/data/logs/curvefs  # __CURVEADM_TEMPLATE__ /curvefs/client/logs __CURVEADM_TEMPLATE__

diff --git a/curvefs/src/client/common/config.cpp b/curvefs/src/client/common/config.cpp
@@ -119,6 +119,14 @@ void InitDiskCacheOption(Configuration *conf,
                               &diskCacheOption->maxUsableSpaceBytes);
     conf->GetValueFatalIfFail("diskCache.cmdTimeoutSec",
                               &diskCacheOption->cmdTimeoutSec);
+    conf->GetValueFatalIfFail("diskCache.maxFlushBytes",
+                              &diskCacheOption->maxFlushBytes);
+    conf->GetValueFatalIfFail("diskCache.maxFlushIops",
+                              &diskCacheOption->maxFlushIops);
+    conf->GetValueFatalIfFail("diskCache.maxReadFileBytes",
+                              &diskCacheOption->maxReadFileBytes);
+    conf->GetValueFatalIfFail("diskCache.maxReadFileIops",
+                              &diskCacheOption->maxReadFileIops);
 }
 
 void InitS3Option(Configuration *conf, S3Option *s3Opt) {

diff --git a/curvefs/src/client/common/config.h b/curvefs/src/client/common/config.h
@@ -87,6 +87,14 @@ struct DiskCacheOption {
     uint64_t maxUsableSpaceBytes;
     // the max time system command can run
     uint32_t cmdTimeoutSec;
+    // the write bps of disk cache
+    uint64_t maxFlushBytes;
+    // the read bps of disk cache
+    uint64_t maxReadFileBytes;
+    // the write iops of disk cache
+    uint64_t maxFlushIops;
+    // the read iops of disk cache
+    uint64_t maxReadFileIops;
 };
 
 struct S3ClientAdaptorOption {

diff --git a/curvefs/src/client/s3/disk_cache_manager.cpp b/curvefs/src/client/s3/disk_cache_manager.cpp
@@ -82,6 +82,15 @@ int DiskCacheManager::Init(S3Client *client,
     TrimRun();
     SetDiskInitUsedBytes();
     SetDiskFsUsedRatio();
+
+    ReadWriteThrottleParams params;
+    params.iopsWrite = ThrottleParams(option.diskCacheOpt.maxFlushIops, 0, 0);
+    params.bpsWrite = ThrottleParams(option.diskCacheOpt.maxFlushBytes, 0, 0);
+    params.iopsRead = ThrottleParams(option.diskCacheOpt.maxReadFileIops, 0, 0);
+    params.bpsRead = ThrottleParams(option.diskCacheOpt.maxReadFileBytes, 0, 0);
+
+    diskCacheThrottle_.UpdateThrottleParams(params);
+
     LOG(INFO) << "DiskCacheManager init success. "
               << ", cache dir is: " << cacheDir_
               << ", maxUsableSpaceBytes is: " << maxUsableSpaceBytes_
@@ -166,6 +175,8 @@ std::string DiskCacheManager::GetCacheWriteFullDir() {
 
 int DiskCacheManager::WriteDiskFile(const std::string fileName, const char *buf,
                                     uint64_t length, bool force) {
+    // write throttle
+    diskCacheThrottle_.Add(false, length);
     int ret = cacheWrite_->WriteDiskFile(fileName, buf, length, force);
     if (ret > 0)
         AddDiskUsedBytes(ret);
@@ -178,11 +189,15 @@ void DiskCacheManager::AsyncUploadEnqueue(const std::string objName) {
 
 int DiskCacheManager::ReadDiskFile(const std::string name, char *buf,
                                    uint64_t offset, uint64_t length) {
+    // read throttle
+    diskCacheThrottle_.Add(true, length);
     return cacheRead_->ReadDiskFile(name, buf, offset, length);
 }
 
 int DiskCacheManager::WriteReadDirect(const std::string fileName,
                  const char* buf, uint64_t length) {
+    // write hrottle
+    diskCacheThrottle_.Add(false, length);
     int ret = cacheRead_->WriteDiskFile(fileName, buf, length);;
     if (ret > 0)
         AddDiskUsedBytes(ret);

diff --git a/curvefs/src/client/s3/disk_cache_manager.h b/curvefs/src/client/s3/disk_cache_manager.h
@@ -30,6 +30,7 @@
 
 #include "src/common/concurrent/concurrent.h"
 #include "src/common/interruptible_sleeper.h"
+#include "src/common/throttle.h"
 #include "curvefs/src/common/wrap_posix.h"
 #include "curvefs/src/common/utils.h"
 #include "curvefs/src/client/s3/client_s3.h"
@@ -40,7 +41,10 @@ namespace curvefs {
 namespace client {
 
 using curvefs::common::PosixWrapper;
+using curve::common::ReadWriteThrottleParams;
 using curvefs::common::SysUtils;
+using curve::common::ThrottleParams;
+using curve::common::Throttle;
 using curvefs::client::common::S3ClientAdaptorOption;
 
 class DiskCacheManager {
@@ -149,6 +153,8 @@ class DiskCacheManager {
     S3Client *client_;
     std::shared_ptr<PosixWrapper> posixWrapper_;
     std::shared_ptr<DiskCacheMetric> metric_;
+
+    Throttle diskCacheThrottle_;
 };
 
 

diff --git a/curvefs/src/client/s3/disk_cache_manager_impl.h b/curvefs/src/client/s3/disk_cache_manager_impl.h
@@ -54,6 +54,10 @@ struct DiskCacheOption {
     bool forceFlush;
     uint64_t maxUsableSpaceBytes;
     uint32_t cmdTimeoutSec;
+    uint64_t maxFlushBytes;
+    uint64_t maxReadFileBytes;
+    uint64_t maxFlushIops;
+    uint64_t maxReadFileIops;
 };
 
 class DiskCacheManagerImpl {