17582991793

Committed 09 Sep 2025 12:38PM UTC coverage: 72.6% (-0.06%) from 72.663%

Build Type

github

Committed by hu-qing-hua

Commit Message

fix logic risk errors in raft.go

Pull Request Pull Request #258: fix logic risk errors in raft.go

Run Details

4 of 11 new or added lines in 1 file covered. (36.36%)

4 existing lines in 2 files now uncovered.

8667 of 11938 relevant lines covered (72.6%)

0.81 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

86.91

/database/database.go

// Package database is a memory database with redis compatible interface
package database

import (
        "strings"
        "time"

        "github.com/hdt3213/godis/datastruct/dict"
        "github.com/hdt3213/godis/interface/database"
        "github.com/hdt3213/godis/interface/redis"
        "github.com/hdt3213/godis/lib/logger"
        "github.com/hdt3213/godis/lib/timewheel"
        "github.com/hdt3213/godis/redis/protocol"
)

const (
        dataDictSize = 1 << 16
        ttlDictSize  = 1 << 10
)

// DB stores data and execute user's commands
type DB struct {
        index int
        // key -> DataEntity
        data *dict.ConcurrentDict
        // key -> expireTime (time.Time)
        ttlMap *dict.ConcurrentDict
        // key -> version(uint32)
        versionMap *dict.ConcurrentDict

        // addaof is used to add command to aof
        addAof func(CmdLine)

        // callbacks
        insertCallback database.KeyEventCallback
        deleteCallback database.KeyEventCallback
}

// ExecFunc is interface for command executor
// args don't include cmd line
type ExecFunc func(db *DB, args [][]byte) redis.Reply

// PreFunc analyses command line when queued command to `multi`
// returns related write keys and read keys
type PreFunc func(args [][]byte) ([]string, []string)

// CmdLine is alias for [][]byte, represents a command line
type CmdLine = [][]byte

// UndoFunc returns undo logs for the given command line
// execute from head to tail when undo
type UndoFunc func(db *DB, args [][]byte) []CmdLine

// makeDB create DB instance
func makeDB() *DB {
        db := &DB{
                data:       dict.MakeConcurrent(dataDictSize),
                ttlMap:     dict.MakeConcurrent(ttlDictSize),
                versionMap: dict.MakeConcurrent(dataDictSize),
                addAof:     func(line CmdLine) {},
        }
        return db
}

// makeBasicDB create DB instance only with basic abilities.
func makeBasicDB() *DB {
        db := &DB{
                data:       dict.MakeConcurrent(dataDictSize),
                ttlMap:     dict.MakeConcurrent(ttlDictSize),
                versionMap: dict.MakeConcurrent(dataDictSize),
                addAof:     func(line CmdLine) {},
        }
        return db
}

// Exec executes command within one database
func (db *DB) Exec(c redis.Connection, cmdLine [][]byte) redis.Reply {
        // transaction control commands and other commands which cannot execute within transaction
        cmdName := strings.ToLower(string(cmdLine[0]))
        if cmdName == "multi" {
                if len(cmdLine) != 1 {
                        return protocol.MakeArgNumErrReply(cmdName)
                }
                return StartMulti(c)
        } else if cmdName == "discard" {
                if len(cmdLine) != 1 {
                        return protocol.MakeArgNumErrReply(cmdName)
                }
                return DiscardMulti(c)
        } else if cmdName == "exec" {
                if len(cmdLine) != 1 {
                        return protocol.MakeArgNumErrReply(cmdName)
                }
                return execMulti(db, c)
        } else if cmdName == "watch" {
                if !validateArity(-2, cmdLine) {
                        return protocol.MakeArgNumErrReply(cmdName)
                }
                return Watch(db, c, cmdLine[1:])
        }
        if c != nil && c.InMultiState() {
                return EnqueueCmd(c, cmdLine)
        }

        return db.execNormalCommand(cmdLine)
}

func (db *DB) execNormalCommand(cmdLine [][]byte) redis.Reply {
        cmdName := strings.ToLower(string(cmdLine[0]))
        cmd, ok := cmdTable[cmdName]
        if !ok {
                return protocol.MakeErrReply("ERR unknown command '" + cmdName + "'")
        }
        if !validateArity(cmd.arity, cmdLine) {
                return protocol.MakeArgNumErrReply(cmdName)
        }

        prepare := cmd.prepare
        write, read := prepare(cmdLine[1:])
        db.addVersion(write...)
        db.RWLocks(write, read)
        defer db.RWUnLocks(write, read)
        fun := cmd.executor
        return fun(db, cmdLine[1:])
}

// execWithLock executes normal commands, invoker should provide locks
func (db *DB) execWithLock(cmdLine [][]byte) redis.Reply {
        cmdName := strings.ToLower(string(cmdLine[0]))
        cmd, ok := cmdTable[cmdName]
        if !ok {
                return protocol.MakeErrReply("ERR unknown command '" + cmdName + "'")
        }
        if !validateArity(cmd.arity, cmdLine) {
                return protocol.MakeArgNumErrReply(cmdName)
        }
        fun := cmd.executor
        return fun(db, cmdLine[1:])
}

func validateArity(arity int, cmdArgs [][]byte) bool {
        argNum := len(cmdArgs)
        if arity >= 0 {
                return argNum == arity
        }
        return argNum >= -arity
}

/* ---- Data Access ----- */

// GetEntity returns DataEntity bind to given key
func (db *DB) GetEntity(key string) (*database.DataEntity, bool) {
        raw, ok := db.data.GetWithLock(key)
        if !ok {
                return nil, false
        }
        if db.IsExpired(key) {
                return nil, false
        }
        entity, _ := raw.(*database.DataEntity)
        return entity, true
}

// PutEntity a DataEntity into DB
func (db *DB) PutEntity(key string, entity *database.DataEntity) int {
        ret := db.data.PutWithLock(key, entity)
        // db.insertCallback may be set as nil, during `if` and actually callback
        // so introduce a local variable `cb`
        if cb := db.insertCallback; ret > 0 && cb != nil {
                cb(db.index, key, entity)
        }
        return ret
}

// PutIfExists edit an existing DataEntity
func (db *DB) PutIfExists(key string, entity *database.DataEntity) int {
        return db.data.PutIfExistsWithLock(key, entity)
}

// PutIfAbsent insert an DataEntity only if the key not exists
func (db *DB) PutIfAbsent(key string, entity *database.DataEntity) int {
        ret := db.data.PutIfAbsentWithLock(key, entity)
        // db.insertCallback may be set as nil, during `if` and actually callback
        // so introduce a local variable `cb`
        if cb := db.insertCallback; ret > 0 && cb != nil {
                cb(db.index, key, entity)
        }
        return ret
}

// Remove the given key from db
func (db *DB) Remove(key string) {
        raw, deleted := db.data.RemoveWithLock(key)
        db.ttlMap.Remove(key)
        taskKey := genExpireTask(key)
        timewheel.Cancel(taskKey)
        if cb := db.deleteCallback; cb != nil {
                var entity *database.DataEntity
                if deleted > 0 {
                        entity = raw.(*database.DataEntity)
                }
                cb(db.index, key, entity)
        }
}

// Removes the given keys from db
func (db *DB) Removes(keys ...string) (deleted int) {
        deleted = 0
        for _, key := range keys {
                _, exists := db.data.GetWithLock(key)
                if exists {
                        db.Remove(key)
                        deleted++
                }
        }
        return deleted
}

// Flush clean database
// deprecated
// for test only
func (db *DB) Flush() {
        db.data.Clear()
        db.ttlMap.Clear()
}

/* ---- Lock Function ----- */

// RWLocks lock keys for writing and reading
func (db *DB) RWLocks(writeKeys []string, readKeys []string) {
        db.data.RWLocks(writeKeys, readKeys)
}

// RWUnLocks unlock keys for writing and reading
func (db *DB) RWUnLocks(writeKeys []string, readKeys []string) {
        db.data.RWUnLocks(writeKeys, readKeys)
}

/* ---- TTL Functions ---- */

func genExpireTask(key string) string {
        return "expire:" + key
}

// Expire sets ttlCmd of key
func (db *DB) Expire(key string, expireTime time.Time) {
        db.ttlMap.Put(key, expireTime)
        taskKey := genExpireTask(key)
        timewheel.At(expireTime, taskKey, func() {
                keys := []string{key}
                db.RWLocks(keys, nil)
                defer db.RWUnLocks(keys, nil)
                // check-lock-check, ttl may be updated during waiting lock
                logger.Info("expire " + key)
                rawExpireTime, ok := db.ttlMap.Get(key)
                if !ok {
                        return
                }
                expireTime, _ := rawExpireTime.(time.Time)
                expired := time.Now().After(expireTime)
                if expired {
                        db.Remove(key)
                }
        })
}

// Persist cancel ttlCmd of key
func (db *DB) Persist(key string) {
        db.ttlMap.Remove(key)
        taskKey := genExpireTask(key)
        timewheel.Cancel(taskKey)
}

// IsExpired check whether a key is expired
func (db *DB) IsExpired(key string) bool {
        rawExpireTime, ok := db.ttlMap.Get(key)
        if !ok {
                return false
        }
        expireTime, _ := rawExpireTime.(time.Time)
        expired := time.Now().After(expireTime)
        if expired {
                db.Remove(key)
        }
        return expired
}

/* --- add version --- */

func (db *DB) addVersion(keys ...string) {
        for _, key := range keys {
                versionCode := db.GetVersion(key)
                db.versionMap.Put(key, versionCode+1)
        }
}

// GetVersion returns version code for given key
func (db *DB) GetVersion(key string) uint32 {
        entity, ok := db.versionMap.Get(key)
        if !ok {
                return 0
        }
        return entity.(uint32)
}

// ForEach traverses all the keys in the database
func (db *DB) ForEach(cb func(key string, data *database.DataEntity, expiration *time.Time) bool) {
        db.data.ForEach(func(key string, raw interface{}) bool {
                entity, _ := raw.(*database.DataEntity)
                var expiration *time.Time
                rawExpireTime, ok := db.ttlMap.Get(key)
                if ok {
                        expireTime, _ := rawExpireTime.(time.Time)
                        expiration = &expireTime
                }

                return cb(key, entity, expiration)
        })
}

1	// Package database is a memory database with redis compatible interface
2	package database
3
4	import (
5	"strings"
6	"time"
7
8	"github.com/hdt3213/godis/datastruct/dict"
9	"github.com/hdt3213/godis/interface/database"
10	"github.com/hdt3213/godis/interface/redis"
11	"github.com/hdt3213/godis/lib/logger"
12	"github.com/hdt3213/godis/lib/timewheel"
13	"github.com/hdt3213/godis/redis/protocol"
14	)
15
16	const (
17	dataDictSize = 1 << 16
18	ttlDictSize = 1 << 10
19	)
20
21	// DB stores data and execute user's commands
22	type DB struct {
23	index int
24	// key -> DataEntity
25	data *dict.ConcurrentDict
26	// key -> expireTime (time.Time)
27	ttlMap *dict.ConcurrentDict
28	// key -> version(uint32)
29	versionMap *dict.ConcurrentDict
30
31	// addaof is used to add command to aof
32	addAof func(CmdLine)
33
34	// callbacks
35	insertCallback database.KeyEventCallback
36	deleteCallback database.KeyEventCallback
37	}
38
39	// ExecFunc is interface for command executor
40	// args don't include cmd line
41	type ExecFunc func(db *DB, args [][]byte) redis.Reply
42
43	// PreFunc analyses command line when queued command to `multi`
44	// returns related write keys and read keys
45	type PreFunc func(args [][]byte) ([]string, []string)
46
47	// CmdLine is alias for [][]byte, represents a command line
48	type CmdLine = [][]byte
49
50	// UndoFunc returns undo logs for the given command line
51	// execute from head to tail when undo
52	type UndoFunc func(db *DB, args [][]byte) []CmdLine
53
54	// makeDB create DB instance
55	func makeDB() *DB {	1✔
56	db := &DB{	1✔
57	data: dict.MakeConcurrent(dataDictSize),	1✔
58	ttlMap: dict.MakeConcurrent(ttlDictSize),	1✔
59	versionMap: dict.MakeConcurrent(dataDictSize),	1✔
60	addAof: func(line CmdLine) {},	2✔
61	}
62	return db	1✔
63	}
64
65	// makeBasicDB create DB instance only with basic abilities.
66	func makeBasicDB() *DB {	1✔
67	db := &DB{	1✔
68	data: dict.MakeConcurrent(dataDictSize),	1✔
69	ttlMap: dict.MakeConcurrent(ttlDictSize),	1✔
70	versionMap: dict.MakeConcurrent(dataDictSize),	1✔
71	addAof: func(line CmdLine) {},	2✔
72	}
73	return db	1✔
74	}
75
76	// Exec executes command within one database
77	func (db *DB) Exec(c redis.Connection, cmdLine [][]byte) redis.Reply {	1✔
78	// transaction control commands and other commands which cannot execute within transaction	1✔
79	cmdName := strings.ToLower(string(cmdLine[0]))	1✔
80	if cmdName == "multi" {	2✔
81	if len(cmdLine) != 1 {	1✔
82	return protocol.MakeArgNumErrReply(cmdName)	×
83	}	×
84	return StartMulti(c)	1✔
85	} else if cmdName == "discard" {	2✔
86	if len(cmdLine) != 1 {	1✔
87	return protocol.MakeArgNumErrReply(cmdName)	×
88	}	×
89	return DiscardMulti(c)	1✔
90	} else if cmdName == "exec" {	2✔
91	if len(cmdLine) != 1 {	1✔
92	return protocol.MakeArgNumErrReply(cmdName)	×
93	}	×
94	return execMulti(db, c)	1✔
95	} else if cmdName == "watch" {	2✔
96	if !validateArity(-2, cmdLine) {	1✔
97	return protocol.MakeArgNumErrReply(cmdName)	×
98	}	×
99	return Watch(db, c, cmdLine[1:])	1✔
100	}
101	if c != nil && c.InMultiState() {	2✔
102	return EnqueueCmd(c, cmdLine)	1✔
103	}	1✔
104
105	return db.execNormalCommand(cmdLine)	1✔
106	}
107
108	func (db *DB) execNormalCommand(cmdLine [][]byte) redis.Reply {	1✔
109	cmdName := strings.ToLower(string(cmdLine[0]))	1✔
110	cmd, ok := cmdTable[cmdName]	1✔
111	if !ok {	2✔
112	return protocol.MakeErrReply("ERR unknown command '" + cmdName + "'")	1✔
113	}	1✔
114	if !validateArity(cmd.arity, cmdLine) {	2✔
115	return protocol.MakeArgNumErrReply(cmdName)	1✔
116	}	1✔
117
118	prepare := cmd.prepare	1✔
119	write, read := prepare(cmdLine[1:])	1✔
120	db.addVersion(write...)	1✔
121	db.RWLocks(write, read)	1✔
122	defer db.RWUnLocks(write, read)	1✔
123	fun := cmd.executor	1✔
124	return fun(db, cmdLine[1:])	1✔
125	}
126
127	// execWithLock executes normal commands, invoker should provide locks
128	func (db *DB) execWithLock(cmdLine [][]byte) redis.Reply {	1✔
129	cmdName := strings.ToLower(string(cmdLine[0]))	1✔
130	cmd, ok := cmdTable[cmdName]	1✔
131	if !ok {	1✔
132	return protocol.MakeErrReply("ERR unknown command '" + cmdName + "'")	×
133	}	×
134	if !validateArity(cmd.arity, cmdLine) {	1✔
135	return protocol.MakeArgNumErrReply(cmdName)	×
136	}	×
137	fun := cmd.executor	1✔
138	return fun(db, cmdLine[1:])	1✔
139	}
140
141	func validateArity(arity int, cmdArgs [][]byte) bool {	1✔
142	argNum := len(cmdArgs)	1✔
143	if arity >= 0 {	2✔
144	return argNum == arity	1✔
145	}	1✔
146	return argNum >= -arity	1✔
147	}
148
149	/* ---- Data Access ----- */
150
151	// GetEntity returns DataEntity bind to given key
152	func (db DB) GetEntity(key string) (database.DataEntity, bool) {	1✔
153	raw, ok := db.data.GetWithLock(key)	1✔
154	if !ok {	2✔
155	return nil, false	1✔
156	}	1✔
157	if db.IsExpired(key) {	2✔
158	return nil, false	1✔
159	}	1✔
160	entity, _ := raw.(*database.DataEntity)	1✔
161	return entity, true	1✔
162	}
163
164	// PutEntity a DataEntity into DB
165	func (db DB) PutEntity(key string, entity database.DataEntity) int {	1✔
166	ret := db.data.PutWithLock(key, entity)	1✔
167	// db.insertCallback may be set as nil, during `if` and actually callback	1✔
168	// so introduce a local variable `cb`	1✔
169	if cb := db.insertCallback; ret > 0 && cb != nil {	1✔
170	cb(db.index, key, entity)	×
171	}	×
172	return ret	1✔
173	}
174
175	// PutIfExists edit an existing DataEntity
176	func (db DB) PutIfExists(key string, entity database.DataEntity) int {	1✔
177	return db.data.PutIfExistsWithLock(key, entity)	1✔
178	}	1✔
179
180	// PutIfAbsent insert an DataEntity only if the key not exists
181	func (db DB) PutIfAbsent(key string, entity database.DataEntity) int {	1✔
182	ret := db.data.PutIfAbsentWithLock(key, entity)	1✔
183	// db.insertCallback may be set as nil, during `if` and actually callback	1✔
184	// so introduce a local variable `cb`	1✔
185	if cb := db.insertCallback; ret > 0 && cb != nil {	1✔
186	cb(db.index, key, entity)	×
187	}	×
188	return ret	1✔
189	}
190
191	// Remove the given key from db
192	func (db *DB) Remove(key string) {	1✔
193	raw, deleted := db.data.RemoveWithLock(key)	1✔
194	db.ttlMap.Remove(key)	1✔
195	taskKey := genExpireTask(key)	1✔
196	timewheel.Cancel(taskKey)	1✔
197	if cb := db.deleteCallback; cb != nil {	1✔
198	var entity *database.DataEntity	×
199	if deleted > 0 {	×
200	entity = raw.(*database.DataEntity)	×
201	}	×
202	cb(db.index, key, entity)	×
203	}
204	}
205
206	// Removes the given keys from db
207	func (db *DB) Removes(keys ...string) (deleted int) {	1✔
208	deleted = 0	1✔
209	for _, key := range keys {	2✔
210	_, exists := db.data.GetWithLock(key)	1✔
211	if exists {	2✔
212	db.Remove(key)	1✔
213	deleted++	1✔
214	}	1✔
215	}
216	return deleted	1✔
217	}
218
219	// Flush clean database
220	// deprecated
221	// for test only
222	func (db *DB) Flush() {	1✔
223	db.data.Clear()	1✔
224	db.ttlMap.Clear()	1✔
225	}	1✔
226
227	/* ---- Lock Function ----- */
228
229	// RWLocks lock keys for writing and reading
230	func (db *DB) RWLocks(writeKeys []string, readKeys []string) {	1✔
231	db.data.RWLocks(writeKeys, readKeys)	1✔
232	}	1✔
233
234	// RWUnLocks unlock keys for writing and reading
235	func (db *DB) RWUnLocks(writeKeys []string, readKeys []string) {	1✔
236	db.data.RWUnLocks(writeKeys, readKeys)	1✔
237	}	1✔
238
239	/* ---- TTL Functions ---- */
240
241	func genExpireTask(key string) string {	1✔
242	return "expire:" + key	1✔
243	}	1✔
244
245	// Expire sets ttlCmd of key
246	func (db *DB) Expire(key string, expireTime time.Time) {	1✔
247	db.ttlMap.Put(key, expireTime)	1✔
248	taskKey := genExpireTask(key)	1✔
249	timewheel.At(expireTime, taskKey, func() {	2✔
250	keys := []string{key}	1✔
251	db.RWLocks(keys, nil)	1✔
252	defer db.RWUnLocks(keys, nil)	1✔
253	// check-lock-check, ttl may be updated during waiting lock	1✔
254	logger.Info("expire " + key)	1✔
255	rawExpireTime, ok := db.ttlMap.Get(key)	1✔
256	if !ok {	1✔
257	return	×
258	}	×
259	expireTime, _ := rawExpireTime.(time.Time)	1✔
260	expired := time.Now().After(expireTime)	1✔
261	if expired {	1✔
UNCOV 262	db.Remove(key)	×
UNCOV 263	}	×
264	})
265	}
266
267	// Persist cancel ttlCmd of key
268	func (db *DB) Persist(key string) {	1✔
269	db.ttlMap.Remove(key)	1✔
270	taskKey := genExpireTask(key)	1✔
271	timewheel.Cancel(taskKey)	1✔
272	}	1✔
273
274	// IsExpired check whether a key is expired
275	func (db *DB) IsExpired(key string) bool {	1✔
276	rawExpireTime, ok := db.ttlMap.Get(key)	1✔
277	if !ok {	2✔
278	return false	1✔
279	}	1✔
280	expireTime, _ := rawExpireTime.(time.Time)	1✔
281	expired := time.Now().After(expireTime)	1✔
282	if expired {	2✔
283	db.Remove(key)	1✔
284	}	1✔
285	return expired	1✔
286	}
287
288	/* --- add version --- */
289
290	func (db *DB) addVersion(keys ...string) {	1✔
291	for _, key := range keys {	2✔
292	versionCode := db.GetVersion(key)	1✔
293	db.versionMap.Put(key, versionCode+1)	1✔
294	}	1✔
295	}
296
297	// GetVersion returns version code for given key
298	func (db *DB) GetVersion(key string) uint32 {	1✔
299	entity, ok := db.versionMap.Get(key)	1✔
300	if !ok {	2✔
301	return 0	1✔
302	}	1✔
303	return entity.(uint32)	1✔
304	}
305
306	// ForEach traverses all the keys in the database
307	func (db DB) ForEach(cb func(key string, data database.DataEntity, expiration *time.Time) bool) {	1✔
308	db.data.ForEach(func(key string, raw interface{}) bool {	2✔
309	entity, _ := raw.(*database.DataEntity)	1✔
310	var expiration *time.Time	1✔
311	rawExpireTime, ok := db.ttlMap.Get(key)	1✔
312	if ok {	2✔
313	expireTime, _ := rawExpireTime.(time.Time)	1✔
314	expiration = &expireTime	1✔
315	}	1✔
316
317	return cb(key, entity, expiration)	1✔
318	})
319	}

HDT3213 / godis / 17582991793

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous