supply function of expire/ttl for the other data types - list, hash, zset

fix tiny mistaks

ledis/ledis_db.go

@@ -26,6 +26,9 @@ func (db *DB) FlushAll() (drop int64, err error) {
 func (db *DB) activeExpireCycle() {
 	eliminator := newEliminator(db)
 	eliminator.regRetireContext(kvExpType, db.kvTx, db.delete)
+	eliminator.regRetireContext(lExpType, db.listTx, db.lDelete)
+	eliminator.regRetireContext(hExpType, db.hashTx, db.hDelete)
+	eliminator.regRetireContext(zExpType, db.zsetTx, db.zDelete)
 
 	go func() {
 		tick := time.NewTicker(1 * time.Second)

ledis/t_hash.go

@@ -4,6 +4,7 @@ import (
 	"encoding/binary"
 	"errors"
 	"github.com/siddontang/go-leveldb/leveldb"
+	"time"
 )
 
 type FVPair struct {
@@ -105,10 +106,6 @@ func (db *DB) hEncodeStopKey(key []byte) []byte {
 	return k
 }
 
-func (db *DB) HLen(key []byte) (int64, error) {
-	return Int64(db.db.Get(db.hEncodeSizeKey(key)))
-}
-
 func (db *DB) hSetItem(key []byte, field []byte, value []byte) (int64, error) {
 	t := db.hashTx
 
@@ -127,6 +124,49 @@ func (db *DB) hSetItem(key []byte, field []byte, value []byte) (int64, error) {
 	return n, nil
 }
 
+//	ps : here just focus on deleting the hash data,
+//		 any other likes expire is ignore.
+func (db *DB) hDelete(t *tx, key []byte) int64 {
+	sk := db.hEncodeSizeKey(key)
+	start := db.hEncodeStartKey(key)
+	stop := db.hEncodeStopKey(key)
+
+	var num int64 = 0
+	it := db.db.Iterator(start, stop, leveldb.RangeROpen, 0, -1)
+	for ; it.Valid(); it.Next() {
+		t.Delete(it.Key())
+		num++
+	}
+	it.Close()
+
+	t.Delete(sk)
+	return num
+}
+
+func (db *DB) hExpireAt(key []byte, when int64) (int64, error) {
+	t := db.hashTx
+	t.Lock()
+	defer t.Unlock()
+
+	if hlen, err := db.HLen(key); err != nil || hlen == 0 {
+		return 0, err
+	} else {
+		db.expireAt(t, hExpType, key, when)
+		if err := t.Commit(); err != nil {
+			return 0, err
+		}
+	}
+	return 1, nil
+}
+
+func (db *DB) HLen(key []byte) (int64, error) {
+	if err := checkKeySize(key); err != nil {
+		return 0, err
+	}
+
+	return Int64(db.db.Get(db.hEncodeSizeKey(key)))
+}
+
 func (db *DB) HSet(key []byte, field []byte, value []byte) (int64, error) {
 	if err := checkHashKFSize(key, field); err != nil {
 		return 0, err
@@ -265,6 +305,7 @@ func (db *DB) hIncrSize(key []byte, delta int64) (int64, error) {
 		if size <= 0 {
 			size = 0
 			t.Delete(sk)
+			db.rmExpire(t, hExpType, key)
 		} else {
 			t.Put(sk, PutInt64(size))
 		}
@@ -378,24 +419,12 @@ func (db *DB) HClear(key []byte) (int64, error) {
 		return 0, err
 	}
 
-	sk := db.hEncodeSizeKey(key)
-	start := db.hEncodeStartKey(key)
-	stop := db.hEncodeStopKey(key)
-
 	t := db.hashTx
 	t.Lock()
 	defer t.Unlock()
 
-	var num int64 = 0
+	num := db.hDelete(t, key)
-	it := db.db.Iterator(start, stop, leveldb.RangeROpen, 0, -1)
+	db.rmExpire(t, hExpType, key)
-	for ; it.Valid(); it.Next() {
-		t.Delete(it.Key())
-		num++
-	}
-
-	it.Close()
-
-	t.Delete(sk)
 
 	err := t.Commit()
 	return num, err
@@ -418,7 +447,7 @@ func (db *DB) HFlush() (drop int64, err error) {
 	for ; it.Valid(); it.Next() {
 		t.Delete(it.Key())
 		drop++
-		if drop%1000 == 0 {
+		if drop&1023 == 0 {
 			if err = t.Commit(); err != nil {
 				return
 			}
@@ -426,6 +455,8 @@ func (db *DB) HFlush() (drop int64, err error) {
 	}
 	it.Close()
 
+	db.expFlush(t, hExpType)
+
 	err = t.Commit()
 	return
 }
@@ -466,3 +497,27 @@ func (db *DB) HScan(key []byte, field []byte, count int, inclusive bool) ([]FVPa
 
 	return v, nil
 }
+
+func (db *DB) HExpire(key []byte, duration int64) (int64, error) {
+	if duration <= 0 {
+		return 0, errExpireValue
+	}
+
+	return db.hExpireAt(key, time.Now().Unix()+duration)
+}
+
+func (db *DB) HExpireAt(key []byte, when int64) (int64, error) {
+	if when <= time.Now().Unix() {
+		return 0, errExpireValue
+	}
+
+	return db.hExpireAt(key, when)
+}
+
+func (db *DB) HTTL(key []byte) (int64, error) {
+	if err := checkKeySize(key); err != nil {
+		return -1, err
+	}
+
+	return db.ttl(hExpType, key)
+}

ledis/t_kv.go

@@ -84,12 +84,30 @@ func (db *DB) incr(key []byte, delta int64) (int64, error) {
 	return n, err
 }
 
+//	ps : here just focus on deleting the key-value data,
+//		 any other likes expire is ignore.
 func (db *DB) delete(t *tx, key []byte) int64 {
 	key = db.encodeKVKey(key)
 	t.Delete(key)
 	return 1
 }
 
+func (db *DB) setExpireAt(key []byte, when int64) (int64, error) {
+	t := db.kvTx
+	t.Lock()
+	defer t.Unlock()
+
+	if exist, err := db.Exists(key); err != nil || exist == 0 {
+		return 0, err
+	} else {
+		db.expireAt(t, kvExpType, key, when)
+		if err := t.Commit(); err != nil {
+			return 0, err
+		}
+	}
+	return 1, nil
+}
+
 func (db *DB) Decr(key []byte) (int64, error) {
 	return db.incr(key, -1)
 }
@@ -115,7 +133,6 @@ func (db *DB) Del(keys ...[]byte) (int64, error) {
 	for i, k := range keys {
 		t.Delete(codedKeys[i])
 		db.rmExpire(t, kvExpType, k)
-		//todo binlog
 	}
 
 	err := t.Commit()
@@ -363,20 +380,7 @@ func (db *DB) Expire(key []byte, duration int64) (int64, error) {
 		return 0, errExpireValue
 	}
 
-	t := db.kvTx
+	return db.setExpireAt(key, time.Now().Unix()+duration)
-	t.Lock()
-	defer t.Unlock()
-
-	if exist, err := db.Exists(key); err != nil || exist == 0 {
-		return 0, err
-	} else {
-		db.expire(t, kvExpType, key, duration)
-		if err := t.Commit(); err != nil {
-			return 0, err
-		} else {
-			return 1, nil
-		}
-	}
 }
 
 func (db *DB) ExpireAt(key []byte, when int64) (int64, error) {
@@ -384,23 +388,10 @@ func (db *DB) ExpireAt(key []byte, when int64) (int64, error) {
 		return 0, errExpireValue
 	}
 
-	t := db.kvTx
+	return db.setExpireAt(key, when)
-	t.Lock()
-	defer t.Unlock()
-
-	if exist, err := db.Exists(key); err != nil || exist == 0 {
-		return 0, err
-	} else {
-		db.expireAt(t, kvExpType, key, when)
-		if err := t.Commit(); err != nil {
-			return 0, err
-		} else {
-			return 1, nil
-		}
-	}
 }
 
-func (db *DB) Ttl(key []byte) (int64, error) {
+func (db *DB) TTL(key []byte) (int64, error) {
 	if err := checkKeySize(key); err != nil {
 		return -1, err
 	}

ledis/t_list.go

@@ -4,6 +4,7 @@ import (
 	"encoding/binary"
 	"errors"
 	"github.com/siddontang/go-leveldb/leveldb"
+	"time"
 )
 
 const (
@@ -172,12 +173,17 @@ func (db *DB) lpop(key []byte, whereSeq int32) ([]byte, error) {
 	}
 
 	t.Delete(itemKey)
-	db.lSetMeta(metaKey, headSeq, tailSeq)
+	size := db.lSetMeta(metaKey, headSeq, tailSeq)
+	if size == 0 {
+		db.rmExpire(t, hExpType, key)
+	}
 
 	err = t.Commit()
 	return value, err
 }
 
+//	ps : here just focus on deleting the list data,
+//		 any other likes expire is ignore.
 func (db *DB) lDelete(t *tx, key []byte) int64 {
 	mk := db.lEncodeMetaKey(key)
 
@@ -230,7 +236,7 @@ func (db *DB) lGetMeta(ek []byte) (headSeq int32, tailSeq int32, size int32, err
 	return
 }
 
-func (db *DB) lSetMeta(ek []byte, headSeq int32, tailSeq int32) {
+func (db *DB) lSetMeta(ek []byte, headSeq int32, tailSeq int32) int32 {
 	t := db.listTx
 
 	var size int32 = tailSeq - headSeq + 1
@@ -243,10 +249,27 @@ func (db *DB) lSetMeta(ek []byte, headSeq int32, tailSeq int32) {
 
 		binary.LittleEndian.PutUint32(buf[0:4], uint32(headSeq))
 		binary.LittleEndian.PutUint32(buf[4:8], uint32(tailSeq))
-		//binary.LittleEndian.PutUint32(buf[8:], uint32(size))
 
 		t.Put(ek, buf)
 	}
+
+	return size
+}
+
+func (db *DB) lExpireAt(key []byte, when int64) (int64, error) {
+	t := db.listTx
+	t.Lock()
+	defer t.Unlock()
+
+	if llen, err := db.LLen(key); err != nil || llen == 0 {
+		return 0, err
+	} else {
+		db.expireAt(t, lExpType, key, when)
+		if err := t.Commit(); err != nil {
+			return 0, err
+		}
+	}
+	return 1, nil
 }
 
 func (db *DB) LIndex(key []byte, index int32) ([]byte, error) {
@@ -366,6 +389,7 @@ func (db *DB) LClear(key []byte) (int64, error) {
 	defer t.Unlock()
 
 	num := db.lDelete(t, key)
+	db.rmExpire(t, lExpType, key)
 
 	err := t.Commit()
 	return num, err
@@ -396,6 +420,32 @@ func (db *DB) LFlush() (drop int64, err error) {
 	}
 	it.Close()
 
+	db.expFlush(t, lExpType)
+
 	err = t.Commit()
 	return
 }
+
+func (db *DB) LExpire(key []byte, duration int64) (int64, error) {
+	if duration <= 0 {
+		return 0, errExpireValue
+	}
+
+	return db.lExpireAt(key, time.Now().Unix()+duration)
+}
+
+func (db *DB) LExpireAt(key []byte, when int64) (int64, error) {
+	if when <= time.Now().Unix() {
+		return 0, errExpireValue
+	}
+
+	return db.lExpireAt(key, when)
+}
+
+func (db *DB) LTTL(key []byte) (int64, error) {
+	if err := checkKeySize(key); err != nil {
+		return -1, err
+	}
+
+	return db.ttl(lExpType, key)
+}

ledis/t_ttl.go

@@ -15,7 +15,7 @@ var mapExpMetaType = map[byte]byte{
 
 type retireCallback func(*tx, []byte) int64
 
-type Elimination struct {
+type elimination struct {
 	db         *DB
 	exp2Tx     map[byte]*tx
 	exp2Retire map[byte]retireCallback
@@ -92,10 +92,12 @@ func (db *DB) ttl(expType byte, key []byte) (t int64, err error) {
 
 func (db *DB) rmExpire(t *tx, expType byte, key []byte) {
 	mk := db.expEncodeMetaKey(expType+1, key)
-	when, err := Int64(db.db.Get(mk))
+	if v, err := db.db.Get(mk); err != nil || v == nil {
-	if err == nil && when > 0 {
+		return
+	} else if when, err2 := Int64(v, nil); err2 != nil {
+		return
+	} else {
 		tk := db.expEncodeTimeKey(expType, key, when)
-
 		t.Delete(mk)
 		t.Delete(tk)
 	}
@@ -127,6 +129,7 @@ func (db *DB) expFlush(t *tx, expType byte) (err error) {
 			}
 		}
 	}
+	it.Close()
 
 	err = t.Commit()
 	return
@@ -136,21 +139,21 @@ func (db *DB) expFlush(t *tx, expType byte) (err error) {
 //
 //////////////////////////////////////////////////////////
 
-func newEliminator(db *DB) *Elimination {
+func newEliminator(db *DB) *elimination {
-	eli := new(Elimination)
+	eli := new(elimination)
 	eli.db = db
 	eli.exp2Tx = make(map[byte]*tx)
 	eli.exp2Retire = make(map[byte]retireCallback)
 	return eli
 }
 
-func (eli *Elimination) regRetireContext(expType byte, t *tx, onRetire retireCallback) {
+func (eli *elimination) regRetireContext(expType byte, t *tx, onRetire retireCallback) {
 	eli.exp2Tx[expType] = t
 	eli.exp2Retire[expType] = onRetire
 }
 
 //	call by outside ... (from *db to another *db)
-func (eli *Elimination) active() {
+func (eli *elimination) active() {
 	now := time.Now().Unix()
 	db := eli.db
 	dbGet := db.db.Get
@@ -202,6 +205,7 @@ func (eli *Elimination) active() {
 			t.Commit()
 			t.Unlock()
 		} // end : it
+		it.Close()
 	} // end : expType
 
 	return

ledis/t_ttl_test.go

@@ -69,7 +69,7 @@ func TestKvExpireAt(t *testing.T) {
 	}
 }
 
-func TestKvTtl(t *testing.T) {
+func TestKvTTL(t *testing.T) {
 	db := getTestDB()
 	m.Lock()
 	defer m.Unlock()
@@ -80,17 +80,17 @@ func TestKvTtl(t *testing.T) {
 	db.Set(k, []byte("1"))
 	db.Expire(k, 2)
 
-	if tRemain, _ := db.Ttl(k); tRemain != 2 {
+	if tRemain, _ := db.TTL(k); tRemain != 2 {
 		t.Fatal(tRemain)
 	}
 
 	//	err - check ttl on an inexisting key
-	if tRemain, _ := db.Ttl(ek); tRemain != -1 {
+	if tRemain, _ := db.TTL(ek); tRemain != -1 {
 		t.Fatal(tRemain)
 	}
 
 	db.Del(k)
-	if tRemain, _ := db.Ttl(k); tRemain != -1 {
+	if tRemain, _ := db.TTL(k); tRemain != -1 {
 		t.Fatal(tRemain)
 	}
 }
@@ -112,35 +112,35 @@ func TestKvExpCompose(t *testing.T) {
 	db.Expire(k1, 2)
 	db.Expire(k2, 60)
 
-	if tRemain, _ := db.Ttl(k0); tRemain != 5 {
+	if tRemain, _ := db.TTL(k0); tRemain != 5 {
 		t.Fatal(tRemain)
 	}
-	if tRemain, _ := db.Ttl(k1); tRemain != 2 {
+	if tRemain, _ := db.TTL(k1); tRemain != 2 {
 		t.Fatal(tRemain)
 	}
-	if tRemain, _ := db.Ttl(k2); tRemain != 60 {
+	if tRemain, _ := db.TTL(k2); tRemain != 60 {
 		t.Fatal(tRemain)
 	}
 
 	// after 1 sec
 	time.Sleep(1 * time.Second)
-	if tRemain, _ := db.Ttl(k0); tRemain != 4 {
+	if tRemain, _ := db.TTL(k0); tRemain != 4 {
 		t.Fatal(tRemain)
 	}
-	if tRemain, _ := db.Ttl(k1); tRemain != 1 {
+	if tRemain, _ := db.TTL(k1); tRemain != 1 {
 		t.Fatal(tRemain)
 	}
 
 	// after 2 sec
 	time.Sleep(2 * time.Second)
-	if tRemain, _ := db.Ttl(k1); tRemain != -1 {
+	if tRemain, _ := db.TTL(k1); tRemain != -1 {
 		t.Fatal(tRemain)
 	}
 	if v, _ := db.Get(k1); v != nil {
 		t.Fatal(v)
 	}
 
-	if tRemain, _ := db.Ttl(k0); tRemain != 2 {
+	if tRemain, _ := db.TTL(k0); tRemain != 2 {
 		t.Fatal(tRemain)
 	}
 	if v, _ := db.Get(k0); v == nil {
@@ -148,7 +148,7 @@ func TestKvExpCompose(t *testing.T) {
 	}
 
 	// refresh the expiration of key
-	if tRemain, _ := db.Ttl(k2); !(0 < tRemain && tRemain < 60) {
+	if tRemain, _ := db.TTL(k2); !(0 < tRemain && tRemain < 60) {
 		t.Fatal(tRemain)
 	}
 
@@ -156,7 +156,7 @@ func TestKvExpCompose(t *testing.T) {
 		t.Fatal(false)
 	}
 
-	if tRemain, _ := db.Ttl(k2); tRemain != 100 {
+	if tRemain, _ := db.TTL(k2); tRemain != 100 {
 		t.Fatal(tRemain)
 	}
 
@@ -164,7 +164,7 @@ func TestKvExpCompose(t *testing.T) {
 	if ok, _ := db.Expire(k1, 10); ok == 1 {
 		t.Fatal(false)
 	}
-	if tRemain, _ := db.Ttl(k1); tRemain != -1 {
+	if tRemain, _ := db.TTL(k1); tRemain != -1 {
 		t.Fatal(tRemain)
 	}
 

ledis/t_zset.go

@@ -5,6 +5,7 @@ import (
 	"encoding/binary"
 	"errors"
 	"github.com/siddontang/go-leveldb/leveldb"
+	"time"
 )
 
 const (
@@ -191,13 +192,11 @@ func (db *DB) zDecodeScoreKey(ek []byte) (key []byte, member []byte, score int64
 	return
 }
 
-func (db *DB) zSetItem(key []byte, score int64, member []byte) (int64, error) {
+func (db *DB) zSetItem(t *tx, key []byte, score int64, member []byte) (int64, error) {
 	if score <= MinScore || score >= MaxScore {
 		return 0, errScoreOverflow
 	}
 
-	t := db.zsetTx
-
 	var exists int64 = 0
 	ek := db.zEncodeSetKey(key, member)
 
@@ -222,9 +221,7 @@ func (db *DB) zSetItem(key []byte, score int64, member []byte) (int64, error) {
 	return exists, nil
 }
 
-func (db *DB) zDelItem(key []byte, member []byte, skipDelScore bool) (int64, error) {
+func (db *DB) zDelItem(t *tx, key []byte, member []byte, skipDelScore bool) (int64, error) {
-	t := db.zsetTx
-
 	ek := db.zEncodeSetKey(key, member)
 	if v, err := db.db.Get(ek); err != nil {
 		return 0, err
@@ -245,6 +242,29 @@ func (db *DB) zDelItem(key []byte, member []byte, skipDelScore bool) (int64, err
 	}
 
 	t.Delete(ek)
+
+	return 1, nil
+}
+
+func (db *DB) zDelete(t *tx, key []byte) int64 {
+	delMembCnt, _ := db.zRemRange(t, key, MinScore, MaxScore, 0, -1)
+	//	todo : log err
+	return delMembCnt
+}
+
+func (db *DB) zExpireAt(key []byte, when int64) (int64, error) {
+	t := db.zsetTx
+	t.Lock()
+	defer t.Unlock()
+
+	if zcnt, err := db.ZCard(key); err != nil || zcnt == 0 {
+		return 0, err
+	} else {
+		db.expireAt(t, zExpType, key, when)
+		if err := t.Commit(); err != nil {
+			return 0, err
+		}
+	}
 	return 1, nil
 }
 
@@ -266,7 +286,7 @@ func (db *DB) ZAdd(key []byte, args ...ScorePair) (int64, error) {
 			return 0, err
 		}
 
-		if n, err := db.zSetItem(key, score, member); err != nil {
+		if n, err := db.zSetItem(t, key, score, member); err != nil {
 			return 0, err
 		} else if n == 0 {
 			//add new
@@ -274,7 +294,7 @@ func (db *DB) ZAdd(key []byte, args ...ScorePair) (int64, error) {
 		}
 	}
 
-	if _, err := db.zIncrSize(key, num); err != nil {
+	if _, err := db.zIncrSize(t, key, num); err != nil {
 		return 0, err
 	}
 
@@ -283,8 +303,7 @@ func (db *DB) ZAdd(key []byte, args ...ScorePair) (int64, error) {
 	return num, err
 }
 
-func (db *DB) zIncrSize(key []byte, delta int64) (int64, error) {
+func (db *DB) zIncrSize(t *tx, key []byte, delta int64) (int64, error) {
-	t := db.zsetTx
 	sk := db.zEncodeSizeKey(key)
 
 	size, err := Int64(db.db.Get(sk))
@@ -295,6 +314,7 @@ func (db *DB) zIncrSize(key []byte, delta int64) (int64, error) {
 		if size <= 0 {
 			size = 0
 			t.Delete(sk)
+			db.rmExpire(t, zExpType, key)
 		} else {
 			t.Put(sk, PutInt64(size))
 		}
@@ -348,14 +368,14 @@ func (db *DB) ZRem(key []byte, members ...[]byte) (int64, error) {
 			return 0, err
 		}
 
-		if n, err := db.zDelItem(key, members[i], false); err != nil {
+		if n, err := db.zDelItem(t, key, members[i], false); err != nil {
 			return 0, err
 		} else if n == 1 {
 			num++
 		}
 	}
 
-	if _, err := db.zIncrSize(key, -num); err != nil {
+	if _, err := db.zIncrSize(t, key, -num); err != nil {
 		return 0, err
 	}
 
@@ -374,34 +394,35 @@ func (db *DB) ZIncrBy(key []byte, delta int64, member []byte) (int64, error) {
 
 	ek := db.zEncodeSetKey(key, member)
 
-	var score int64 = delta
+	var oldScore int64 = 0
-
 	v, err := db.db.Get(ek)
 	if err != nil {
 		return InvalidScore, err
-	} else if v != nil {
+	} else if v == nil {
-		if s, err := Int64(v, err); err != nil {
+		db.zIncrSize(t, key, 1)
-			return InvalidScore, err
-		} else {
-			sk := db.zEncodeScoreKey(key, member, s)
-			t.Delete(sk)
-
-			score = s + delta
-			if score >= MaxScore || score <= MinScore {
-				return InvalidScore, errScoreOverflow
-			}
-		}
 	} else {
-		db.zIncrSize(key, 1)
+		if oldScore, err = Int64(v, err); err != nil {
+			return InvalidScore, err
+		}
 	}
 
-	t.Put(ek, PutInt64(score))
+	newScore := oldScore + delta
+	if newScore >= MaxScore || newScore <= MinScore {
+		return InvalidScore, errScoreOverflow
+	}
 
-	sk := db.zEncodeScoreKey(key, member, score)
+	sk := db.zEncodeScoreKey(key, member, newScore)
 	t.Put(sk, []byte{})
+	t.Put(ek, PutInt64(newScore))
+
+	if v != nil {
+		// so as to update score, we must delete the old one
+		oldSk := db.zEncodeScoreKey(key, member, oldScore)
+		t.Delete(oldSk)
+	}
 
 	err = t.Commit()
-	return score, err
+	return newScore, err
 }
 
 func (db *DB) ZCount(key []byte, min int64, max int64) (int64, error) {
@@ -482,42 +503,35 @@ func (db *DB) zIterator(key []byte, min int64, max int64, offset int, limit int,
 	}
 }
 
-func (db *DB) zRemRange(key []byte, min int64, max int64, offset int, limit int) (int64, error) {
+func (db *DB) zRemRange(t *tx, key []byte, min int64, max int64, offset int, limit int) (int64, error) {
 	if len(key) > MaxKeySize {
 		return 0, errKeySize
 	}
 
-	t := db.zsetTx
-	t.Lock()
-	defer t.Unlock()
-
 	it := db.zIterator(key, min, max, offset, limit, false)
 	var num int64 = 0
 	for ; it.Valid(); it.Next() {
-		k := it.Key()
+		sk := it.Key()
-		_, m, _, err := db.zDecodeScoreKey(k)
+		_, m, _, err := db.zDecodeScoreKey(sk)
 		if err != nil {
 			continue
 		}
 
-		if n, err := db.zDelItem(key, m, true); err != nil {
+		if n, err := db.zDelItem(t, key, m, true); err != nil {
 			return 0, err
 		} else if n == 1 {
 			num++
 		}
 
-		t.Delete(k)
+		t.Delete(sk)
 	}
 	it.Close()
 
-	if _, err := db.zIncrSize(key, -num); err != nil {
+	if _, err := db.zIncrSize(t, key, -num); err != nil {
 		return 0, err
 	}
 
-	//todo add binlog
+	return num, nil
-
-	err := t.Commit()
-	return num, err
 }
 
 func (db *DB) zReverse(s []interface{}, withScores bool) []interface{} {
@@ -624,7 +638,16 @@ func (db *DB) zParseLimit(key []byte, start int, stop int) (offset int, limit in
 }
 
 func (db *DB) ZClear(key []byte) (int64, error) {
-	return db.zRemRange(key, MinScore, MaxScore, 0, -1)
+	t := db.zsetTx
+	t.Lock()
+	defer t.Unlock()
+
+	rmCnt, err := db.zRemRange(t, key, MinScore, MaxScore, 0, -1)
+	if err == nil {
+		err = t.Commit()
+	}
+
+	return rmCnt, err
 }
 
 func (db *DB) ZRange(key []byte, start int, stop int, withScores bool) ([]interface{}, error) {
@@ -647,12 +670,33 @@ func (db *DB) ZRemRangeByRank(key []byte, start int, stop int) (int64, error) {
 	if err != nil {
 		return 0, err
 	}
-	return db.zRemRange(key, MinScore, MaxScore, offset, limit)
+
+	var rmCnt int64
+
+	t := db.zsetTx
+	t.Lock()
+	defer t.Unlock()
+
+	rmCnt, err = db.zRemRange(t, key, MinScore, MaxScore, offset, limit)
+	if err == nil {
+		err = t.Commit()
+	}
+
+	return rmCnt, err
 }
 
 //min and max must be inclusive
 func (db *DB) ZRemRangeByScore(key []byte, min int64, max int64) (int64, error) {
-	return db.zRemRange(key, min, max, 0, -1)
+	t := db.zsetTx
+	t.Lock()
+	defer t.Unlock()
+
+	rmCnt, err := db.zRemRange(t, key, min, max, 0, -1)
+	if err == nil {
+		err = t.Commit()
+	}
+
+	return rmCnt, err
 }
 
 func (db *DB) ZRevRange(key []byte, start int, stop int, withScores bool) ([]interface{}, error) {
@@ -705,7 +749,7 @@ func (db *DB) ZFlush() (drop int64, err error) {
 	for ; it.Valid(); it.Next() {
 		t.Delete(it.Key())
 		drop++
-		if drop%1000 == 0 {
+		if drop&1023 == 0 {
 			if err = t.Commit(); err != nil {
 				return
 			}
@@ -713,8 +757,9 @@ func (db *DB) ZFlush() (drop int64, err error) {
 	}
 	it.Close()
 
+	db.expFlush(t, zExpType)
+
 	err = t.Commit()
-	// to do : binlog
 	return
 }
 
@@ -756,3 +801,27 @@ func (db *DB) ZScan(key []byte, member []byte, count int, inclusive bool) ([]Sco
 
 	return v, nil
 }
+
+func (db *DB) ZExpire(key []byte, duration int64) (int64, error) {
+	if duration <= 0 {
+		return 0, errExpireValue
+	}
+
+	return db.zExpireAt(key, time.Now().Unix()+duration)
+}
+
+func (db *DB) ZExpireAt(key []byte, when int64) (int64, error) {
+	if when <= time.Now().Unix() {
+		return 0, errExpireValue
+	}
+
+	return db.zExpireAt(key, when)
+}
+
+func (db *DB) ZTTL(key []byte) (int64, error) {
+	if err := checkKeySize(key); err != nil {
+		return -1, err
+	}
+
+	return db.ttl(zExpType, key)
+}