-
Notifications
You must be signed in to change notification settings - Fork 86
/
db.go
212 lines (180 loc) · 3.82 KB
/
db.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
package minidb
import (
"io"
"os"
"sync"
)
type MiniDB struct {
indexes map[string]int64 // 内存中的索引信息
dbFile *DBFile // 数据文件
dirPath string // 数据目录
mu sync.RWMutex
}
// Open 开启一个数据库实例
func Open(dirPath string) (*MiniDB, error) {
// 如果数据库目录不存在,则新建一个
if _, err := os.Stat(dirPath); os.IsNotExist(err) {
if err := os.MkdirAll(dirPath, os.ModePerm); err != nil {
return nil, err
}
}
// 加载数据文件
dbFile, err := NewDBFile(dirPath)
if err != nil {
return nil, err
}
db := &MiniDB{
dbFile: dbFile,
indexes: make(map[string]int64),
dirPath: dirPath,
}
// 加载索引
db.loadIndexesFromFile()
return db, nil
}
// Merge 合并数据文件,在rosedb当中是 Reclaim 方法
func (db *MiniDB) Merge() error {
// 没有数据,忽略
if db.dbFile.Offset == 0 {
return nil
}
var (
validEntries []*Entry
offset int64
)
// 读取原数据文件中的 Entry
for {
e, err := db.dbFile.Read(offset)
if err != nil {
if err == io.EOF {
break
}
return err
}
// 内存中的索引状态是最新的,直接对比过滤出有效的 Entry
if off, ok := db.indexes[string(e.Key)]; ok && off == offset {
validEntries = append(validEntries, e)
}
offset += e.GetSize()
}
if len(validEntries) > 0 {
// 新建临时文件
mergeDBFile, err := NewMergeDBFile(db.dirPath)
if err != nil {
return err
}
defer os.Remove(mergeDBFile.File.Name())
// 重新写入有效的 entry
for _, entry := range validEntries {
writeOff := mergeDBFile.Offset
err := mergeDBFile.Write(entry)
if err != nil {
return err
}
// 更新索引
db.indexes[string(entry.Key)] = writeOff
}
// 获取文件名
dbFileName := db.dbFile.File.Name()
// 关闭文件
db.dbFile.File.Close()
// 删除旧的数据文件
os.Remove(dbFileName)
// 获取文件名
mergeDBFileName := mergeDBFile.File.Name()
// 关闭文件
mergeDBFile.File.Close()
// 临时文件变更为新的数据文件
os.Rename(mergeDBFileName, db.dirPath+string(os.PathSeparator)+FileName)
db.dbFile = mergeDBFile
}
return nil
}
// Put 写入数据
func (db *MiniDB) Put(key []byte, value []byte) (err error) {
if len(key) == 0 {
return
}
db.mu.Lock()
defer db.mu.Unlock()
offset := db.dbFile.Offset
// 封装成 Entry
entry := NewEntry(key, value, PUT)
// 追加到数据文件当中
err = db.dbFile.Write(entry)
// 写到内存
db.indexes[string(key)] = offset
return
}
// Get 取出数据
func (db *MiniDB) Get(key []byte) (val []byte, err error) {
if len(key) == 0 {
return
}
db.mu.RLock()
defer db.mu.RUnlock()
// 从内存当中取出索引信息
offset, ok := db.indexes[string(key)]
// key 不存在
if !ok {
return
}
// 从磁盘中读取数据
var e *Entry
e, err = db.dbFile.Read(offset)
if err != nil && err != io.EOF {
return
}
if e != nil {
val = e.Value
}
return
}
// Del 删除数据
func (db *MiniDB) Del(key []byte) (err error) {
if len(key) == 0 {
return
}
db.mu.Lock()
defer db.mu.Unlock()
// 从内存当中取出索引信息
_, ok := db.indexes[string(key)]
// key 不存在,忽略
if !ok {
return
}
// 封装成 Entry 并写入
e := NewEntry(key, nil, DEL)
err = db.dbFile.Write(e)
if err != nil {
return
}
// 删除内存中的 key
delete(db.indexes, string(key))
return
}
// 从文件当中加载索引
func (db *MiniDB) loadIndexesFromFile() {
if db.dbFile == nil {
return
}
var offset int64
for {
e, err := db.dbFile.Read(offset)
if err != nil {
// 读取完毕
if err == io.EOF {
break
}
return
}
// 设置索引状态
db.indexes[string(e.Key)] = offset
if e.Mark == DEL {
// 删除内存中的 key
delete(db.indexes, string(e.Key))
}
offset += e.GetSize()
}
return
}