fixed one bug in redis_client_pipeline.cpp about handling redirect message.

Author: zhengshuxin

Makefile

@@ -82,7 +82,7 @@ endif
 ##############################################################################
 
 .PHONY = check help all_lib all samples all clean install uninstall uninstall_all build_one
-VERSION = 3.5.3-13
+VERSION = 3.5.3-14
 
 default: build_one acl_master
 help h:

lib_acl/src/init/acl_init.c

@@ -25,7 +25,7 @@
 
 #include "init.h"
 
-static char *version = "3.5.3-13 20220607-20:36";
+static char *version = "3.5.3-14 20220615-15:00";
 
 const char *acl_version(void)
 {

lib_acl_cpp/include/acl_cpp/redis/redis_client_pipeline.hpp

@@ -22,15 +22,15 @@ class token_tree;
 class redis_client;
 
 typedef enum {
-	redis_pipeline_t_cmd,		// redis command type
-	redis_pipeline_t_redirect,	// should redirect to another node
- 	redis_pipeline_t_clusterdonw,	// the redis node has been down
-	redis_pipeline_t_stop,		// the current channel should stop
+	redis_pipeline_t_cmd,		// Redis command type
+	redis_pipeline_t_redirect,	// Should redirect to another node
+ 	redis_pipeline_t_clusterdonw,	// The redis node has been down
+	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.
+ * The message for transfering between redis command, redis client pipline
+ * and redis pipeline channel, which holds the redis command.
  */
 class redis_pipeline_message {
 public:
@@ -176,6 +176,10 @@ class redis_pipeline_message {
 
 class redis_client_pipeline;
 
+/**
+ * One pipeline channel thread for one redis node, which waits for message
+ * from pipline thread and try to combine more messages and sends to redis.
+ */
 class redis_pipeline_channel : public thread {
 public:
 	redis_pipeline_channel(redis_client_pipeline& pipeline,
@@ -214,7 +218,7 @@ class redis_pipeline_channel : public thread {
 };
 
 /**
- * redis pipline communication, be set and used in redis_command to
+ * Redis pipline communication, be set and used in redis_command to
  * improve the performance of redis commands, but not all redis commands
  * in acl can be used in pipeline mode, such as below:
  * 1. multiple keys operation
@@ -225,36 +229,36 @@ class ACL_CPP_API redis_client_pipeline : public thread {
 	redis_client_pipeline(const char* addr);
 	~redis_client_pipeline(void);
 
-	// start the pipeline thread
+	// Start the pipeline thread
 	void start_thread(void);
 
-	// stop the pipeline thread
+	// Stop the pipeline thread
 	void stop_thread(void);
 
 public:
-	// called by redis_command in pipeline mode
+	// Called by redis_command in pipeline mode
 	const redis_result* run(redis_pipeline_message& msg);
 
-	// called by redis_pipeline_channel
+	// Called by redis_pipeline_channel
 	void push(redis_pipeline_message* msg);
 
 public:
-	// set the password for connecting the redis server
+	// Set the password for connecting the redis server
 	redis_client_pipeline& set_password(const char* passwd);
 
-	// set network IO timeout
+	// Set network IO timeout
 	redis_client_pipeline& set_timeout(int conn_timeout, int rw_timeout);
 
-	// set if retry on IO failed in redis_client
+	// Set if retry on IO failed in redis_client
 	redis_client_pipeline& set_retry(bool on);
 
-	// set the max hash slot of redis, the default valud is 16384
+	// Set the max hash slot of redis, the default valud is 16384
 	redis_client_pipeline& set_max_slot(int max_slot);
 
-	// set if connecting all the redis nodes after starting
+	// Set if connecting all the redis nodes after starting
 	redis_client_pipeline& set_preconnect(bool yes);
 
-	// get the max hash slot of redis
+	// Get the max hash slot of redis
 	int get_max_slot(void) const {
 		return max_slot_;
 	}
@@ -264,62 +268,62 @@ class ACL_CPP_API redis_client_pipeline : public thread {
 	void* run(void);
 
 private:
-	string addr_;		// the default redis address
-	string passwd_;		// password for connecting redis
-	int max_slot_;		// the max hash slot for redis cluster
-	int    conn_timeout_;	// timeout to connect redis
+	string addr_;		// The default redis address
+	string passwd_;		// Password for connecting redis
+	int max_slot_;		// The max hash slot for redis cluster
+	int    conn_timeout_;	// Timeout to connect redis
 	int    rw_timeout_;	// IO timeout with redis
-	bool   retry_;		// if try again when disconnect from redis
-	bool   preconn_;	// if connecting all redis nodes when starting
+	bool   retry_;		// If try again when disconnect from redis
+	bool   preconn_;	// If connecting all redis nodes when starting
 
 	token_tree* channels_;	// holds and manage all pipeline channels
 
-	// the message queue for receiving redis message from other threads
+	// The message queue for receiving redis message from other threads
 	BOX<redis_pipeline_message> box_;
 
-	std::vector<char*> addrs_;	// hold all redises addresses
-	const char** slot_addrs_;	// map hash slot with address
+	std::vector<char*> addrs_;	// Hold all redis's addresses
+	const char** slot_addrs_;	// Map hash slot with address
 
-	// set the hash slot with the specified redis address
+	// Set the hash slot with the specified redis address
 	void set_slot(int slot, const char* addr);
 
-	// set all hash slots' addresses of all redises
+	// Set all hash slots' addresses of all redises
 	void set_all_slot(void);
 
-	// start all pipeline channels threads
+	// Start all pipeline channels threads
 	void start_channels(void);
 
-	// stop all pipeline channels threads
+	// Stop all pipeline channels threads
 	void stop_channels(void);
 
-	// start one pipeline channel thread with the specified redis address
+	// Start one pipeline channel thread with the specified redis address
 	redis_pipeline_channel* start_channel(const char* addr);
 
-	// stop one pipeline channel thread with the specified redis address
+	// Stop one pipeline channel thread with the specified redis address
 	void stop_channel(const char* addr);
 
-	// get one pipeline channel thread with the specified hash slot
+	// Get one pipeline channel thread with the specified hash slot
 	redis_pipeline_channel* get_channel(int slot);
 
-	// redirect one slot to another redis address
+	// Redirect one slot to another redis address
 	void redirect(const redis_pipeline_message& msg, int slot);
 
-	// when one redis node down, we should clear the node's hash slot map
+	// When one redis node down, we should clear the node's hash slot map
 	// and stop the pipeline channel thread
 	void cluster_down(const redis_pipeline_message& msg);
 };
 
 /**
- * sample:
+ * Sample:
  * void main_thread(void) {
  *	acl::redis_client_pipeline pipeline("127.0.0.1:6379");
  *	pipeline.start_thread();
- *	// start some threads
+ *	// Start some threads
  *	...
- *	// wait for thease threads to exit and stop pipeline thread.
+ *	// Wait for thease threads to exit and stop pipeline thread.
  *	pipeline.stop_thread();
  * }
- * // execute redis command in one thread
+ * // Execute redis command in one thread
  * void test_thread(acl::redis_client_pipeline& pipeline) {
  *	acl::redis cmd;
  *	cmd.set_pipeline(&pipeline);

lib_acl_cpp/samples/redis/redis_pipeline3/Makefile

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

lib_acl_cpp/samples/redis/redis_pipeline3/redis_pipeline.cpp

@@ -0,0 +1,67 @@
+#include "stdafx.h"
+
+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 loop_count[default: 10]\r\n"
+		"-p password [set the password of redis cluster]\r\n"
+		"-a cmd[hset|get|expire|ttl|exists|type|del]\r\n",
+		procname);
+}
+
+int main(int argc, char* argv[]) {
+	int  ch, count = 2;
+	acl::string addr("127.0.0.1:6379"), passwd;
+	acl::string cmd("del");
+
+	while ((ch = getopt(argc, argv, "ha:s:n:p:")) > 0) {
+		switch (ch) {
+		case 'h':
+			usage(argv[0]);
+			return 0;
+		case 'a':
+			cmd = optarg;
+			break;
+		case 's':
+			addr = optarg;
+			break;
+		case 'n':
+			count = atoi(optarg);
+			break;
+		case 'p':
+			passwd = optarg;
+			break;;
+		default:
+			break;
+		}
+	}
+
+	acl::acl_cpp_init();
+	acl::log::stdout_open(true);
+
+	acl::redis_client_pipeline pipeline(addr);
+	if (!passwd.empty()) {
+		pipeline.set_password(passwd);
+	}
+	pipeline.start_thread();
+
+	acl::string key, name, value;
+
+	acl::redis redis(&pipeline);
+	for (int i = 0; i < count; i++) {
+		key.format("hkey-%d", i);
+		name.format("hname-%d", i);
+		value.format("hvalue-%d", i);
+
+		if (redis.hset(key, name, value) >= 0) {
+			printf("hset %s %s %s ok\n",
+				key.c_str(), name.c_str(), value.c_str());
+		} else {
+			printf("hset %s %s %s error\n",
+				key.c_str(), name.c_str(), value.c_str());
+		}
+	}
+
+	pipeline.stop_thread();
+	return 0;
+}

lib_acl_cpp/samples/redis/redis_pipeline3/stdafx.cpp

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

lib_acl_cpp/samples/redis/redis_pipeline3/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_pipeline3/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