replica_server: reimplement uniq_timestamp generator

include/dsn/tool-api/uniq_timestamp_us.h

@@ -0,0 +1,59 @@
+/*
+ * The MIT License (MIT)
+ *
+ * Copyright (c) 2015 Microsoft Corporation
+ *
+ * -=- Robust Distributed System Nucleus (rDSN) -=-
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ * THE SOFTWARE.
+ */
+#pragma once
+
+#include <cstdint>
+#include <algorithm>
+#include <dsn/utility/ports.h>
+#include <dsn/c/api_layer1.h>
+
+namespace dsn {
+//
+// uniq_timestamp_us is used to generate an increasing unique microsecond timestamp
+// in rdsn, it's mainly used for replica to set mutation's timestamp
+//
+// Notice: this module is not thread-safe, 
+// please ensure that it is accessed only by one thread
+//
+class uniq_timestamp_us {
+private:
+    uint64_t _last_ts;
+public:
+    uniq_timestamp_us() { _last_ts = dsn_now_us(); }
+
+    void try_update(uint64_t new_ts)
+    {
+        if ( dsn_likely(new_ts > _last_ts) )
+            _last_ts = new_ts;
+    }
+
+    uint64_t next()
+    {
+        _last_ts = std::max(dsn_now_us(), _last_ts+1);
+        return _last_ts;
+    }
+};
+}

src/dist/replication/lib/replica.h

@@ -42,6 +42,7 @@
 // which is binded to this replication partition
 //
 
+#include <dsn/tool-api/uniq_timestamp_us.h>
 #include <dsn/cpp/serverlet.h>
 #include <dsn/cpp/perf_counter_wrapper.h>
 #include "dist/replication/client_lib/replication_common.h"
@@ -329,6 +330,19 @@ class replica : public serverlet<replica>, public ref_counter
     const app_info _app_info;
     std::map<std::string, std::string> _extra_envs;
 
+    // uniq timestamp generator for this replica.
+    //
+    // we use it to generate an increasing timestamp for current replica
+    // and replicate it to secondary in preparing mutations, and secodaries'
+    // timestamp value will also updated if value from primary is larger
+    //
+    // as the timestamp is recorded in mutation log with mutations, we also update the value
+    // when do replaying
+    //
+    // in addition, as a replica can only be accessed by one thread,
+    // so the "thread-unsafe" generator works fine
+    uniq_timestamp_us _uniq_timestamp_us;
+
     // replica status specific states
     primary_context _primary_states;
     secondary_context _secondary_states;

src/dist/replication/lib/replica_2pc.cpp

@@ -42,16 +42,6 @@
 namespace dsn {
 namespace replication {
 
-static int64_t get_uniq_timestamp()
-{
-    static int64_t last = 0;
-    static ::dsn::utils::ex_lock_nr_spin _lock;
-    int64_t time = dsn_now_ns() / 1000;
-    ::dsn::utils::auto_lock<::dsn::utils::ex_lock_nr_spin> l(_lock);
-    last = std::max(time, last + 1);
-    return last;
-}
-
 void replica::on_client_write(task_code code, dsn_message_t request)
 {
     check_hashed_access();
@@ -93,7 +83,7 @@ void replica::init_prepare(mutation_ptr &mu, bool reconciliation)
         if (_options->prepare_decree_gap_for_debug_logging > 0 &&
             mu->get_decree() % _options->prepare_decree_gap_for_debug_logging == 0)
             level = LOG_LEVEL_DEBUG;
-        mu->set_timestamp(get_uniq_timestamp());
+        mu->set_timestamp(_uniq_timestamp_us.next());
     } else {
         mu->set_id(get_ballot(), mu->data.header.decree);
     }
@@ -335,7 +325,7 @@ void replica::on_prepare(dsn_message_t request)
     }
 
     // real prepare start
-
+    _uniq_timestamp_us.try_update(mu->data.header.timestamp);
     auto mu2 = _prepare_list->get_mutation_by_decree(decree);
     if (mu2 != nullptr && mu2->data.header.ballot == mu->data.header.ballot) {
         if (mu2->is_logged()) {

src/dist/replication/lib/replica_init.cpp

@@ -437,6 +437,7 @@ bool replica::replay_mutation(mutation_ptr &mu, bool is_private)
           mu->data.header.last_committed_decree);
 
     // prepare
+    _uniq_timestamp_us.try_update(mu->data.header.timestamp);
     error_code err = _prepare_list->prepare(mu, partition_status::PS_INACTIVE);
     dassert(err == ERR_OK, "prepare failed, err = %s", err.to_string());