This repository has been archived by the owner on Jun 23, 2022. It is now read-only.
feat(dup): add interface mutation_duplicator & duplication procedure #304
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
This PR added interface
mutation_duplicator
and pipeline of duplication procedure.mutation_duplicator
This is the interface of data "sender" to be extensible for multiple endpoint types.
For example, if the remote endpoint is a pegasus cluster, the
pegasus_mutation_duplicator
willtransfer client RPCs to the remote pegasus.
If there's a further requirement to duplicate data to HBase/MySQL/etc., we can implement this interface.
Duplication Pipeline
The other part of this PR is simple. In each round of duplication, a batch of mutations is loaded
from private-log, and is sent via
mutation_duplicator
to remote endpoint.So the pipeline can be declared like the following (in bash convention):
Each procedure in our implementation is declared as a class:
load_from_private_log
, run in thread pool,REPLICATION_LONG
ship_mutations
, run in thread pool,REPLICATION
In our real implementation, we make this design slightly complicated:
load_mutations
runs in thread poolREPLICATION
, because we need to obtain the latest information of this replica. Leave to future refactoring if we find this unnecessary and we can delete this class.replica_duplicator
replica_duplicator
is the basement of this pipeline, so it extendspipeline::base
.thread_pool(LPC_REPLICATION_LOW).task_tracker(tracker()).thread_hash(get_gpid().thread_hash());
It means that the pipeline is running in the environment of:
LPC_REPLICATION_LOW
: in low priority of thread poolREPLICATION
.