test redis_pipeline

lib_acl_cpp/include/acl_cpp/redis/redis_client_pipeline.hpp

@@ -22,12 +22,16 @@ class token_tree;
 class redis_client;
 
 typedef enum {
-	redis_pipeline_t_cmd,
+	redis_pipeline_t_cmd,		// redis command type
-	redis_pipeline_t_redirect,
+	redis_pipeline_t_redirect,	// should redirect to another node
-	redis_pipeline_t_clusterdonw,
+ 	redis_pipeline_t_clusterdonw,	// the redis node has been down
-	redis_pipeline_t_stop,
+	redis_pipeline_t_stop,		// the current channel should stop
 } redis_pipeline_type_t;
 
+/**
+ * the message for transfering between redis command, redis client pipline
+ * and redis pipeline channel, which holds the redis command or not.
+ */
 class redis_pipeline_message {
 public:
 	redis_pipeline_message(redis_command* cmd, redis_pipeline_type_t type)

lib_acl_cpp/include/acl_cpp/redis/redis_command.hpp

@@ -345,10 +345,17 @@ public:
 	static void build_request(size_t argc, const char* argv[],
 		size_t lens[], string& out);
 
+	/**
+	 * 根据命令字列表创建 redis 请求协议并存储于 redis_command 中，以备在请求时使用
+	 * @param argc {size_t} 命令参数个数
+	 * @param argv {const char* []} 命令参数数组
+	 * @param lens {size_t []} 每个命令参数长度数组
+	 */
+	void build_request(size_t argc, const char* argv[], size_t lens[]);
+
 protected:
 	const redis_result* run(size_t nchild = 0, int* timeout = NULL);
 
-	void build_request(size_t argc, const char* argv[], size_t lens[]);
 	void clear_request(void);
 	const redis_result** scan_keys(const char* cmd, const char* key,
 		int& cursor, size_t& size, const char* pattern,
@@ -480,6 +487,7 @@ public:
 		return slice_req_;
 	}
 
+	// get pipline message bound with the current command
 	redis_pipeline_message& get_pipeline_message(void);
 
 protected:
@@ -492,8 +500,13 @@ protected:
 	size_t* argv_lens_;
 	size_t  argc_;
 
+	// reserve the argv space with the specified value at least
 	void argv_space(size_t n);
+
+	// build request in one request buffer
 	void build_request1(size_t argc, const char* argv[], size_t lens[]);
+
+	// build request with slice request obj
 	void build_request2(size_t argc, const char* argv[], size_t lens[]);
 
 protected:
@@ -502,6 +515,7 @@ protected:
 	redis_pipeline_message* pipe_msg_;
 	const redis_result* result_;
 
+	// save the error info into log
 	void logger_result(const redis_result* result);
 };
 

lib_acl_cpp/samples/redis/redis_pipeline2/Makefile

@@ -0,0 +1,3 @@
+base_path = ../../..
+PROG = redis_pipeline
+include ../../Makefile.in

lib_acl_cpp/samples/redis/redis_pipeline2/redis_pipeline.cpp

@@ -0,0 +1,207 @@
+#include "stdafx.h"
+#include "util.h"
+
+static int __threads_exit = 0;
+
+class redis_command {
+public:
+	redis_command(acl::redis_client_pipeline& pipeline)
+	: msg_(cmd_.get_pipeline_message())
+	{
+		cmd_.set_pipeline(&pipeline);
+		argc_ = 2;
+		argv_[0] = "del";
+		argv_[1] = "test-key";
+		lens_[0] = strlen(argv_[0]);
+		lens_[1] = strlen(argv_[1]);
+		cmd_.build_request(argc_, argv_, lens_);
+	}
+
+	~redis_command(void) {}
+
+	acl::redis_pipeline_message& get_message(void) {
+		return msg_;
+	}
+
+private:
+	acl::redis cmd_;
+	acl::redis_pipeline_message& msg_;
+	size_t argc_;
+	const char* argv_[2];
+	size_t lens_[2];
+
+};
+
+class test_thread : public acl::thread {
+public:
+	test_thread(acl::locker& locker, acl::redis_client_pipeline& pipeline,
+		const char* cmd, int n)
+	: locker_(locker)
+	, pipeline_(pipeline)
+	, cmd_(cmd)
+	, n_(n)
+	{
+		(void) cmd_;
+	}
+
+	~test_thread(void) {}
+
+protected:
+	// @override
+	void* run(void) {
+		std::vector<redis_command*> commands;
+		for (size_t i = 0; i < 10; i++) {
+			redis_command* command = new redis_command(pipeline_);
+			commands.push_back(command);
+		}
+
+		for (int i = 0; i < n_; i++) {
+			request(commands);
+		}
+
+		for (std::vector<redis_command*>::iterator it = commands.begin();
+			    it != commands.end(); ++it) {
+			delete *it;
+		}
+
+		locker_.lock();
+		__threads_exit++;
+		locker_.unlock();
+
+		return NULL;
+	}
+
+private:
+	void request(std::vector<redis_command*>& commands) {
+		// send all request commands
+		for (std::vector<redis_command*>::iterator it = commands.begin();
+			it != commands.end(); ++it) {
+			acl::redis_pipeline_message& msg =
+				(*it)->get_message();
+			pipeline_.push(&msg);
+		}
+
+		// wait for all results
+		for (std::vector<redis_command*>::iterator it = commands.begin();
+			it != commands.end(); ++it) {
+			acl::redis_pipeline_message& msg =
+				(*it)->get_message();
+			const acl::redis_result* result = msg.wait();
+			if (result == NULL) {
+				printf("wait result error\r\n");
+				break;
+			}
+		}
+	}
+
+private:
+	acl::locker& locker_;
+	acl::redis_client_pipeline& pipeline_;
+	acl::string cmd_;
+	int n_;
+};
+
+static void usage(const char* procname) {
+	printf("usage: %s -h[help]\r\n"
+		"-s one_redis_addr[127.0.0.1:6379]\r\n"
+		"-n count[default: 10]\r\n"
+		"-C connect_timeout[default: 10]\r\n"
+		"-I rw_timeout[default: 10]\r\n"
+		"-c max_threads[default: 10]\r\n"
+		"-w wait_for_cluster_resume[default: 500 ms]\r\n"
+		"-r retry_for_cluster_resnum[default: 10]\r\n"
+		"-P password [set the password of redis cluster]\r\n"
+		"-a cmd[set|get|expire|ttl|exists|type|del]\r\n",
+		procname);
+}
+
+int main(int argc, char* argv[]) {
+	int  ch, n = 1, conn_timeout = 10, rw_timeout = 10;
+	int  max_threads = 10;
+	acl::string addr("127.0.0.1:6379"), cmd("del"), passwd;
+
+	while ((ch = getopt(argc, argv, "hs:n:C:I:c:a:p:")) > 0) {
+		switch (ch) {
+		case 'h':
+			usage(argv[0]);
+			return 0;
+		case 's':
+			addr = optarg;
+			break;
+		case 'n':
+			n = atoi(optarg);
+			break;
+		case 'C':
+			conn_timeout = atoi(optarg);
+			break;
+		case 'I':
+			rw_timeout = atoi(optarg);
+			break;
+		case 'c':
+			max_threads = atoi(optarg);
+			break;
+		case 'a':
+			cmd = optarg;
+			break;
+		case 'p':
+			passwd = optarg;
+			break;;
+		default:
+			break;
+		}
+	}
+
+	acl::acl_cpp_init();
+	acl::log::stdout_open(true);
+
+	acl::redis_client_pipeline pipeline(addr);
+	pipeline.start_thread();
+
+	struct timeval begin;
+	gettimeofday(&begin, NULL);
+
+	acl::locker locker;
+
+	std::vector<test_thread*> threads;
+	for (int i = 0; i < max_threads; i++) {
+		test_thread* thread = new test_thread(locker, pipeline,
+			cmd.c_str(), n);
+		threads.push_back(thread);
+		thread->set_detachable(true);
+		thread->start();
+	}
+
+	while (true) {
+		locker.lock();
+		if (__threads_exit == max_threads) {
+			locker.unlock();
+			printf("All threads over now!\r\n");
+			break;
+		}
+		locker.unlock();
+
+		//printf("max_threads: %d, threads_exit: %d\r\n",
+		//	max_threads, __threads_exit);
+		sleep(1);
+	}
+
+	std::vector<test_thread*>::iterator it = threads.begin();
+	for (; it != threads.end(); ++it) {
+		//(*it)->wait();
+		delete (*it);
+	}
+
+	struct timeval end;
+	gettimeofday(&end, NULL);
+
+	long long int total = max_threads * n;
+	double inter = util::stamp_sub(&end, &begin);
+	printf("total %s: %lld, spent: %0.2f ms, speed: %0.2f\r\n", cmd.c_str(),
+		total, inter, (total * 1000) /(inter > 0 ? inter : 1));
+
+#ifdef WIN32
+	printf("enter any key to exit\r\n");
+	getchar();
+#endif
+	return 0;
+}

lib_acl_cpp/samples/redis/redis_pipeline2/stdafx.cpp

@@ -0,0 +1,8 @@
+// stdafx.cpp : 只包括标准包含文件的源文件
+// xml.pch 将成为预编译头
+// stdafx.obj 将包含预编译类型信息
+
+#include "stdafx.h"
+
+// TODO: 在 STDAFX.H 中
+//引用任何所需的附加头文件，而不是在此文件中引用

lib_acl_cpp/samples/redis/redis_pipeline2/stdafx.h

@@ -0,0 +1,14 @@
+// stdafx.h : 标准系统包含文件的包含文件，
+// 或是常用但不常更改的项目特定的包含文件
+//
+
+#pragma once
+
+//
+//#include <iostream>
+//#include <tchar.h>
+
+// TODO: 在此处引用程序要求的附加头文件
+#include "acl_cpp/lib_acl.hpp"
+#include "lib_acl.h"
+

lib_acl_cpp/samples/redis/redis_pipeline2/valgrind.sh

@@ -0,0 +1,3 @@
+#!/bin/sh
+
+valgrind --tool=memcheck --leak-check=yes -v ./redis_pipeline -s 127.0.0.1:6379 -a del -n 10 -c 10

lib_acl_cpp/src/redis/redis_client_pipeline.cpp

@@ -356,6 +356,8 @@ void redis_client_pipeline::push(redis_pipeline_message *msg)
 	box_.push(msg, false);
 }
 
+// called after the thread started
+// @override from acl::thread
 void* redis_client_pipeline::run(void)
 {
 	set_all_slot();