feat(duplication): add a new option to configure the duration of the …

…delay until the next execution if there is no mutation to be loaded (#2099)

A new option is added for duplication to configure the duration of the delay
until the next execution if there is no mutation to be loaded:

```diff
[replication]
+ dup_no_mutation_load_delay_ms = 100
```

src/replica/duplication/duplication_pipeline.cpp

@@ -17,22 +17,30 @@
 
 #include "duplication_pipeline.h"
 
-#include <string_view>
-#include <stddef.h>
 #include <algorithm>
+#include <chrono>
+#include <cstddef>
 #include <functional>
 #include <string>
+#include <string_view>
 #include <utility>
 
 #include "load_from_private_log.h"
 #include "replica/duplication/replica_duplicator.h"
 #include "replica/mutation_log.h"
 #include "replica/replica.h"
-#include "runtime/rpc/rpc_holder.h"
 #include "utils/autoref_ptr.h"
 #include "utils/errors.h"
+#include "utils/flags.h"
 #include "utils/fmt_logging.h"
 
+DSN_DEFINE_uint64(
+    replication,
+    dup_no_mutation_load_delay_ms,
+    100,
+    "The duration of the delay until the next execution if there is no mutation to be loaded.");
+DSN_TAG_VARIABLE(dup_no_mutation_load_delay_ms, FT_MUTABLE);
+
 METRIC_DEFINE_counter(replica,
                       dup_shipped_bytes,
                       dsn::metric_unit::kBytes,
@@ -63,8 +71,8 @@ void load_mutation::run()
     const auto max_plog_committed_decree =
         std::min(_replica->private_log()->max_decree_on_disk(), _replica->last_applied_decree());
     if (_start_decree > max_plog_committed_decree) {
-        // wait 100ms for next try if no mutation was added.
-        repeat(100_ms);
+        // Wait for a while if no mutation was added.
+        repeat(std::chrono::milliseconds(FLAGS_dup_no_mutation_load_delay_ms));
         return;
     }