mds/client: support discard

conf/client.conf

@@ -156,3 +156,11 @@ global.turnOffHealthCheck=true
 ### throttle config
 #
 throttle.enable=false
+
+##### discard configurations #####
+# enable/disable discard
+discard.enable=true
+# discard granularity
+discard.granularity=4096
+# discard cleanup task delay times in millisecond
+discard.taskDelayMs=60000

conf/cs_client.conf

@@ -151,3 +151,11 @@ global.metricDummyServerStartPort=9000
 # session map文件，存储打开文件的filename到path的映射
 #
 global.sessionMapPath=./session_map.json
+
+##### discard configurations #####
+# enable/disable discard
+discard.enable=false
+# discard granularity
+discard.granularity=4096
+# discard cleanup task delay times in millisecond
+discard.taskDelayMs=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

@@ -145,3 +145,11 @@ global.metricDummyServerStartPort=10000
 # session map文件，存储打开文件的filename到path的映射
 #
 global.sessionMapPath=./session_map.json
+
+##### discard configurations #####
+# enable/disable discard
+discard.enable=false
+# discard granularity
+discard.granularity=4096
+# discard cleanup task delay times in millisecond
+discard.taskDelayMs=60000

conf/snap_client.conf

@@ -151,3 +151,11 @@ global.metricDummyServerStartPort=9000
 # session map文件，存储打开文件的filename到path的映射
 #
 global.sessionMapPath=./session_map.json
+
+##### discard configurations #####
+# enable/disable discard
+discard.enable=false
+# discard granularity
+discard.granularity=4096
+# discard cleanup task delay times in millisecond
+discard.taskDelayMs=60000

curve-ansible/roles/generate_config/defaults/main.yml

@@ -30,6 +30,7 @@ mds_etcd_operation_timeout_ms: 5000
 mds_etcd_retry_times: 3
 mds_segment_alloc_periodic_persist_inter_ms: 10000
 mds_segment_alloc_retry_inter_ms: 1000
+mds_segment_discard_scan_interval_ms: 5000
 mds_leader_session_inter_sec: 5
 mds_leader_election_timeout_ms: 0
 mds_enable_copyset_scheduler: true
@@ -220,6 +221,9 @@ client_session_map_path: ./session_map.json
 client_closefd_timeout_sec: 300
 client_closefd_time_interval_sec: 600
 client_throttle_enable: false
+client_discard_enable: true
+client_discard_granularity: 4096
+client_discard_task_delay_ms: 60000
 
 # nebd默认配置
 client_config_path: /etc/curve/client.conf

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

@@ -167,3 +167,11 @@ global.sessionMapPath={{ client_session_map_path }}
 ### throttle config
 #
 throttle.enable={{ client_throttle_enable }}
+
+##### discard configurations #####
+# enable/disable discard
+discard.enable={{ client_discard_enable }}
+# discard granularity
+discard.granularity={{ client_discard_granularity }}
+# discard cleanup task delay times in millisecond
+discard.taskDelayMs={{ client_discard_task_delay_ms }}

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

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

include/client/libcurve.h

@@ -113,6 +113,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;
 
@@ -233,6 +234,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返回的文件描述符
@@ -249,6 +260,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是用户信息
@@ -510,6 +529,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,33 @@ 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) {
+        LOG(ERROR) << "Parse curve fd failed";
+        return -1;
+    }
 
-    aioctx->ret = 0;
-    aioctx->cb(aioctx);
+    CurveAioCombineContext* curveCombineCtx = new CurveAioCombineContext();
+    curveCombineCtx->nebdCtx = aioctx;
+    int ret = FromNebdCtxToCurveCtx(aioctx, &curveCombineCtx->curveCtx);
+    if (ret < 0) {
+        LOG(ERROR) << "Convert nebd aio context to curve aio context failed, "
+                      "curve fd: "
+                   << curveFd;
+        delete curveCombineCtx;
+        return -1;
+    }
 
-    return 0;
+    ret = client_->AioDiscard(curveFd, &curveCombineCtx->curveCtx);
+    if (ret == LIBCURVE_ERROR::OK) {
+        return 0;
+    }
+
+    LOG(ERROR) << "Curve client return failed, curve fd: " << curveFd;
+    delete curveCombineCtx;
+    return -1;
 }
 
 int CurveRequestExecutor::AioRead(
@@ -271,7 +291,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

@@ -184,6 +184,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;
@@ -254,6 +259,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;
@@ -569,6 +586,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

@@ -43,12 +43,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,
@@ -58,6 +63,7 @@ enum class OpType {
     CREATE_CLONE,
     RECOVER_CHUNK,
     GET_CHUNK_INFO,
+    DISCARD,
     UNKNOWN
 };
 
@@ -245,6 +251,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;
@@ -251,6 +254,23 @@ int ClientConfig::Init(const char* configpath) {
         << "config no throttle.enable info, using default value "
         << fileServiceOption_.ioOpt.throttleOption.enable;
 
+    ret = conf_.GetBoolValue("discard.enable",
+                             &fileServiceOption_.ioOpt.discardOption.enable);
+    LOG_IF(ERROR, ret == false) << "config no discard.enable info";
+    RETURN_IF_FALSE(ret);
+
+    ret = conf_.GetUInt32Value(
+        "discard.granularity",
+        &fileServiceOption_.ioOpt.metaCacheOpt.discardGranularity);
+    LOG_IF(ERROR, ret == false) << "config no discard.granularity info";
+    RETURN_IF_FALSE(ret);
+
+    ret = conf_.GetUInt32Value(
+        "discard.taskDelayMs",
+        &fileServiceOption_.ioOpt.discardOption.taskDelayMs);
+    LOG_IF(ERROR, ret == false) << "config no discard.taskDelayMs info";
+    RETURN_IF_FALSE(ret);
+
     return 0;
 }