Bugfix of hub senders's close; Bugfix of pexpireat's tobinlog

beckyame · Nov 29, 2017 · b172afa · b172afa
1 parent c30999c
commit b172afa
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 15 deletions.
diff --git a/include/pika_hub_manager.h b/include/pika_hub_manager.h
@@ -37,21 +37,12 @@ class PikaHubSenderThread : public pink::Thread {
 
   int TryStartThread(const std::string& hub_ip, const int hub_port);
 
-  int GetTid() {
-    return tid_;
-  }
+  int GetTid() { return tid_; }
 
   enum STATUS { UNSTARTED, WORKING, WAITING, ENDOFFILE };
-  STATUS SenderStatus() {
-    return status_;
-  }
+  STATUS SenderStatus() { return status_; }
 
-  void CloseClient() {
-    cli_->Close();
-    // Close socket connnection
-    cli_.reset(pink::NewRedisCli());
-    should_reset_ = true;
-  }
+  void Reset() { should_reset_ = true; should_reconnect_ = true; }
 
  private:
   bool ResetStatus();
@@ -66,6 +57,7 @@ class PikaHubSenderThread : public pink::Thread {
 
   STATUS status_;
   std::atomic<bool> should_reset_;
+  std::atomic<bool> should_reconnect_;
 
   uint32_t filenum_;
   uint64_t con_offset_;

diff --git a/src/pika_hub_manager.cc b/src/pika_hub_manager.cc
@@ -19,6 +19,7 @@ extern PikaServer* g_pika_server;
 PikaHubManager::PikaHubManager(const std::set<std::string> &ips, int port,
                                int cron_interval)
     : hub_stage_(STOPED),
+      hub_port_(0),
       hub_filenum_(0),
       hub_con_offset_(0),
       sending_window_({0, -1}),
@@ -245,7 +246,7 @@ void PikaHubManager::StopHub(int connnection_num) {
     } else {
       hub_stage_ = STOPED;
       for (int i = 0; i < kMaxHubSender; i++) {
-        sender_threads_[i]->CloseClient();
+        sender_threads_[i]->Reset();
       }
       slash::MutexLock l(&sending_window_protector_);
       sending_window_.left = 0;
@@ -524,6 +525,7 @@ void* PikaHubSenderThread::ThreadMain() {
   bool should_wait = true;
   while (!should_stop()) {
     if (should_reset_) {
+      cli_.reset(pink::NewRedisCli());
       should_wait = ResetStatus();
       if (should_wait) {
         status_ = UNSTARTED;
@@ -539,6 +541,7 @@ void* PikaHubSenderThread::ThreadMain() {
     // 1. Connect to hub
     result = cli_->Connect(hub_ip_, hub_port_, g_pika_server->host());
     if (result.ok()) {
+      should_reconnect_ = false;
       LOG(INFO) << "Hub Sender Connect hub(" << hub_ip_ << ":" <<
         hub_port_ << ") " << result.ToString() << ", filenum: " << filenum_;
 
@@ -559,7 +562,7 @@ void* PikaHubSenderThread::ThreadMain() {
         }
 
         // 3. After successful parse, we send msg;
-        if (!cli_->Available()) {
+        if (should_reconnect_) {
           last_send_flag = false;
           break;
         }

diff --git a/src/pika_kv.cc b/src/pika_kv.cc
@@ -924,7 +924,7 @@ std::string PexpireatCmd::ToBinlog(
   RedisAppendContent(res, key_);
   // sec
   char buf[100];
-  int64_t expireat = time(nullptr) + time_stamp_ms_ / 1000;
+  int64_t expireat = time_stamp_ms_ / 1000;
   slash::ll2string(buf, 100, expireat);
   std::string at(buf);
   RedisAppendLen(res, at.size(), "$");