From 8d76a58e5e933051ea0809555cf899771e131ea6 Mon Sep 17 00:00:00 2001 From: silentsai Date: Thu, 5 Jun 2014 16:13:35 +0800 Subject: [PATCH] supply function of expire/ttl for the other data types - list, hash, zset fix tiny mistaks --- ledis/ledis_db.go | 3 + ledis/t_hash.go | 93 +++++++++++++++++++----- ledis/t_kv.go | 51 ++++++-------- ledis/t_list.go | 56 ++++++++++++++- ledis/t_ttl.go | 20 +++--- ledis/t_ttl_test.go | 28 ++++---- ledis/t_zset.go | 167 +++++++++++++++++++++++++++++++------------- 7 files changed, 295 insertions(+), 123 deletions(-) diff --git a/ledis/ledis_db.go b/ledis/ledis_db.go index cae79f9..2390464 100644 --- a/ledis/ledis_db.go +++ b/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) diff --git a/ledis/t_hash.go b/ledis/t_hash.go index 68bdc49..ac5d5cf 100644 --- a/ledis/t_hash.go +++ b/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 - 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) + num := db.hDelete(t, key) + db.rmExpire(t, hExpType, key) 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) +} diff --git a/ledis/t_kv.go b/ledis/t_kv.go index caa26dd..3208761 100644 --- a/ledis/t_kv.go +++ b/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 - 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 - } - } + return db.setExpireAt(key, time.Now().Unix()+duration) } 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 - 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 - } - } + return db.setExpireAt(key, when) } -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 } diff --git a/ledis/t_list.go b/ledis/t_list.go index 5d1355b..1532dd8 100644 --- a/ledis/t_list.go +++ b/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) +} diff --git a/ledis/t_ttl.go b/ledis/t_ttl.go index 167acce..b353ce7 100644 --- a/ledis/t_ttl.go +++ b/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 err == nil && when > 0 { + if v, err := db.db.Get(mk); err != nil || v == nil { + 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 { - eli := new(Elimination) +func newEliminator(db *DB) *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 diff --git a/ledis/t_ttl_test.go b/ledis/t_ttl_test.go index 0483ae7..0f2cdac 100644 --- a/ledis/t_ttl_test.go +++ b/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) } diff --git a/ledis/t_zset.go b/ledis/t_zset.go index 37bbc61..f10ce3f 100644 --- a/ledis/t_zset.go +++ b/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) { - t := db.zsetTx - +func (db *DB) zDelItem(t *tx, key []byte, member []byte, skipDelScore bool) (int64, error) { 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) { - t := db.zsetTx +func (db *DB) zIncrSize(t *tx, key []byte, delta int64) (int64, error) { 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 { - if s, err := Int64(v, err); err != nil { - 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 if v == nil { + db.zIncrSize(t, key, 1) } 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() - _, m, _, err := db.zDecodeScoreKey(k) + sk := it.Key() + _, 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 - - err := t.Commit() - return num, err + return num, nil } 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) +}