Optimize and test redis module in pipeline mode.

Author: zhengshuxin

lib_acl_cpp/include/acl_cpp/redis/redis_client_pipeline.hpp

@@ -67,10 +67,10 @@ class redis_pipeline_message {
 void set_option(size_t nchild, const int* timeout);
 
 // Called in redis_command::run()
-void move(dbuf_pool* dbuf);
+void set(dbuf_pool* dbuf);
 
 // Called in redis_command::build_request().
-void move(const string* req);
+void set(const string* req);
 
 // Called in redis_command::build_request().
 void set_slot(size_t slot);
@@ -135,8 +135,6 @@ class redis_pipeline_message {
 size_t nchild_;
 dbuf_pool* dbuf_;
 const string* req_;
-std::vector<dbuf_pool*> dbufs_;
-std::vector<const string*> reqs_;
 
 const redis_result* result_;
 

lib_acl_cpp/include/acl_cpp/redis/redis_command.hpp

@@ -446,6 +446,7 @@ class ACL_CPP_API redis_command : public noncopyable {
 
 private:
 void init();
+dbuf_pool *dbuf_create();
 
 public:
 // compute hash slot of the given key and store it in the current

lib_acl_cpp/src/redis/redis_client_pipeline.cpp

@@ -31,17 +31,6 @@ redis_pipeline_message::redis_pipeline_message(redis_pipeline_type_t type,
 }
 
 redis_pipeline_message::~redis_pipeline_message() {
-for (std::vector<const string*>::iterator it = reqs_.begin();
- it != reqs_.end(); ++it) {
-delete *it;
-COUNTER_DEC(pipeline_req);
-}
-
-for (std::vector<dbuf_pool*>::iterator it = dbufs_.begin();
- it != dbufs_.end(); ++it) {
-(*it)->destroy();
-COUNTER_DEC(pipeline_dbuf);
-}
 delete box_;
 COUNTER_DEC(redis_pipeline_message);
 }
@@ -58,21 +47,11 @@ void redis_pipeline_message::set_slot(size_t slot) {
 slot_ = slot;
 }
 
-void redis_pipeline_message::move(const string* req) {
-if (req) {
-reqs_.push_back(req);
-COUNTER_INC(pipeline_req);
-}
-
+void redis_pipeline_message::set(const string* req) {
 req_ = req;
 }
 
-void redis_pipeline_message::move(dbuf_pool* dbuf) {
-if (dbuf) {
-dbufs_.push_back(dbuf);
-COUNTER_INC(pipeline_dbuf);
-}
-
+void redis_pipeline_message::set(dbuf_pool* dbuf) {
 dbuf_ = dbuf;
 }
 

lib_acl_cpp/src/redis/redis_command.cpp

@@ -21,7 +21,7 @@ namespace acl {
 #defineLONG_LEN21
 #defineREDIS_DBUF_NBLOCK1
 
-static dbuf_pool *dbuf_alloc()
+dbuf_pool *redis_command::dbuf_create()
 {
 #ifdef ACL_DBUF_HOOK_NEW
 return new (REDIS_DBUF_NBLOCK) dbuf_pool();
@@ -51,7 +51,7 @@ void redis_command::init()
 result_ = NULL;
 pipe_msg_ = NULL;
 addr_[0] = 0;
-dbuf_ = dbuf_alloc();
+dbuf_ = dbuf_create();
 
 COUNTER_INC(redis_comand);
 }
@@ -400,12 +400,8 @@ const redis_result* redis_command::run(size_t nchild /* = 0 */,
 if (pipeline_ != NULL) {
 assert(pipe_msg_); // build_request() must have been called first.
 pipe_msg_->set_option(nchild, timeout);
-pipe_msg_->move(dbuf_);
-dbuf_ = dbuf_alloc();
+pipe_msg_->set(dbuf_);
 result_ = pipeline_->exec(pipe_msg_);
-pipe_msg_->unrefer();
-pipe_msg_ = NULL;
-
 return result_;
 }
 if (cluster_ != NULL) {
@@ -997,8 +993,7 @@ void redis_command::build_request(size_t argc, const char* argv[], const size_t
 if (pipeline_) {
 redis_pipeline_message* msg = get_pipeline_message();
 build_request1(argc, argv, lens);
-msg->move(request_buf_);
-request_buf_ = NULL;
+msg->set(request_buf_);
 msg->set_slot(static_cast<size_t>(slot_));
 } else if (slice_req_) {
 build_request2(argc, argv, lens);