[GitHub] [incubator-kvrocks] caipengbo commented on a diff in pull request #1197: Add support of stream migration during slot migration process

[GitBox <gi...@apache.org>]

caipengbo commented on code in PR #1197:
URL: https://github.com/apache/incubator-kvrocks/pull/1197#discussion_r1053896272


##########
src/cluster/slot_migrate.h:
##########
@@ -77,89 +83,92 @@ struct SlotMigrateJob {
 
 class SlotMigrate : public Redis::Database {
  public:
-  explicit SlotMigrate(Server *svr, int speed = kMigrateSpeed, int pipeline_size = kPipelineSize,
-                       int seq_gap = kSeqGapLimit);
+  explicit SlotMigrate(Server *svr, int migration_speed = kDefaultMigrationSpeed,
+                       int pipeline_size_limit = kDefaultPipelineSizeLimit, int seq_gap = kDefaultSeqGapLimit);
+  SlotMigrate(const SlotMigrate &other) = delete;
+  SlotMigrate &operator=(const SlotMigrate &other) = delete;
   ~SlotMigrate();
 
-  Status CreateMigrateHandleThread(void);
+  Status CreateMigrateHandleThread();
   void Loop();
   Status MigrateStart(Server *svr, const std::string &node_id, const std::string &dst_ip, int dst_port, int slot,
                       int speed, int pipeline_size, int seq_gap);
   void ReleaseForbiddenSlot();
   void SetMigrateSpeedLimit(int speed) {
-    if (speed >= 0) migrate_speed_ = speed;
+    if (speed >= 0) migration_speed_ = speed;
   }
-  void SetPipelineSize(uint32_t size) {
-    if (size > 0) pipeline_size_limit_ = size;
+  void SetPipelineSize(int value) {
+    if (value > 0) pipeline_size_limit_ = value;
   }
   void SetSequenceGapSize(int size) {
     if (size > 0) seq_gap_limit_ = size;
   }
   void SetMigrateStopFlag(bool state) { stop_migrate_ = state; }
-  int16_t GetMigrateState() { return migrate_state_; }
-  int16_t GetMigrateStateMachine() { return state_machine_; }
-  int16_t GetForbiddenSlot(void) { return forbidden_slot_; }
-  int16_t GetMigratingSlot(void) { return migrate_slot_; }
-  void GetMigrateInfo(std::string *info);
+  bool IsMigrationInProgress() const { return migrate_state_ == kMigrateStarted; }
+  int16_t GetMigrateStateMachine() const { return state_machine_; }
+  int16_t GetForbiddenSlot() const { return forbidden_slot_; }
+  int16_t GetMigratingSlot() const { return migrate_slot_; }
+  void GetMigrateInfo(std::string *info) const;
   bool IsTerminated() { return thread_state_ == ThreadState::Terminated; }
 
  private:
-  void StateMachine(void);
-  Status Start(void);
-  Status SendSnapshot(void);
-  Status SyncWal(void);
-  Status Success(void);
-  Status Fail(void);
-  Status Clean(void);
+  void StateMachine();
+  Status Start();
+  Status SendSnapshot();
+  Status SyncWal();
+  Status Success();
+  Status Fail();
+  Status Clean();
 
   bool AuthDstServer(int sock_fd, const std::string &password);
   bool SetDstImportStatus(int sock_fd, int status);
   bool CheckResponseOnce(int sock_fd);
   bool CheckResponseWithCounts(int sock_fd, int total);
 
-  Status MigrateOneKey(const rocksdb::Slice &key, const rocksdb::Slice &value, std::string *restore_cmds);
+  KeyMigrationResult MigrateOneKey(const rocksdb::Slice &key, const rocksdb::Slice &encoded_metadata,
+                                   std::string *restore_cmds);
   bool MigrateSimpleKey(const rocksdb::Slice &key, const Metadata &metadata, const std::string &bytes,
                         std::string *restore_cmds);
   bool MigrateComplexKey(const rocksdb::Slice &key, const Metadata &metadata, std::string *restore_cmds);
+  bool MigrateStream(const rocksdb::Slice &key, const StreamMetadata &metadata, std::string *restore_cmds);

Review Comment:
   By the way, this type of function uses `bool` as its return value. I think it's a more elegant way to return `Status`, and move the LOG information from these internal functions to `Status` Msg, and return it to the upper level.
   
   



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@kvrocks.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org