constst pika

Author: wuxianrong

CMakeLists.txt

@@ -457,9 +457,9 @@ set(LZ4_INCLUDE_DIR ${INSTALL_INCLUDEDIR})
 ExternalProject_Add(zlib
   DEPENDS
   URL
-  https://github.com/madler/zlib/releases/download/v1.2.13/zlib-1.2.13.tar.gz
+  https://github.com/madler/zlib/releases/download/v1.3.1/zlib-1.3.1.tar.gz 
   URL_HASH
-  MD5=9b8aa094c4e5765dabf4da391f00d15c
+  MD5=9855b6d802d7fe5b7bd5b196a2271655
   DOWNLOAD_NO_PROGRESS
   1
   UPDATE_COMMAND

conf/pika.conf

@@ -42,7 +42,7 @@ slow-cmd-thread-pool-size : 1
 admin-thread-pool-size : 2
 
 # Slow cmd list e.g. hgetall, mset
-slow-cmd-list :
+slow-cmd-list : 
 
 # List of commands considered as administrative. These commands will be handled by the admin thread pool. Modify this list as needed.
 # Default commands: info, ping, monitor
@@ -95,7 +95,7 @@ proto-max-bulk-len : 512M
 # If <= 0, a proper value is automatically calculated.
 # (usually 1/8 of writer-buffer-size, rounded up to a multiple of 4KB)
 # Supported Units [K|M|G], arena-block-size default unit is in [bytes].
-arena-block-size :
+arena-block-size : 
 
 # Timeout of Pika's connection, counting down starts When there are no requests
 # on a connection (it enters sleep state), when the countdown reaches 0, the connection
@@ -109,12 +109,12 @@ timeout : 60
 # [NOTICE] If this admin password is the same as user password (including both being empty),
 # in this scenario, users are not subject to the restrictions imposed by the userblacklist.
 # PS: "user password" refers to value of the parameter below: userpass.
-requirepass :
+requirepass : 
 
 # Password for replication verify, used for authentication when a slave
 # connects to a master to request replication.
 # [NOTICE] The value of this parameter must match the "requirepass" setting on the master.
-masterauth :
+masterauth : 
 
 # The [password of user], which is empty by default.
 # [NOTICE] If this user password is the same as admin password (including both being empty),
@@ -174,7 +174,7 @@ command-batch-enabled : yes
 
 # The Prefix of dump file's name.
 # All the files that generated by command "bgsave" will be name with this prefix.
-dump-prefix :
+dump-prefix : 
 
 # daemonize  [yes | no].
 #daemonize : yes
@@ -555,13 +555,13 @@ cache-num : 16
 # cache-model 0:cache_none 1:cache_read
 cache-model : 1
 # cache-type: string, set, zset, list, hash, bit
-cache-type: string, set, zset, list, hash, bit
+cache-type : string, set, zset, list, hash, bit
 
 # Set the maximum number of elements in the cache of the Set, list, Zset data types
-cache-value-item-max-size: 1024
+cache-value-item-max-size : 1024
 
 # Sets the maximum number of bytes for Key when the String data type is updated in the cache
-max-key-size-in-cache: 1048576
+max-key-size-in-cache : 1048576
 
 # Maximum number of keys in the zset redis cache
 # On the disk DB, a zset field may have many fields. In the memory cache, we limit the maximum
@@ -593,10 +593,10 @@ cache-maxmemory : 10737418240
 cache-maxmemory-policy : 1
 
 # cache-maxmemory-samples
-cache-maxmemory-samples: 5
+cache-maxmemory-samples : 5
 
 # cache-lfu-decay-time
-cache-lfu-decay-time: 1
+cache-lfu-decay-time : 1
 
 
 # is possible to manage access to Pub/Sub channels with ACL rules as well. The
@@ -650,12 +650,12 @@ cache-lfu-decay-time: 1
 # 'internal-used-unfinished-full-sync' is used to generate a metric 'is_eligible_for_master_election'
 # which serves for the scenario of codis-pika cluster reelection
 # You'd better [DO NOT MODIFY IT UNLESS YOU KNOW WHAT YOU ARE DOING]
-internal-used-unfinished-full-sync :
+internal-used-unfinished-full-sync : 
 
 # for wash data from 4.0.0 to 4.0.1
 # https://github.com/OpenAtomFoundation/pika/issues/2886
 # default value: true
-wash-data: true
+wash-data : true
 
 # Pika automatic compact compact strategy, a complement to rocksdb compact.
 # Trigger the compact background task periodically according to `compact-interval`
@@ -690,3 +690,6 @@ dont-compact-sst-created-in-seconds : 20
 # According to the number of sst files in rocksdb, 
 # compact every `compact-every-num-of-files` file.
 best-delete-min-ratio : 10
+# Generated by ReplicationID CONFIG REWRITE
+replication-id : d605afa1b464ddf3e571966482dd934ec7336a4bac49aa0a6b
+run-id : 58ec490577bb7defd64f6fe642d7609af67896b1

include/pika_binlog.h

@@ -57,6 +57,7 @@ class Binlog : public pstd::noncopyable {
   pstd::Status Put(const std::string& item, LogOffset *cur_logoffset,std::string& binlog);
   pstd::Status IsOpened();
   pstd::Status GetProducerStatus(uint32_t* filenum, uint64_t* pro_offset, uint32_t* term = nullptr, uint64_t* logic_id = nullptr);
+  pstd::WritableFile* GetQueue() { return queue_.get(); }
   /*
    * Set Producer pro_num and pro_offset with lock
    */

include/pika_consensus.h

@@ -288,15 +288,15 @@ class ConsensusCoordinator {
       context_->UpdateAppliedIndex(committed_id_);
     }
     notification_counter_.fetch_add(1);
-    LOG(INFO) << "SetCommittedId: Updated to " << offset.ToString();
+    //LOG(INFO) << "Master SetCommittedId: Updated to " << offset.ToString();
   }
 
  private:
   pstd::Status PersistAppendBinlog(const std::shared_ptr<Cmd>& cmd_ptr, LogOffset& cur_offset);
 
  private:
   std::shared_mutex is_consistency_rwlock_;
-  bool is_consistency_ = false;
+  bool is_consistency_ = true;
   std::shared_mutex committed_id_rwlock_;
   LogOffset committed_id_ = LogOffset();
   std::atomic<uint64_t> notification_counter_{0};

include/pika_rm.h

@@ -33,8 +33,9 @@
 #define kBinlogSendBatchNum 100
 
 // unit seconds
-#define kSendKeepAliveTimeout (2 * 1000000)
-#define kRecvKeepAliveTimeout (20 * 1000000)
+// WXR
+#define kSendKeepAliveTimeout (100 * 1000000)
+#define kRecvKeepAliveTimeout (200 * 1000000)
 
 
 class SyncDB {
@@ -302,7 +303,7 @@ class PikaReplicaManager {
 
   // Last committed ID for RocksDB thread processing
   LogOffset last_committed_id_;
-  std::mutex last_committed_id_mutex_;
+  std::shared_mutex last_committed_id_mutex_;
 
   // Background thread processing methods
   void StartCommandQueueThread();

src/net/src/bg_thread.cc

@@ -6,6 +6,7 @@
 #include "net/include/bg_thread.h"
 #include <cstdlib>
 #include <mutex>
+#include <glog/logging.h>
 
 namespace net {
 
@@ -79,7 +80,7 @@ void* BGThread::ThreadMain() {
   while (!should_stop()) {
     std::unique_lock lock(mu_);
 
-    rsignal_.wait(lock, [this]() { return !queue_.empty() || !timer_queue_.empty() || should_stop(); });
+    rsignal_.wait(lock, [this]() { return !queue_.empty() || should_stop();  });
 
     if (should_stop()) {
       break;

src/net/src/pb_conn.cc

@@ -42,7 +42,10 @@ ReadStatus PbConn::GetRequest() {
       case kHeader: {
         int quickack = 1;
         ssize_t nread = read(fd(), rbuf_ + cur_pos_, COMMAND_HEADER_LENGTH - cur_pos_);
-        setsockopt(fd(), IPPROTO_TCP, TCP_QUICKACK, &quickack, sizeof(quickack));
+       #ifdef __linux__
+    int quickack = 1;
+    setsockopt(fd(), IPPROTO_TCP, TCP_QUICKACK, &quickack, sizeof(quickack));
+#endif 
         if (nread == -1) {
           if (errno == EAGAIN) {
             return kReadHalf;
@@ -82,7 +85,10 @@ ReadStatus PbConn::GetRequest() {
         // read msg body
         ssize_t nread = read(fd(), rbuf_ + cur_pos_, remain_packet_len_);
         int quickack = 1;
-        setsockopt(fd(), IPPROTO_TCP, TCP_QUICKACK, &quickack, sizeof(quickack));
+        #ifdef __linux__
+    int quickack = 1;
+    setsockopt(fd(), IPPROTO_TCP, TCP_QUICKACK, &quickack, sizeof(quickack));
+#endif
         if (nread == -1) {
           if (errno == EAGAIN) {
             return kReadHalf;
@@ -131,7 +137,7 @@ WriteStatus PbConn::SendReply() {
     while (item_len - write_buf_.item_pos_ > 0) {
       nwritten = write(fd(), item.data() + write_buf_.item_pos_, item_len - write_buf_.item_pos_);
       if (nwritten <= 0) {
-        LOG(ERROR) << "nwritten less than 0";
+        //LOG(ERROR) << "nwritten less than 0";
         break;
       }
       g_network_statistic->IncrReplOutputBytes(nwritten);

src/pika_auxiliary_thread.cc

@@ -35,11 +35,13 @@ void* PikaAuxiliaryThread::ThreadMain() {
     g_pika_server->CheckLeaderProtectedMode();
 
     // TODO(whoiami) timeout
+    // 将增量数据写入 Write_queue_
     s = g_pika_server->TriggerSendBinlogSync();
     if (!s.ok()) {
       LOG(WARNING) << s.ToString();
     }
     // send to peer
+    // 将 Write_queue 中的数据发送给从节点
     int res = g_pika_server->SendToPeer();
     if (res == 0) {
       // sleep 100 ms

src/pika_binlog.cc

@@ -298,7 +298,9 @@ Status Binlog::EmitPhysicalRecord(RecordType t, const char* ptr, size_t n, int*
   if (s.ok()) {
     s = queue_->Append(pstd::Slice(ptr, n));
     if (s.ok()) {
-      s = queue_->Flush();
+      //LOG(INFO) << "EmitPhysicalRecord Flush";
+      //s = queue_->Sync();
+      //s = queue_->Flush();
     }
   }
   block_offset_ += static_cast<int32_t>(kHeaderSize + n);

src/pika_client_conn.cc

@@ -224,12 +224,13 @@ std::shared_ptr<Cmd> PikaClientConn::DoCmd(const PikaCmdArgsType& argv, const st
   if (c_ptr->is_write() && g_pika_conf->command_batch_enabled()) {
     // Get the appropriate SyncMasterDB for command batching
     auto sync_db = g_pika_rm->GetSyncMasterDBByName(DBInfo(c_ptr->db_name()));
+    // 查看 DB 是不是 Master DB
     if (sync_db) {
       auto command_collector = sync_db->GetCommandCollector();
       if (command_collector) {
         // Create callback to handle command completion
         auto callback = [this, c_ptr](const LogOffset& offset, pstd::Status status) {
-            LOG(INFO) << "Command completed";
+            //LOG(INFO) << "Command completed";
             auto pc = dynamic_cast<PikaClientConn*>(c_ptr->GetConn().get());
             if (pc) {
                 auto resp_ptr = c_ptr->GetResp();