feat: add defragment command

src/server/engine_shard_set.cc

@@ -241,7 +241,8 @@ void EngineShard::DefragTaskState::ResetScanState() {
 // 3. in case the above is OK, make sure that we have a "gap" between usage and commited memory
 // (control by mem_defrag_waste_threshold flag)
 bool EngineShard::DefragTaskState::CheckRequired() {
-  if (cursor > kCursorDoneState || underutilized_found) {
+  if (is_force_defrag || cursor > kCursorDoneState || underutilized_found) {
+    is_force_defrag = false;
     VLOG(2) << "cursor: " << cursor << " and underutilized_found " << underutilized_found;
     return true;
   }
@@ -265,6 +266,10 @@ bool EngineShard::DefragTaskState::CheckRequired() {
   return false;
 }
 
+void EngineShard::ForceDefrag() {
+  defrag_state_.is_force_defrag = true;
+}
+
 bool EngineShard::DoDefrag() {
   // --------------------------------------------------------------------------
   // NOTE: This task is running with exclusive access to the shard.

src/server/engine_shard_set.h

@@ -183,11 +183,14 @@ class EngineShard {
 
   TxQueueInfo AnalyzeTxQueue() const;
 
+  void ForceDefrag();
+
  private:
   struct DefragTaskState {
     size_t dbid = 0u;
     uint64_t cursor = 0u;
     bool underutilized_found = false;
+    bool is_force_defrag = false;
 
     // check the current threshold and return true if
     // we need to do the defragmentation

src/server/memory_cmd.cc

@@ -154,6 +154,14 @@ void MemoryCmd::Run(CmdArgList args) {
     return Track(args);
   }
 
+  if (sub_cmd == "DEFRAGMENT") {
+    shard_set->pool()->DispatchOnAll([this](util::ProactorBase*) {
+      if (auto* shard = EngineShard::tlocal(); shard)
+        shard->ForceDefrag();
+    });
+    return cntx_->SendSimpleString("OK");
+  }
+
   string err = UnknownSubCmd(sub_cmd, "MEMORY");
   return cntx_->SendError(err, kSyntaxErrType);
 }