bugfix binlog sender confict at multi-network interface (#234)

* bugfix binlog sender confict at multi-network interface

* bugfix configing set maxclients has no effect

* set resource
       limits

include/pika_binlog_sender_thread.h

@@ -20,6 +20,7 @@ class PikaBinlogSenderThread : public pink::Thread {
  public:
 
  PikaBinlogSenderThread(const std::string &ip, int port,
+ int64_t sid,
  slash::SequentialFile *queue,
  uint32_t filenum, uint64_t con_offset);
 
@@ -50,6 +51,7 @@ class PikaBinlogSenderThread : public pink::Thread {
 
  std::string ip_;
  int port_;
+ int64_t sid_;
 
  int timeout_ms_;
  pink::PinkCli *cli_;

include/pika_dispatch_thread.h

@@ -22,6 +22,10 @@ class PikaDispatchThread {
  bool ClientKill(const std::string& ip_port);
  void ClientKillAll();
 
+ void SetQueueLimit(int queue_limit) {
+ thread_rep_->SetQueueLimit(queue_limit);
+ }
+
  private:
  class ClientConnFactory : public pink::ConnFactory {
  public:

include/pika_master_conn.h

@@ -16,6 +16,7 @@ class PikaMasterConn: public pink::RedisConn {
  PikaMasterConn(int fd, std::string ip_port, void* worker_specific_data);
  virtual int DealMessage(PikaCmdArgsType& argv, std::string* response);
  private:
+ bool is_first_send_;
  PikaBinlogReceiverThread* binlog_receiver_;
 };
 

include/pika_server.h

@@ -70,6 +70,15 @@ class PikaServer {
  int master_port() {
  return master_port_;
  }
+
+ int64_t sid() {
+ return sid_;
+ }
+
+ void SetSid(int64_t sid) {
+ sid_ = sid;
+ }
+
  const std::shared_ptr<nemo::Nemo> db() {
  return db_;
  }
@@ -184,7 +193,8 @@ class PikaServer {
  */
  Binlog *logger_;
  Status AddBinlogSender(const std::string& ip, int64_t port,
- uint32_t filenum, uint64_t con_offset);
+ int64_t sid,
+ uint32_t filenum, uint64_t con_offset);
 
  /*
  * BGSave used
@@ -457,6 +467,8 @@ class PikaServer {
  void ResetStat();
  slash::RecordMutex mutex_record_;
 
+ void SetDispatchQueueLimit(int queue_limit);
+
  private:
  std::atomic<bool> exit_;
  std::atomic<bool> binlog_io_error_;

src/pika_admin.cc

@@ -163,7 +163,8 @@ void TrysyncCmd::Do() {
  int64_t sid = g_pika_server->TryAddSlave(slave_ip_, slave_port_);
  if (sid >= 0) {
  Status status = g_pika_server->AddBinlogSender(slave_ip_, slave_port_,
- filenum_, pro_offset_);
+ sid,
+ filenum_, pro_offset_);
  if (status.ok()) {
  res_.AppendInteger(sid);
  LOG(INFO) << "Send Sid to Slave: " << sid;
@@ -1320,6 +1321,7 @@ void ConfigCmd::ConfigSet(std::string& ret) {
  ret = "-ERR Invalid argument " + value + " for CONFIG SET 'maxclients'\r\n";
  return;
  }
+ g_pika_server->SetDispatchQueueLimit(ival);
  g_pika_conf->SetMaxConnection(ival);
  ret = "+OK\r\n";
  } else if (set_item == "dump-expire") {

src/pika_binlog_receiver_thread.cc

@@ -38,11 +38,11 @@ bool PikaBinlogReceiverThread::Handles::AccessHandle(std::string& ip) const {
  if (ip == "127.0.0.1") {
  ip = g_pika_server->host();
  }
-//  if (binlog_receiver_->thread_rep_->conn_num() != 0 ||
- if (!g_pika_server->ShouldAccessConnAsMaster(ip)) {
- LOG(WARNING) << "BinlogReceiverThread AccessHandle failed: " << ip;
- return false;
- }
+ // if (binlog_receiver_->thread_rep_->conn_num() != 0 ||
+ // if (!g_pika_server->ShouldAccessConnAsMaster(ip)) {
+ // LOG(WARNING) << "BinlogReceiverThread AccessHandle failed: " << ip;
+ // return false;
+ // }
  g_pika_server->PlusMasterConnection();
  return true;
 }

src/pika_binlog_sender_thread.cc

@@ -17,6 +17,7 @@
 extern PikaServer* g_pika_server;
 
 PikaBinlogSenderThread::PikaBinlogSenderThread(const std::string &ip, int port,
+ int64_t sid,
  slash::SequentialFile *queue,
  uint32_t filenum,
  uint64_t con_offset)
@@ -29,6 +30,7 @@ PikaBinlogSenderThread::PikaBinlogSenderThread(const std::string &ip, int port,
  buffer_(),
  ip_(ip),
  port_(port),
+ sid_(sid),
  timeout_ms_(35000) {
  cli_ = pink::NewRedisCli();
  last_record_offset_ = con_offset % kBlockSize;
@@ -257,6 +259,17 @@ void* PikaBinlogSenderThread::ThreadMain() {
 
  if (result.ok()) {
  cli_->set_send_timeout(timeout_ms_);
+ // Auth sid
+ std::string wbuf_str;
+ pink::RedisCmdArgsType argv;
+ argv.push_back("auth");
+ argv.push_back(std::to_string(sid_));
+ pink::SerializeRedisCommand(argv, &wbuf_str);
+ result = cli_->Send(&wbuf_str);
+ if (!result.ok()) {
+ LOG(WARNING) << "BinlogSender send slave(" << ip_ << ":" << port_ << ") failed, " << result.ToString();
+ break;
+ }
  while (true) {
  // 2. Should Parse new msg;
  if (last_send_flag) {

src/pika_master_conn.cc

@@ -17,6 +17,7 @@ extern PikaConf* g_pika_conf;
 PikaMasterConn::PikaMasterConn(int fd, std::string ip_port,
  void* worker_specific_data)
  : RedisConn(fd, ip_port, NULL) {
+ is_first_send_ = true;
  binlog_receiver_ =
  reinterpret_cast<PikaBinlogReceiverThread*>(worker_specific_data);
 }
@@ -30,6 +31,19 @@ int PikaMasterConn::DealMessage(
  return -2;
  }
 
+ // Auth
+ if (is_first_send_) {
+ if (argv[0] == "auth") {
+ if (argv[1] == std::to_string(g_pika_server->sid())) {
+ is_first_send_ = false;
+ LOG(INFO) << "BinlogReceiverThread AccessHandle success, my sid: " << g_pika_server->sid() << " auth sid: " << argv[1];
+ return 0;
+ }
+ }
+ LOG(WARNING) << "BinlogReceiverThread AccessHandle failed";
+ return -2;
+ }
+
  // TODO(shq) maybe monitor do not need these infomation
  std::string server_id;
  std::string binlog_info;

src/pika_server.cc

@@ -16,6 +16,7 @@
 #include <iterator>
 #include <ctime>
 #include <algorithm>
+#include <sys/resource.h>
 
 #include "slash/include/env.h"
 #include "slash/include/rsync.h"
@@ -850,7 +851,8 @@ void PikaServer::DBSyncSendFile(const std::string& ip, int port) {
  * BinlogSender
  */
 Status PikaServer::AddBinlogSender(const std::string& ip, int64_t port,
- uint32_t filenum, uint64_t con_offset) {
+ int64_t sid,
+ uint32_t filenum, uint64_t con_offset) {
  // Sanity check
  if (con_offset > logger_->file_size()) {
  return Status::InvalidArgument("AddBinlogSender invalid binlog offset");
@@ -885,7 +887,7 @@ Status PikaServer::AddBinlogSender(const std::string& ip, int64_t port,
  }
 
  PikaBinlogSenderThread* sender = new PikaBinlogSenderThread(ip,
- port + 1000, readfile, filenum, con_offset);
+ port + 1000, sid, readfile, filenum, con_offset);
 
  if (sender->trim() == 0 // Error binlog
  && SetSlaveSender(ip, port, sender)) { // SlaveItem not exist
@@ -1720,6 +1722,25 @@ void PikaServer::ResetStat() {
  statistic_data_.last_thread_querynum = 0;
 }
 
+void PikaServer::SetDispatchQueueLimit(int queue_limit) {
+ rlimit limit;
+ if (getrlimit(RLIMIT_NOFILE,&limit) == -1) {
+ LOG(WARNING) << "getrlimit error: " << strerror(errno);
+ } else if (limit.rlim_cur < static_cast<unsigned int>(g_pika_conf->maxclients() + PIKA_MIN_RESERVED_FDS)) {
+ rlim_t old_limit = limit.rlim_cur;
+ rlim_t best_limit = g_pika_conf->maxclients() + PIKA_MIN_RESERVED_FDS;
+ limit.rlim_cur = best_limit > limit.rlim_max ? limit.rlim_max-1 : best_limit;
+ limit.rlim_max = best_limit > limit.rlim_max ? limit.rlim_max-1 : best_limit;
+ if (setrlimit(RLIMIT_NOFILE,&limit) != -1) {
+ LOG(WARNING) << "your 'limit -n ' of " << old_limit << " is not enough for Redis to start. pika have successfully reconfig it to " << limit.rlim_cur;
+ } else {
+ LOG(FATAL) << "your 'limit -n ' of " << old_limit << " is not enough for Redis to start. pika can not reconfig it(" << strerror(errno) << "), do it by yourself";
+ }
+ }
+
+ pika_dispatch_thread_->SetQueueLimit(queue_limit);
+}
+
 uint64_t PikaServer::ServerCurrentQps() {
  slash::ReadLock l(&statistic_data_.statistic_lock);
  return statistic_data_.last_sec_thread_querynum;

src/pika_trysync_thread.cc

@@ -104,6 +104,7 @@ bool PikaTrysyncThread::RecvProc() {
  slash::string2l(reply.data(), reply.size(), &sid_)) {
  // Luckly, I got your point, the sync is comming
  LOG(INFO) << "Recv sid from master: " << sid_;
+ g_pika_server->SetSid(sid_);
  break;
  }