support discard

conf/client.conf

@@ -142,3 +142,13 @@ global.metricDummyServerStartPort=9000
 
 # 是否关闭健康检查: true/关闭 false/不关闭
 global.turnOffHealthCheck=true
+
+# 是否开启discard
+discard.enableDiscard=true
+
+# discard粒度
+discard.discardGranularity=4096
+
+# discard清理任务延迟时间(毫秒)
+discard.discardTaskDelayMs=60000
+

conf/cs_client.conf

@@ -142,3 +142,12 @@ global.metricDummyServerStartPort=9000
 # session map文件，存储打开文件的filename到path的映射
 #
 global.sessionMapPath=./session_map.json
+
+# 是否开启discard
+discard.enableDiscard=false
+
+# discard粒度
+discard.discardGranularity=4096
+
+# discard清理任务延迟时间(毫秒)
+discard.discardTaskDelayMs=60000

conf/mds.conf

@@ -26,6 +26,8 @@ mds.segment.alloc.periodic.persistInterMs=10000
 # 出错情况下的重试间隔,单位ms
 mds.segment.alloc.retryInterMs=1000
 
+mds.segment.discard.scanIntevalMs=5000
+
 
 # leader竞选时会创建session, 单位是秒(go端代码的接口这个值的单位就是s)
 # 该值和etcd集群election timeout相关.

conf/py_client.conf

@@ -136,3 +136,12 @@ global.metricDummyServerStartPort=10000
 # session map文件，存储打开文件的filename到path的映射
 #
 global.sessionMapPath=./session_map.json
+
+# 是否开启discard
+discard.enableDiscard=false
+
+# discard粒度
+discard.discardGranularity=4096
+
+# discard清理任务延迟时间(毫秒)
+discard.discardTaskDelayMs=60000

conf/snap_client.conf

@@ -142,3 +142,12 @@ global.metricDummyServerStartPort=9000
 # session map文件，存储打开文件的filename到path的映射
 #
 global.sessionMapPath=./session_map.json
+
+# 是否开启discard
+discard.enableDiscard=false
+
+# discard粒度
+discard.discardGranularity=4096
+
+# discard清理任务延迟时间(毫秒)
+discard.discardTaskDelayMs=60000

curve-ansible/roles/generate_config/templates/client.conf.j2

@@ -145,3 +145,12 @@ global.turnOffHealthCheck={{ client_turn_off_health_check }}
 # session map文件，存储打开文件的filename到path的映射
 #
 global.sessionMapPath={{ client_session_map_path }}
+
+# 是否开启discard
+discard.enableDiscard={{ client_discard_enable }}
+
+# discard粒度
+discard.discardGranularity={{ client_discard_granularity }}
+
+# discard清理任务延迟时间(毫秒)
+discard.discardTaskDelayMs={{ client_discard_task_delay_ms }}

curve-ansible/roles/generate_config/templates/mds.conf.j2

@@ -200,3 +200,5 @@ mds.snapshotcloneclient.addr={{ mds_snapshotcloneclient_addr }}
 mds.common.logDir={{ mds_common_log_dir }}
 # 单元测试情况下
 # mds.common.logDir=./runlog/
+
+mds.segment.discard.scanIntevalMs=5000

include/client/libcurve.h

@@ -111,6 +111,7 @@ const char* ErrorNum2ErrorName(LIBCURVE_ERROR err);
 typedef enum LIBCURVE_OP {
     LIBCURVE_OP_READ,
     LIBCURVE_OP_WRITE,
+    LIBCURVE_OP_DISCARD,
     LIBCURVE_OP_MAX,
 } LIBCURVE_OP;
 
@@ -215,6 +216,16 @@ int Read(int fd, char* buf, off_t offset, size_t length);
  */
 int Write(int fd, const char* buf, off_t offset, size_t length);
 
+/**
+ * @brief Synchronous discard operation
+ * @param fd file descriptor
+ * @param offset discard offset
+ * @param length discard length
+ * @return On success, return 0.
+ *         On error, returns a negative value.
+ */
+int Discard(int fd, off_t offset, size_t length);
+
 /**
  * 异步模式读
  * @param: fd为当前open返回的文件描述符
@@ -231,6 +242,14 @@ int AioRead(int fd, CurveAioContext* aioctx);
  */
 int AioWrite(int fd, CurveAioContext* aioctx);
 
+/**
+ * @brief Asynchronous discard operation
+ * @param fd file descriptor
+ * @param aioctx async request context
+ * @return 0 means success, otherwise it means failure
+ */
+int AioDiscard(int fd, CurveAioContext* aioctx);
+
 /**
  * 重命名文件
  * @param: userinfo是用户信息
@@ -481,6 +500,14 @@ class CurveClient {
     virtual int AioWrite(int fd, CurveAioContext* aioctx,
                          UserDataType dataType);
 
+    /**
+     * @brief Async Discard
+     * @param fd file descriptor
+     * @param aioctx async request context
+     * @return return error code, 0(LIBCURVE_ERROR::OK) means success
+     */
+    virtual int AioDiscard(int fd, CurveAioContext* aioctx);
+
     /**
      * 测试使用，设置fileclient
      * @param client 需要设置的fileclient

nebd/src/part2/request_executor_curve.cpp

@@ -152,13 +152,28 @@ int CurveRequestExecutor::GetInfo(
     return 0;
 }
 
-int CurveRequestExecutor::Discard(
-    NebdFileInstance* fd, NebdServerAioContext* aioctx) {
+int CurveRequestExecutor::Discard(NebdFileInstance* fd,
+                                  NebdServerAioContext* aioctx) {
+    int curveFd = GetCurveFdFromNebdFileInstance(fd);
+    if (curveFd < 0) {
+        return -1;
+    }
 
-    aioctx->ret = 0;
-    aioctx->cb(aioctx);
+    CurveAioCombineContext* curveCombineCtx = new CurveAioCombineContext();
+    curveCombineCtx->nebdCtx = aioctx;
+    int ret = FromNebdCtxToCurveCtx(aioctx, &curveCombineCtx->curveCtx);
+    if (ret < 0) {
+        delete curveCombineCtx;
+        return -1;
+    }
 
-    return 0;
+    ret = client_->AioDiscard(curveFd, &curveCombineCtx->curveCtx);
+    if (ret == LIBCURVE_ERROR::OK) {
+        return 0;
+    }
+
+    delete curveCombineCtx;
+    return -1;
 }
 
 int CurveRequestExecutor::AioRead(
@@ -271,7 +286,9 @@ int CurveRequestExecutor::FromNebdOpToCurveOp(LIBAIO_OP op, LIBCURVE_OP *out) {
     case LIBAIO_OP::LIBAIO_OP_WRITE:
         *out = LIBCURVE_OP_WRITE;
         return 0;
-
+    case LIBAIO_OP::LIBAIO_OP_DISCARD:
+        *out = LIBCURVE_OP_DISCARD;
+        return 0;
     default:
         return -1;
     }

nebd/test/part2/mock_curve_client.h

@@ -45,6 +45,7 @@ class MockCurveClient : public ::curve::client::CurveClient {
                  int(int, CurveAioContext*, curve::client::UserDataType));
     MOCK_METHOD3(AioWrite,
                  int(int, CurveAioContext*, curve::client::UserDataType));
+    MOCK_METHOD2(AioDiscard, int(int, CurveAioContext*));
 };
 
 }  // namespace server

nebd/test/part2/test_request_executor_curve.cpp

@@ -358,20 +358,58 @@ TEST_F(TestReuqestExecutorCurve, test_AioWrite) {
 
 TEST_F(TestReuqestExecutorCurve, test_Discard) {
     auto executor = CurveRequestExecutor::GetInstance();
+    NebdServerAioContext aioctx;
+    aioctx.cb = NebdUnitTestCallback;
     std::string curveFilename("/cinder/volume-1234_cinder_");
-    std::unique_ptr<CurveFileInstance> curveFileIns(new CurveFileInstance());
-    NebdServerAioContext* aioctx = new NebdServerAioContext();
-    nebd::client::DiscardResponse response;
-    TestReuqestExecutorCurveClosure done;
 
-    aioctx->op = LIBAIO_OP::LIBAIO_OP_DISCARD;
-    aioctx->cb = NebdFileServiceCallback;
-    aioctx->response = &response;
-    aioctx->done = &done;
+    // 1. not an curve volume
+    {
+        std::unique_ptr<NebdFileInstance> nebdFileIns(new NebdFileInstance());
+        EXPECT_CALL(*curveClient_, AioDiscard(_, _))
+            .Times(0);
+        ASSERT_EQ(-1, executor.Discard(nebdFileIns.get(), &aioctx));
+    }
 
-    ASSERT_EQ(0, executor.Discard(curveFileIns.get(), aioctx));
-    ASSERT_TRUE(done.IsRunned());
-    ASSERT_EQ(response.retcode(), nebd::client::RetCode::kOK);
+    // 2. fd is invalid
+    {
+        std::unique_ptr<CurveFileInstance> curveFileIns(
+            new CurveFileInstance());
+        curveFileIns->fd = -1;
+        EXPECT_CALL(*curveClient_, AioDiscard(_, _))
+            .Times(0);
+        ASSERT_EQ(-1, executor.Discard(curveFileIns.get(), &aioctx));
+    }
+
+    // 3. curve client return failed
+    {
+        std::unique_ptr<CurveFileInstance> curveFileIns(
+            new CurveFileInstance());
+        aioctx.size = 1;
+        aioctx.offset = 0;
+        aioctx.op = LIBAIO_OP::LIBAIO_OP_DISCARD;
+        curveFileIns->fd = 1;
+        curveFileIns->fileName = curveFilename;
+        EXPECT_CALL(*curveClient_, AioDiscard(_, _))
+            .WillOnce(Return(LIBCURVE_ERROR::FAILED));
+        ASSERT_EQ(-1, executor.Discard(curveFileIns.get(), &aioctx));
+    }
+
+    // 4. ok
+    {
+        std::unique_ptr<CurveFileInstance> curveFileIns(
+            new CurveFileInstance());
+        aioctx.size = 1;
+        aioctx.offset = 0;
+        aioctx.op = LIBAIO_OP::LIBAIO_OP_DISCARD;
+        curveFileIns->fd = 1;
+        curveFileIns->fileName = curveFilename;
+        CurveAioContext* curveCtx;
+        EXPECT_CALL(*curveClient_, AioDiscard(_, _))
+            .WillOnce(DoAll(SaveArg<1>(&curveCtx),
+                            Return(LIBCURVE_ERROR::OK)));
+        ASSERT_EQ(0, executor.Discard(curveFileIns.get(), &aioctx));
+        curveCtx->cb(curveCtx);
+    }
 }
 
 TEST_F(TestReuqestExecutorCurve, test_Flush) {

proto/nameserver2.proto

@@ -155,6 +155,11 @@ message PageFileSegment {
     repeated  PageFileChunkInfo chunks = 5;
 }
 
+message DiscardSegmentInfo {
+    required FileInfo fileInfo = 1;
+    required PageFileSegment pageFileSegment = 2;
+}
+
 message CreateFileRequest {
     required string     fileName = 1;
     required FileType   fileType = 3;
@@ -211,6 +216,18 @@ message GetOrAllocateSegmentResponse {
     optional PageFileSegment pageFileSegment = 2;
 }
 
+message DeAllocateSegmentRequest {
+    required string fileName = 1;
+    required string owner = 2;
+    required uint64 offset = 3;
+    optional string signature = 4;
+    required uint64 date = 5;
+}
+
+message DeAllocateSegmentResponse {
+    required StatusCode statusCode = 1;
+}
+
 message RenameFileRequest {
     required string     oldFileName = 1;
     required string     newFileName = 2;
@@ -496,6 +513,7 @@ service CurveFSService {
     rpc     GetFileInfo(GetFileInfoRequest) returns (GetFileInfoResponse);
     rpc     GetOrAllocateSegment(GetOrAllocateSegmentRequest)
                 returns (GetOrAllocateSegmentResponse);
+    rpc     DeAllocateSegment(DeAllocateSegmentRequest) returns (DeAllocateSegmentResponse);
     rpc     RenameFile(RenameFileRequest) returns (RenameFileResponse);
     rpc     ExtendFile(ExtendFileRequest) returns (ExtendFileResponse);
     rpc     ChangeOwner(ChangeOwnerRequest) returns (ChangeOwnerResponse);

src/client/client_common.h

@@ -41,12 +41,17 @@ using CopysetID     = uint32_t;
 using LogicPoolID   = uint32_t;
 using ChunkServerID = uint32_t;
 using ChunkIndex    = uint32_t;
+using SegmentIndex  = uint32_t;
 
 using EndPoint  = butil::EndPoint;
 using Status    = butil::Status;
 
 using IOManagerID = uint64_t;
 
+constexpr uint64_t KiB = 1024;
+constexpr uint64_t MiB = 1024 * KiB;
+constexpr uint64_t GiB = 1024 * MiB;
+
 // 操作类型
 enum class OpType {
     READ = 0,
@@ -56,6 +61,7 @@ enum class OpType {
     CREATE_CLONE,
     RECOVER_CHUNK,
     GET_CHUNK_INFO,
+    DISCARD,
     UNKNOWN
 };
 
@@ -218,6 +224,8 @@ inline const char* OpTypeToString(OpType optype) {
         return "RecoverChunk";
     case OpType::GET_CHUNK_INFO:
         return "GetChunkInfo";
+    case OpType::DISCARD:
+        return "Discard";
     case OpType::UNKNOWN:
     default:
         return "Unknown";

src/client/client_config.cpp

@@ -39,6 +39,9 @@ namespace curve {
 namespace client {
 int ClientConfig::Init(const char* configpath) {
     conf_.SetConfigPath(configpath);
+
+    LOG(INFO) << "Init config from " << configpath;
+
     if (!conf_.LoadConfig()) {
         LOG(ERROR) << "Load config failed, config path = " << configpath;
         return -1;
@@ -232,6 +235,25 @@ int ClientConfig::Init(const char* configpath) {
         << "config no closefd.timeInterval info, using default value "
         << fileServiceOption_.ioOpt.closeFdThreadOption.fdCloseTimeInterval;
 
+    ret = conf_.GetBoolValue(
+        "discard.enableDiscard",
+        &fileServiceOption_.ioOpt.discardOption.enableDiscard);
+    LOG_IF(ERROR, ret == false)
+        << "config no discard.enableDiscard info";
+    RETURN_IF_FALSE(ret);
+
+    ret = conf_.GetUInt32Value("discard.discardGranularity",
+        &fileServiceOption_.ioOpt.metaCacheOpt.discardGranularity);
+    LOG_IF(ERROR, ret == false)
+        << "config no discard.discardGranularity info";
+    RETURN_IF_FALSE(ret);
+
+    ret = conf_.GetUInt32Value("discard.discardTaskDelayMs",
+        &fileServiceOption_.ioOpt.discardOption.discardTaskDelayMs);
+    LOG_IF(ERROR, ret == false)
+        << "config no discard.discardTaskDelayMs info";
+    RETURN_IF_FALSE(ret);
+
     return 0;
 }
 

src/client/client_metric.h

@@ -155,6 +155,8 @@ struct MDSClientMetric {
     InterfaceMetric getServerList;
     // GetOrAllocateSegment接口统计信息
     InterfaceMetric getOrAllocateSegment;
+    // DeAllocateSegment接口统计信息
+    InterfaceMetric deAllocateSegment;
     // RenameFile接口统计信息
     InterfaceMetric renameFile;
     // Extend接口统计信息
@@ -185,6 +187,7 @@ struct MDSClientMetric {
           refreshSession(prefix, "refreshSession"),
           getServerList(prefix, "getServerList"),
           getOrAllocateSegment(prefix, "getOrAllocateSegment"),
+          deAllocateSegment(prefix, "deAllocateSegment"),
           renameFile(prefix, "renameFile"),
           extendFile(prefix, "extendFile"),
           deleteFile(prefix, "deleteFile"),