split: parent replica create child replica

include/dsn/dist/replication/replication.codes.h

@@ -67,6 +67,7 @@ MAKE_EVENT_CODE_AIO(LPC_WRITE_REPLICATION_LOG_PRIVATE, TASK_PRIORITY_HIGH)
 MAKE_EVENT_CODE_AIO(LPC_WRITE_REPLICATION_LOG_SHARED, TASK_PRIORITY_HIGH)
 MAKE_EVENT_CODE(LPC_QUERY_CONFIGURATION_ALL, TASK_PRIORITY_HIGH)
 MAKE_EVENT_CODE(LPC_MEM_RELEASE, TASK_PRIORITY_COMMON)
+MAKE_EVENT_CODE(LPC_CREATE_CHILD, TASK_PRIORITY_COMMON)
 #undef CURRENT_THREAD_POOL
 
 // THREAD_POOL_META_SERVER
@@ -142,6 +143,8 @@ MAKE_EVENT_CODE_AIO(LPC_REPLICA_COPY_LAST_CHECKPOINT_DONE, TASK_PRIORITY_COMMON)
 MAKE_EVENT_CODE_RPC(RPC_COLD_BACKUP, TASK_PRIORITY_COMMON)
 MAKE_EVENT_CODE(LPC_REPLICATION_COLD_BACKUP, TASK_PRIORITY_COMMON)
 MAKE_EVENT_CODE(LPC_EXEC_COMMAND_ON_REPLICA, TASK_PRIORITY_LOW)
+MAKE_EVENT_CODE(LPC_PARTITION_SPLIT, TASK_PRIORITY_LOW)
+MAKE_EVENT_CODE(LPC_PARTITION_SPLIT_ERROR, TASK_PRIORITY_COMMON)
 MAKE_EVENT_CODE(LPC_REPLICATION_LOW, TASK_PRIORITY_LOW)
 MAKE_EVENT_CODE(LPC_REPLICATION_COMMON, TASK_PRIORITY_COMMON)
 MAKE_EVENT_CODE(LPC_REPLICATION_HIGH, TASK_PRIORITY_HIGH)

src/dist/replication/lib/CMakeLists.txt

@@ -24,6 +24,7 @@ set(MY_PROJ_LIBS
     dsn_nfs
     dsn_cli
     dsn_http
+    dsn_runtime
     )
 
 # Extra files that will be installed

src/dist/replication/lib/duplication/test/CMakeLists.txt

@@ -22,7 +22,7 @@ set(MY_PROJ_LIBS dsn_meta_server
         fmt
 )
 
-set(MY_BOOST_LIBS Boost::system Boost::filesystem)
+set(MY_BOOST_LIBS Boost::system Boost::filesystem Boost::regex)
 
 set(MY_BINPLACES
         config-test.ini

src/dist/replication/lib/replica.h

@@ -76,8 +76,12 @@ class replica : public serverlet<replica>, public ref_counter, public replica_ba
     //    routines for replica stub
     //
     static replica *load(replica_stub *stub, const char *dir);
-    static replica *
-    newr(replica_stub *stub, gpid gpid, const app_info &app, bool restore_if_necessary);
+    // {parent_dir} is used in partition split for get_child_dir in replica_stub
+    static replica *newr(replica_stub *stub,
+                         gpid gpid,
+                         const app_info &app,
+                         bool restore_if_necessary,
+                         const std::string &parent_dir = "");
 
     // return true when the mutation is valid for the current replica
     bool replay_mutation(mutation_ptr &mu, bool is_private);
@@ -309,6 +313,17 @@ class replica : public serverlet<replica>, public ref_counter, public replica_ba
 
     std::string query_compact_state() const;
 
+    /////////////////////////////////////////////////////////////////
+    // partition split
+    // parent partition create child
+    void on_add_child(const group_check_request &request);
+
+    // child replica initialize config and state info
+    void init_child_replica(gpid parent_gpid, dsn::rpc_address primary_address, ballot init_ballot);
+
+    // parent reset child information when partition split failed
+    void clean_up_parent_split_context();
+
 private:
     friend class ::dsn::replication::replication_checker;
     friend class ::dsn::replication::test::test_checker;
@@ -318,6 +333,7 @@ class replica : public serverlet<replica>, public ref_counter, public replica_ba
     friend class replica_learn_test;
     friend class replica_duplicator_manager;
     friend class load_mutation;
+    friend class replica_split_test;
 
     // replica configuration, updated by update_local_configuration ONLY
     replica_configuration _config;
@@ -364,6 +380,7 @@ class replica : public serverlet<replica>, public ref_counter, public replica_ba
     potential_secondary_context _potential_secondary_states;
     // policy_name --> cold_backup_context
     std::map<std::string, cold_backup_context_ptr> _cold_backup_contexts;
+    partition_split_context _split_states;
 
     // timer task that running in replication-thread
     dsn::task_ptr _collect_info_timer;
@@ -391,6 +408,14 @@ class replica : public serverlet<replica>, public ref_counter, public replica_ba
     // duplication
     std::unique_ptr<replica_duplicator_manager> _duplication_mgr;
 
+    // partition split
+    // _child_gpid = gpid({app_id},{pidx}+{old_partition_count}) for parent partition
+    // _child_gpid.app_id = 0 if parent partition not during partition split and child partition
+    dsn::gpid _child_gpid{0, 0};
+    // ballot when starting partition split coz split will stop if ballot changed
+    // _child_init_ballot = 0 if partition not during partition split
+    ballot _child_init_ballot{0};
+
     // perf counters
     perf_counter_wrapper _counter_private_log_size;
     perf_counter_wrapper _counter_recent_write_throttling_delay_count;

src/dist/replication/lib/replica_context.cpp

@@ -1287,5 +1287,12 @@ void cold_backup_context::file_upload_complete(const std::string &filename)
     _cur_upload_file_cnt -= 1;
     _file_status[filename] = file_status::FileUploadComplete;
 }
+
+bool partition_split_context::cleanup(bool force)
+{
+    parent_gpid.set_app_id(0);
+    return true;
 }
-} // end namespace
+
+} // namespace replication
+} // namespace dsn

src/dist/replication/lib/replica_context.h

@@ -523,6 +523,17 @@ class cold_backup_context : public ref_counter
 
 typedef dsn::ref_ptr<cold_backup_context> cold_backup_context_ptr;
 
+class partition_split_context
+{
+public:
+    partition_split_context() {}
+    // TODO(heyuchen): force will be used in further pull request
+    bool cleanup(bool force);
+
+public:
+    gpid parent_gpid;
+};
+
 //---------------inline impl----------------------------------------------------------------
 
 inline partition_status::type primary_context::get_node_status(::dsn::rpc_address addr) const

src/dist/replication/lib/replica_init.cpp

@@ -74,10 +74,18 @@ error_code replica::initialize_on_new()
     return init_app_and_prepare_list(true);
 }
 
-/*static*/ replica *
-replica::newr(replica_stub *stub, gpid gpid, const app_info &app, bool restore_if_necessary)
+/*static*/ replica *replica::newr(replica_stub *stub,
+                                  gpid gpid,
+                                  const app_info &app,
+                                  bool restore_if_necessary,
+                                  const std::string &parent_dir)
 {
-    std::string dir = stub->get_replica_dir(app.app_type.c_str(), gpid);
+    std::string dir;
+    if (parent_dir.empty()) {
+        dir = stub->get_replica_dir(app.app_type.c_str(), gpid);
+    } else {
+        dir = stub->get_child_dir(app.app_type.c_str(), gpid, parent_dir);
+    }
     replica *rep = new replica(stub, gpid, app, dir.c_str(), restore_if_necessary);
     error_code err;
     if (restore_if_necessary && (err = rep->restore_checkpoint()) != dsn::ERR_OK) {