fix bugs and add TODOS

Author: wangshaoyi

src/net/src/client_thread.cc

@@ -356,7 +356,6 @@ void ClientThread::ProcessNotifyEvents(const NetFiredEvent* pfe) {
             NotifyWrite(ip_port);
           }
         } else if (ti.notify_type() == kNotiClose) {
-          LOG(INFO) << "received kNotiClose";
           net_multiplexer_->NetDelEvent(fd, 0);
           CloseFd(fd, ip_port);
           fd_conns_.erase(fd);

src/net/src/holy_thread.cc

@@ -19,7 +19,7 @@ namespace net {
 HolyThread::HolyThread(int port, ConnFactory* conn_factory, int cron_interval, const ServerHandle* handle, bool async)
     : ServerThread::ServerThread(port, cron_interval, handle),
       conn_factory_(conn_factory),
-      
+
       keepalive_timeout_(kDefaultKeepAliveTime),
       async_(async) {}
 
@@ -290,7 +290,6 @@ void HolyThread::ProcessNotifyEvents(const net::NetFiredEvent* pfe) {
         if (ti.notify_type() == net::kNotiWrite) {
           net_multiplexer_->NetModEvent(ti.fd(), 0, kReadable | kWritable);
         } else if (ti.notify_type() == net::kNotiClose) {
-          LOG(INFO) << "receive noti close";
           std::shared_ptr<net::NetConn> conn = get_conn(fd);
           if (!conn) {
             continue;

src/net/src/pb_conn.cc

@@ -3,6 +3,10 @@
 // LICENSE file in the root directory of this source tree. An additional grant
 // of patent rights can be found in the PATENTS file in the same directory.
 
+#include <sys/socket.h>
+#include <netinet/tcp.h>
+#include <netinet/in.h>
+
 #include "net/include/pb_conn.h"
 
 #include <arpa/inet.h>
@@ -20,7 +24,7 @@ namespace net {
 
 PbConn::PbConn(const int fd, const std::string& ip_port, Thread* thread, NetMultiplexer* mpx)
     : NetConn(fd, ip_port, thread, mpx),
-      
+
       write_buf_(0)
       {
   rbuf_ = reinterpret_cast<char*>(malloc(sizeof(char) * PB_IOBUF_LEN));
@@ -36,7 +40,9 @@ ReadStatus PbConn::GetRequest() {
   while (true) {
     switch (connStatus_) {
       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));
         if (nread == -1) {
           if (errno == EAGAIN) {
             return kReadHalf;
@@ -75,6 +81,8 @@ 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));
         if (nread == -1) {
           if (errno == EAGAIN) {
             return kReadHalf;
@@ -123,6 +131,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";
         break;
       }
       g_network_statistic->IncrReplOutputBytes(nwritten);
@@ -144,6 +153,7 @@ WriteStatus PbConn::SendReply() {
     if (item_len - write_buf_.item_pos_ != 0) {
       return kWriteHalf;
     }
+    LOG(ERROR) << "write item success";
   }
   return kWriteAll;
 }

src/pika_command.cc

@@ -874,20 +874,24 @@ void Cmd::InternalProcessCommand(const HintKeys& hint_keys) {
     db_->DBLockShared();
   }
   if(g_pika_server->IsConsistency()){
+    uint64_t before_do_binlog_us = pstd::NowMicros();
     DoBinlog();
+    uint64_t end_us = pstd::NowMicros();
+    this->binlog_duration_ms = (end_us - before_do_binlog_us) / 1000;
     if(res().ok()){
+        uint64_t before_do_command_us = pstd::NowMicros();
         DoCommand(hint_keys);
-    }
-    if (g_pika_conf->slowlog_slower_than() >= 0) {
-        do_duration_ += pstd::NowMicros() - start_us;
+        this->command_duration_ms = (pstd::NowMicros() - before_do_command_us) / 1000;
     }
   }else{
+    uint64_t before_do_command_us = pstd::NowMicros();
     DoCommand(hint_keys);
-    if (g_pika_conf->slowlog_slower_than() >= 0) {
-        do_duration_ += pstd::NowMicros() - start_us;
-    }
+    this->command_duration_ms = (pstd::NowMicros() - before_do_command_us) / 1000;
 
+    uint64_t before_do_binlog_us = pstd::NowMicros();
     DoBinlog();
+    uint64_t end_us = pstd::NowMicros();
+    this->binlog_duration_ms = (end_us - before_do_binlog_us) / 1000;
   }
 
   if (!IsSuspend()) {
@@ -896,9 +900,6 @@ void Cmd::InternalProcessCommand(const HintKeys& hint_keys) {
   if (is_write()) {
     record_lock.Unlock(current_key());
   }
-
-  uint64_t end_us = pstd::NowMicros();
-  this->binlog_duration_ms = (end_us - before_do_binlog_us) / 1000;
 }
 
 void Cmd::DoCommand(const HintKeys& hint_keys) {
@@ -977,7 +978,6 @@ void Cmd::DoBinlog() {
                         << s.ToString();
         res().SetRes(CmdRes::kErrOther, s.ToString());
        }
-
       return;
     }
   }

src/pika_consensus.cc

@@ -65,7 +65,8 @@ Status Context::Init() {
 void Context::UpdateAppliedIndex(const LogOffset& offset) {
   std::lock_guard l(rwlock_);
   LogOffset cur_offset;
-  applied_win_.Update(SyncWinItem(offset), SyncWinItem(offset), &cur_offset);
+  // TODO: 暂时注释掉这一行，因为applied_win_没有push调用，只有update，窗口永远对不上
+  //applied_win_.Update(SyncWinItem(offset), SyncWinItem(offset), &cur_offset);
   if (cur_offset > applied_index_) {
     applied_index_ = cur_offset;
     StableSave();
@@ -811,6 +812,8 @@ bool ConsensusCoordinator::GetISConsistency() {
 }
 
 bool ConsensusCoordinator::checkFinished(const LogOffset& offset) {
+  //TODO: 暂时加了读写锁，后期考虑替换为原子变量
+  std::lock_guard l(committed_id_rwlock_);
   if (offset <= committed_id_) {
     return true;
   }
@@ -836,6 +839,8 @@ Status ConsensusCoordinator::PersistAppendBinlog(const std::shared_ptr<Cmd>& cmd
     return s;
   }
   // If successful, append the log entry to the logs
+  // TODO: 这里logs_的appendlog操作和上边的stable_logger_->Logger()->Put不是原子的，可能导致offset大的先被追加到logs_中，
+  // 多线程写入的时候窗口会对不上，最终主从断开连接。需要加逻辑保证原子性
   logs_->AppendLog(Log::LogItem(cur_offset, cmd_ptr, binlog));
 
   SetPreparedId(cur_offset);
@@ -948,7 +953,8 @@ Status ConsensusCoordinator::ApplyBinlog(const std::shared_ptr<Cmd>& cmd_ptr) {
   } else {
     int32_t wait_ms = 250;
     while (g_pika_rm->GetUnfinishedAsyncWriteDBTaskCount(db_name_) > 0) {
-      std::this_thread::sleep_for(std::chrono::milliseconds(wait_ms));
+      // TODO: 暂时去掉了sleep的逻辑，考虑使用条件变量唤醒
+      //std::this_thread::sleep_for(std::chrono::milliseconds(wait_ms));
       wait_ms *= 2;
       wait_ms = wait_ms < 3000 ? wait_ms : 3000;
     }

src/pika_rm.cc

@@ -437,9 +437,9 @@ Status SyncMasterDB::CommitAppLog(const LogOffset& master_committed_id){
   return coordinator_.CommitAppLog(master_committed_id);
 }
 Status SyncMasterDB::AppendCandidateBinlog(const std::string& ip, int port, const LogOffset& offset) {
-   std::shared_ptr<SlaveNode> slave_ptr = GetSlaveNode(ip, port);  
+   std::shared_ptr<SlaveNode> slave_ptr = GetSlaveNode(ip, port);
   if (!slave_ptr) {
-    return Status::NotFound("ip " + ip + " port " + std::to_string(port)); 
+    return Status::NotFound("ip " + ip + " port " + std::to_string(port));
   }
 
   {
@@ -452,23 +452,23 @@ Status SyncMasterDB::AppendCandidateBinlog(const std::string& ip, int port, cons
     if(slave_ptr->slave_state == KCandidate){
       LOG(INFO)<<"PacificA first binlog slave_state is Candidate";
     }
-    slave_ptr->sent_offset = offset;           
+    slave_ptr->sent_offset = offset;
     slave_ptr->acked_offset = offset;
     slave_ptr->target_offset =GetPreparedId();
     Status s = slave_ptr->InitBinlogFileReader(Logger(), offset.b_offset);
     if (!s.ok()) {
       return Status::Corruption("Init binlog file reader failed" + s.ToString());  // 如果初始化失败，返回错误状态
     }
-    g_pika_rm->DropItemInOneWriteQueue(ip, port, slave_ptr->DBName());  
-    slave_ptr->b_state = kReadFromFile; 
+    g_pika_rm->DropItemInOneWriteQueue(ip, port, slave_ptr->DBName());
+    slave_ptr->b_state = kReadFromFile;
   }
 
   Status s = coordinator_.SendBinlog(slave_ptr, slave_ptr->DBName());
   if (!s.ok()) {
-    return s;  
+    return s;
   }
 
-  return Status::OK();  
+  return Status::OK();
 }
 
 Status SyncMasterDB::ConsensusProposeLog(const std::shared_ptr<Cmd>& cmd_ptr) {
@@ -489,9 +489,10 @@ Status SyncMasterDB::ConsensusProposeLog(const std::shared_ptr<Cmd>& cmd_ptr) {
     while (std::chrono::duration_cast<std::chrono::seconds>(std::chrono::steady_clock::now() - start).count() < 10) {
         // Check if consensus has been achieved for the given log offset
         if (checkFinished(offset)) {
-            return Status::OK(); 
+            return Status::OK();
         }
-        std::this_thread::sleep_for(std::chrono::milliseconds(50));
+        // TODO: 这里暂时注掉了sleep等待，50ms耗时过长，影响写入链路，后期需要改成条件变量唤醒方式
+        //std::this_thread::sleep_for(std::chrono::milliseconds(50));
     }
 
     return Status::Timeout("No consistency achieved within 10 seconds");