Merge branch 'GoRedisDev'

Author: latermoon

.gitignore

@@ -1,5 +1,3 @@
-
-
 .DS_Store
 
 main/goredis-server

doc/release notes.md

@@ -1,10 +1,20 @@
 GoRedis release notes
 =====================
 
+**SlaveOf 1.0.4** @2014.4.11
+
+* [Feature] 支持配置dbpath
+
 **Proxy 1.0.4** @2014.3.30
 
 * [Feature] 实现基本功能
 
+
+**GoRedis 1.0.69** @2014.4.10
+* [Fix] 修正PrefixEnumerate时全部遍历没有quit
+* [ADD]	启动参数新增logpath、datapath用于自定义数据和Log路径
+
+
 **GoRedis 1.0.68** @2014.3.30
 
 * [Fix] 修正SADD返回值

goredis/session.go

@@ -15,7 +15,7 @@ import (
 // Session继承了net.Conn，代表一个客户端会话
 // 提供各种标准的Reply方法, Status/Error/Integer/Bulk/MultiBulks
 // cmd, err := session.ReadCommand()
-// session.ReplyReply(StatusReply("OK"))
+// session.WriteReply(StatusReply("OK"))
 // 协议参考：http://redis.io/topics/protocol
 type Session struct {
 	net.Conn

goredis_proxy/goredis_proxy.go

@@ -88,7 +88,7 @@ func (server *GoRedisProxy) On(session *Session, cmd *Command) (reply *Reply) {
 		// 写入主库
 		if server.master.Available() {
 			if server.options.CanWrite() {
-				reply, err = server.master.Invoke(session, cmd)
+				reply, err = server.master.Send(session, cmd)
 				session.SetAttribute(S_LAST_WRITE_KEY, key)
 			} else {
 				err = errors.New("reject write")
@@ -107,7 +107,7 @@ func (server *GoRedisProxy) On(session *Session, cmd *Command) (reply *Reply) {
 			remotes[0], remotes[1] = remotes[1], remotes[0]
 		}
 		for _, remote := range remotes {
-			reply, err = remote.Invoke(session, cmd)
+			reply, err = remote.Send(session, cmd)
 			session.SetAttribute(S_LAST_WRITE_KEY, "")
 			if err == nil {
 				break
@@ -132,6 +132,7 @@ func (server *GoRedisProxy) invokeCommand(session *Session, cmd *Command) (reply
 	case "PING":
 		return StatusReply("PONG")
 	}
+	// 禁止执行某些危险指令
 	if ignoreSync[cmdName] {
 		return ErrorReply("not support")
 	}

goredis_proxy/remote_session.go

@@ -66,23 +66,29 @@ func (s *RemoteSession) secondTicker() {
 }
 
 // 发送指令到远程Redis，并返回结果
-func (s *RemoteSession) Invoke(session *Session, cmd *Command) (reply *Reply, err error) {
+func (s *RemoteSession) Send(session *Session, cmd *Command) (reply *Reply, err error) {
 	if !s.available {
 		err = errors.New("unavailable")
 		return
 	}
 	i := s.indexOf([]byte(session.RemoteAddr().String()))
-	s.counters.Get("total").Incr(1)
 	// lock
 	s.mus[i].Lock()
 	defer s.mus[i].Unlock()
+
+	s.counters.Get("total").Incr(1)
 	// redirect
 	err = s.sessions[i].WriteCommand(cmd)
 	if err == nil {
 		reply, err = s.sessions[i].ReadReply()
 	}
 	if err != nil {
-		s.available = false
+		s.counters.Get("error").Incr(1)
+		s.sessions[i].Close()
+		s.sessions[i], err = s.createSession() // reconnect
+		if err != nil {
+			s.available = false
+		}
 	}
 	return
 }
@@ -113,7 +119,7 @@ func (s *RemoteSession) Close() {
 
 func (s *RemoteSession) createSession() (session *Session, err error) {
 	var conn net.Conn
-	conn, err = net.Dial("tcp", s.host)
+	conn, err = net.DialTimeout("tcp", s.host, time.Millisecond*1000)
 	if err != nil {
 		return
 	}

goredis_server/go_redis_server.go

@@ -17,7 +17,7 @@ import (
 )
 
 // 版本号，每次更新都需要增加
-const VERSION = "1.0.68"
+const VERSION = "1.0.69"
 const PREFIX = "__goredis:"
 
 var (

libs/levelredis/level_redis.go

@@ -376,12 +376,22 @@ func (l *LevelRedis) Keys(prefix []byte, fn func(i int, key, keytype []byte, qui
 // 前缀扫描
 func (l *LevelRedis) PrefixEnumerate(prefix []byte, direction IterDirection, fn func(i int, key, value []byte, quit *bool)) {
 	min := prefix
+	/**
+	*
+	 */
 	max := append(prefix, MAXBYTE)
 	j := -1
 	l.RangeEnumerate(min, max, direction, func(i int, key, value []byte, quit *bool) {
 		if bytes.HasPrefix(key, prefix) {
 			j++
 			fn(j, key, value, quit)
+		} else {
+			/**
+			 * 根据leveldb 的 key有序，因此具有相同前缀的key必定是在一起的
+			 * 所以一旦碰见了没有该前缀的key那么就直接退出，结束遍历
+			 */
+
+			*quit = true
 		}
 	})
 	return

main/goredis-server.go

@@ -28,8 +28,8 @@ func main() {
 	slaveof := flag.String("slaveof", "", "replication")
 	procs := flag.Int("procs", 8, "GOMAXPROCS")
 	repair := flag.Bool("repair", false, "repaire rocksdb")
-	logpath := flag.String("logpath", "/home/data", "config goredis log path and synclog path")
-
+	datapath := flag.String("datapath", "/data", "config goredis data path default path [/data/goredis_${port}/]")
+	logpath := flag.String("logpath", "/home/logs/", "config goredis log path and synclog path ,default path [/home/logs/goredis_${port}]")
 	flag.Parse()
 
 	if *version {
@@ -41,9 +41,9 @@ func main() {
 
 	opt := goredis_server.NewOptions()
 	opt.SetBind(fmt.Sprintf("%s:%d", *host, *port))
-
-	dbhome := dbHome(*port)
+	dbhome := dbHome(*datapath, *port)
 	opt.SetDirectory(dbhome)
+
 	if len(*slaveof) > 0 {
 		h, p, e := splitHostPort(*slaveof)
 		if e != nil {
@@ -53,7 +53,9 @@ func main() {
 	}
 
 	// 重定向日志输出位置
-	opt.SetLogDir(redirectLogOutput(*logpath, *port, dbhome))
+	logdir := redirectLogOutput(*logpath, *port)
+	opt.SetLogDir(logdir)
+	stdlog.Println("logdir:[" + logdir + "]\tdbhome:[" + dbhome + "]")
 
 	// repair
 	if *repair {
@@ -88,35 +90,42 @@ func init() {
 	})
 }
 
-// 主路径，默认在/data下创建，否则在/tmp下
-func dbHome(port int) string {
-	dbhome := "/data"
-	finfo, e1 := os.Stat(dbhome)
-	if os.IsNotExist(e1) || !finfo.IsDir() {
-		dbhome = "/tmp"
+/**
+ * 构建dbhome，使用datapath中的路径
+ */
+func dbHome(datapath string, port int) string {
+
+	dbhome := fmt.Sprintf("%s/goredis_%d/", datapath, port)
+
+	finfo, err := os.Stat(dbhome)
+	if os.IsNotExist(err) || !finfo.IsDir() {
+		os.MkdirAll(dbhome, os.ModePerm)
 	}
-	directory := fmt.Sprintf("%s/goredis_%d/", dbhome, port)
-	os.MkdirAll(directory, os.ModePerm)
-	return directory
+	return dbhome
 }
 
-// 将Stdout, Stderr重定向到指定文件
-func redirectLogOutput(directory string, port int, defdir string) string {
+/**
+ * 将Stdout, Stderr重定向到指定文件
+ * 并返回当前日志路径
+ */
+func redirectLogOutput(directory string, port int) string {
 
-	logpath := directory
+	oldout := os.Stdout
 
-	loginfo, err := os.Stat(directory)
-	//如果没有就是用默认的即 dbhome
+	logpath := fmt.Sprintf("%s/goredis_%d/", directory, port)
+
+	loginfo, err := os.Stat(logpath)
+
+	/**
+	 * 如果logpath不存在
+	 * 则创建
+	 */
 	if os.IsNotExist(err) || !loginfo.IsDir() {
-		directory = defdir
+		os.MkdirAll(logpath, os.ModePerm)
 	}
 
-	logpath = fmt.Sprintf("%s/logs/%d/", logpath, port)
-	os.MkdirAll(logpath, os.ModePerm)
+	os.Stdout, err = os.OpenFile(logpath+"stdout.log", os.O_WRONLY|os.O_CREATE|os.O_APPEND, os.ModePerm)
 
-	oldout := os.Stdout
-
-	os.Stdout, err = os.OpenFile(logpath+"/stdout.log", os.O_WRONLY|os.O_CREATE|os.O_APPEND, os.ModePerm)
 	if err != nil {
 		panic(err)
 	}
@@ -127,6 +136,7 @@ func redirectLogOutput(directory string, port int, defdir string) string {
 	if err != nil {
 		panic(err)
 	}
+
 	return logpath
 }
 

main/slaveof-proxy.go

@@ -17,6 +17,7 @@ func main() {
 	pullrate := flag.Int("pullrate", 400, "pull rate in Mbits/s")
 	pushrate := flag.Int("pushrate", 400, "push rate in Mbits/s")
 	buffer := flag.Int("buffer", 100, "buffer x10000 records")
+	dbpath := flag.String("dbpath", "/tmp", "rdb path")
 	flag.Parse()
 
 	if *pullrate < 100 {
@@ -31,7 +32,12 @@ func main() {
 		*buffer = 1000
 	}
 
-	stdlog.Println("slaveof-proxy 1.0.3")
+	stdlog.Println("slaveof-proxy 1.0.4")
+	if len(*src) == 0 || len(*dest) == 0 {
+		stdlog.Println("Usage: ./slaveof-proxy -src master:port -dest slave:6379 -pullrate 400 -pushrate 400 -buffer 100 -dbpath /tmp")
+		return
+	}
+
 	stdlog.Printf("from [%s] to [%s]\n", *src, *dest)
 	stdlog.Printf("pull [%d] buffer [%d]\n", *pullrate, *buffer)
 	stdlog.Println("SYNC ...")
@@ -43,7 +49,11 @@ func main() {
 	}
 	client.SetPullRate(*pullrate / 8 * 1024 * 1024)
 	client.SetPushRate(*pushrate / 8 * 1024 * 1024)
-	client.Sync()
+	client.SetDbPath(*dbpath)
+	err = client.Sync()
+	if err != nil {
+		panic(err)
+	}
 }
 
 func init() {

main/tool/slaveof/slave_client.go

@@ -77,6 +77,10 @@ func (s *SlaveClient) SetPushRate(n int) {
 	s.pushrate = n
 }
 
+func (s *SlaveClient) SetDbPath(dbpath string) {
+	s.directory = dbpath
+}
+
 func (s *SlaveClient) Writer() io.Writer {
 	if s.destwriter == nil {
 		s.destwriter = ratelimit.NewRateLimiter(s.Dest(), s.pushrate)