[WIP][feat]curvefs/client: warmup manager

curvefs/src/client/BUILD

@@ -43,6 +43,8 @@ cc_library(
             "s3/*.h",
             "volume/*.cpp",
             "volume/*.h",
+            "warmup/*.h",
+            "warmup/*.cpp",
         ],
         exclude = ["main.cpp"],
     ),

curvefs/src/client/common/common.cpp

@@ -79,6 +79,7 @@ std::ostream &operator<<(std::ostream &os, MetaServerOpType optype) {
 }
 
 const char kCurveFsWarmupOpAdd[] = "add";
+const char kCurveFsWarmupOpQuery[] = "query";
 const char kCurveFsWarmupTypeList[] = "list";
 const char kCurveFsWarmupTypeSingle[] = "single";
 
@@ -87,6 +88,9 @@ WarmupOpType GetWarmupOpType(const std::string& op) {
     if (op == kCurveFsWarmupOpAdd) {
         ret = WarmupOpType::kWarmupOpAdd;
     }
+    if (op == kCurveFsWarmupOpQuery) {
+        ret = WarmupOpType::kWarmupOpQuery;
+    }
     return ret;
 }
 

curvefs/src/client/common/common.h

@@ -72,6 +72,7 @@ const char kCurveFsWarmupXAttr[] = "curvefs.warmup.op";
 enum class WarmupOpType {
     kWarmupOpUnknown = 0,
     kWarmupOpAdd = 1,
+    kWarmupOpQuery = 2,
 };
 
 WarmupOpType GetWarmupOpType(const std::string& op);

curvefs/src/client/curve_fuse_op.cpp

@@ -42,6 +42,7 @@
 #include "curvefs/src/client/metric/client_metric.h"
 #include "curvefs/src/common/metric_utils.h"
 #include "curvefs/src/common/dynamic_vlog.h"
+#include "curvefs/src/client/warmup/warmup_manager.h"
 
 using ::curve::common::Configuration;
 using ::curvefs::client::CURVEFS_ERROR;
@@ -279,25 +280,47 @@ void FuseOpGetAttr(fuse_req_t req, fuse_ino_t ino, struct fuse_file_info *fi) {
     fuse_reply_attr(req, &attr, g_fuseClientOption->attrTimeOut);
 }
 
-int AddWarmupTask(const std::string& type, const std::string& path) {
+int AddWarmupTask(curvefs::client::common::WarmupType type, fuse_ino_t key,
+                  const std::string &path) {
     int ret = 0;
-    switch (curvefs::client::common::GetWarmupType(type)) {
-        case curvefs::client::common::WarmupType::kWarmupTypeList:
-            g_ClientInstance->PutWarmTask(path);
-            break;
-        case curvefs::client::common::WarmupType::kWarmupTypeSingle:
-            g_ClientInstance->FetchDentryEnqueue(path);
-            break;
-        default:
-            // not support add warmup type (warmup single file/dir or filelist)
-            LOG(ERROR) << "not support warmup type, only support single/list";
-            ret = ERANGE;
+    bool result = true;
+    switch (type) {
+    case curvefs::client::common::WarmupType::kWarmupTypeList:
+        result = g_ClientInstance->PutWarmFilelistTask(key);
+        break;
+    case curvefs::client::common::WarmupType::kWarmupTypeSingle:
+        result = g_ClientInstance->PutWarmFileTask(key, path);
+        break;
+    default:
+        // not support add warmup type (warmup single file/dir or filelist)
+        LOG(ERROR) << "not support warmup type, only support single/list";
+        ret = EOPNOTSUPP;
+    }
+    if (!result) {
+        ret = ERANGE;
     }
     return ret;
 }
 
-int Warmup(const std::string& name, const std::string& value) {
+void QueryWarmupTask(fuse_ino_t key, std::string *data) {
+    curvefs::client::warmup::WarmupProgress progress;
+    bool ret = g_ClientInstance->GetWarmupProgress(key, &progress);
+    if (!ret) {
+        *data = "finished";
+    } else {
+        *data = std::to_string(progress.GetFinished()) + "/" +
+                std::to_string(progress.GetTotal());
+    }
+    VLOG(9) << "Warmup [" << key << "]" << *data;
+}
+
+int Warmup(fuse_ino_t key, const std::string& name, const std::string& value) {
     // warmup
+    if (g_ClientInstance->GetFsInfo()->fstype() != FSType::TYPE_S3) {
+        LOG(ERROR) << "warmup only support s3";
+        return EOPNOTSUPP;
+    }
+
     std::vector<std::string> opTypePath;
     curve::common::SplitString(value, "\n", &opTypePath);
     if (opTypePath.size() != 3) {
@@ -306,15 +329,17 @@ int Warmup(const std::string& name, const std::string& value) {
     }
     int ret = 0;
     switch (curvefs::client::common::GetWarmupOpType(opTypePath[0])) {
-        case curvefs::client::common::WarmupOpType::kWarmupOpAdd:
-            ret = AddWarmupTask(opTypePath[1], opTypePath[2]);
-            if (ret != 0) {
-                LOG(ERROR) << name << " has invalid xattr value " << value;
-            }
-            break;
-        default:
+    case curvefs::client::common::WarmupOpType::kWarmupOpAdd:
+        ret =
+            AddWarmupTask(curvefs::client::common::GetWarmupType(opTypePath[1]),
+                          key, opTypePath[2]);
+        if (ret != 0) {
             LOG(ERROR) << name << " has invalid xattr value " << value;
-            ret = ERANGE;
+        }
+        break;
+    default:
+        LOG(ERROR) << name << " has invalid xattr value " << value;
+        ret = ERANGE;
     }
     return ret;
 }
@@ -327,7 +352,7 @@ void FuseOpSetXattr(fuse_req_t req, fuse_ino_t ino, const char* name,
             << " flags " << flags;
     if (strcmp(name, curvefs::client::common::kCurveFsWarmupXAttr) == 0) {
         // warmup
-        int code = Warmup(name, xattrValue);
+        int code = Warmup(ino, name, xattrValue);
         fuse_reply_err(req, code);
     } else {
         // set xattr
@@ -340,7 +365,18 @@ void FuseOpSetXattr(fuse_req_t req, fuse_ino_t ino, const char* name,
 }
 
 void FuseOpGetXattr(fuse_req_t req, fuse_ino_t ino, const char *name,
-    size_t size) {
+                    size_t size) {
+    if (strcmp(name, curvefs::client::common::kCurveFsWarmupXAttr) == 0) {
+        // warmup
+        std::string data;
+        QueryWarmupTask(ino, &data);
+        if (size == 0) {
+            fuse_reply_xattr(req, data.length());
+        } else {
+            fuse_reply_buf(req, data.data(), data.length());
+        }
+        return;
+    }
     InflightGuard guard(&g_clientOpMetric->opGetXattr.inflightOpNum);
     LatencyUpdater updater(&g_clientOpMetric->opGetXattr.latency);
     std::string buf;