@@ -8,10 +8,12 @@ import (
8
8
"path/filepath"
9
9
"strings"
10
10
"sync"
11
+ "sync/atomic"
11
12
12
13
"git.tcp.direct/Mirrors/bitcask-mirror"
13
14
14
15
"git.tcp.direct/tcp.direct/database"
16
+ "git.tcp.direct/tcp.direct/database/kv"
15
17
"git.tcp.direct/tcp.direct/database/metadata"
16
18
"git.tcp.direct/tcp.direct/database/models"
17
19
)
@@ -20,23 +22,43 @@ import (
20
22
type Store struct {
21
23
* bitcask.Bitcask
22
24
database.Searcher
23
- closed bool
25
+ closed * atomic.Bool
26
+ }
27
+
28
+ // Get is a wrapper around the bitcask Get function for error regularization.
29
+ func (s * Store ) Get (key []byte ) ([]byte , error ) {
30
+ if s .closed .Load () {
31
+ return nil , fs .ErrClosed
32
+ }
33
+ ret , err := s .Bitcask .Get (key )
34
+ err = kv .RegularizeKVError (key , ret , err )
35
+ return ret , err
36
+ }
37
+
38
+ // Close is a wrapper around the bitcask Close function.
39
+ func (s * Store ) Close () error {
40
+ if s .closed .Load () {
41
+ return fs .ErrClosed
42
+ }
43
+ s .closed .Store (true )
44
+ return s .Bitcask .Close ()
24
45
}
25
46
26
47
// Backend returns the underlying bitcask instance.
27
- func (s Store ) Backend () any {
48
+ func (s * Store ) Backend () any {
28
49
return s .Bitcask
29
50
}
30
51
31
52
// DB is a mapper of a Filer and Searcher implementation using Bitcask.
32
53
type DB struct {
33
- store map [string ]Store
54
+ store map [string ]* Store
34
55
path string
35
56
mu * sync.RWMutex
36
57
meta * metadata.Metadata
37
- initialized bool
58
+ initialized * atomic. Bool
38
59
}
39
60
61
+ // Meta returns the [models.Metadata] implementation of the bitcask keeper.
40
62
func (db * DB ) Meta () models.Metadata {
41
63
db .mu .RLock ()
42
64
m := db .meta
@@ -60,12 +82,9 @@ func (db *DB) AllStores() map[string]database.Filer {
60
82
61
83
func (db * DB ) init () error {
62
84
var err error
63
- db .mu .RLock ()
64
- if db .initialized {
65
- db .mu .RUnlock ()
85
+ if db .initialized .Load () {
66
86
return nil
67
87
}
68
- db .mu .RUnlock ()
69
88
db .mu .Lock ()
70
89
defer db .mu .Unlock ()
71
90
if _ , err = os .Stat (db .path ); os .IsNotExist (err ) {
@@ -86,7 +105,7 @@ func (db *DB) init() error {
86
105
if db .meta .Type () != db .Type () {
87
106
return fmt .Errorf ("meta.json is not a bitcask meta file" )
88
107
}
89
- db .initialized = true
108
+ db .initialized . Store ( true )
90
109
return nil
91
110
}
92
111
@@ -95,7 +114,7 @@ func (db *DB) init() error {
95
114
if err != nil {
96
115
return fmt .Errorf ("error creating meta file: %w" , err )
97
116
}
98
- db .initialized = true
117
+ db .initialized . Store ( true )
99
118
return nil
100
119
}
101
120
@@ -104,12 +123,14 @@ func (db *DB) init() error {
104
123
105
124
// OpenDB will either open an existing set of bitcask datastores at the given directory, or it will create a new one.
106
125
func OpenDB (path string ) * DB {
126
+ ainit := & atomic.Bool {}
127
+ ainit .Store (false )
107
128
return & DB {
108
- store : make (map [string ]Store ),
129
+ store : make (map [string ]* Store ),
109
130
path : path ,
110
131
mu : & sync.RWMutex {},
111
132
meta : nil ,
112
- initialized : false ,
133
+ initialized : ainit ,
113
134
}
114
135
}
115
136
@@ -123,9 +144,8 @@ func (db *DB) Discover() ([]string, error) {
123
144
stores := make ([]string , 0 )
124
145
errs := make ([]error , 0 )
125
146
if db .store == nil {
126
- db .store = make (map [string ]Store )
147
+ db .store = make (map [string ]* Store )
127
148
}
128
- os .Stat (filepath .Join (db .path , "meta.json" ))
129
149
130
150
entries , err := fs .ReadDir (os .DirFS (db .path ), "." )
131
151
if err != nil {
@@ -157,17 +177,19 @@ func (db *DB) Discover() ([]string, error) {
157
177
}
158
178
println ("WARN: bitcask store" , name , "has bad metadata, attempting to repair" )
159
179
oldMeta := filepath .Join (db .path , name , "meta.json" )
160
- newMeta := filepath .Join (db .path , name , "meta.json.backup" )
161
- println ("WARN: renaming" , oldMeta , "to" , newMeta )
162
- // likely defunct lockfile is present too, remove it
163
- if osErr := os .Rename (oldMeta , newMeta ); osErr != nil {
164
- println ("WARN: failed to rename" , oldMeta , "to" , newMeta , ":" , osErr )
165
- return
180
+ oldMetaBackup := filepath .Join (db .path , name , "meta.json.backup" )
181
+ println ("WARN: renaming" , oldMeta , "to" , oldMetaBackup )
182
+ if osErr := os .Rename (oldMeta , oldMetaBackup ); osErr != nil {
183
+ println ("Fatal: failed to rename" , oldMeta , "to" , oldMetaBackup , ":" , osErr )
184
+ panic (osErr )
166
185
}
186
+
187
+ // likely defunct lockfile is present too, remove it
167
188
if _ , serr := os .Stat (filepath .Join (db .path , name , "lock" )); serr == nil {
168
189
println ("WARN: removing defunct lockfile" )
169
190
_ = os .Remove (filepath .Join (db .path , name , "lock" ))
170
191
}
192
+
171
193
retry = true
172
194
})
173
195
if retry {
@@ -176,7 +198,9 @@ func (db *DB) Discover() ([]string, error) {
176
198
errs = append (errs , e )
177
199
continue
178
200
}
179
- db .store [name ] = Store {Bitcask : c }
201
+ aclosed := & atomic.Bool {}
202
+ aclosed .Store (false )
203
+ db .store [name ] = & Store {Bitcask : c , closed : aclosed }
180
204
stores = append (stores , name )
181
205
}
182
206
@@ -225,44 +249,82 @@ func (db *DB) Init(storeName string, opts ...any) error {
225
249
}
226
250
var bitcaskopts []bitcask.Option
227
251
for _ , opt := range opts {
228
- if _ , ok := opt .(bitcask.Option ); ! ok {
229
- return errors .New ("invalid bitcask option type" )
252
+ _ , isOptOK := opt .(bitcask.Option )
253
+ _ , isOptsOK := opt .([]bitcask.Option )
254
+ if ! isOptOK && ! isOptsOK {
255
+ return fmt .Errorf ("invalid bitcask option type (%T): %v" , opt , opt )
256
+ }
257
+ if isOptOK {
258
+ bitcaskopts = append (bitcaskopts , opt .(bitcask.Option ))
259
+ }
260
+ if isOptsOK {
261
+ bitcaskopts = append (bitcaskopts , opt .([]bitcask.Option )... )
230
262
}
231
- bitcaskopts = append (bitcaskopts , opt .(bitcask.Option ))
232
263
}
233
- db .mu .Lock ()
234
- defer db .mu .Unlock ()
235
264
236
265
if len (defaultBitcaskOptions ) > 0 {
237
- bitcaskopts = append (bitcaskopts , defaultBitcaskOptions ... )
266
+ bitcaskopts = append (defaultBitcaskOptions , bitcaskopts ... )
238
267
}
239
268
269
+ db .mu .Lock ()
270
+ err := db .initStore (storeName , bitcaskopts ... )
271
+ db .mu .Unlock ()
272
+
273
+ return err
274
+ }
275
+
276
+ // initStore is a helper function to initialize a bitcask store, caller must hold keeper's lock.
277
+ func (db * DB ) initStore (storeName string , opts ... bitcask.Option ) error {
240
278
if _ , ok := db .store [storeName ]; ok {
241
279
return ErrStoreExists
242
280
}
243
- path := db .Path ()
244
- if ! strings .HasSuffix (db .Path (), "/" ) {
245
- path = db .Path () + "/"
246
- }
247
- c , e := bitcask .Open (path + storeName , bitcaskopts ... )
281
+
282
+ c , e := bitcask .Open (filepath .Join (db .Path (), storeName ), opts ... )
248
283
if e != nil {
249
284
return e
250
285
}
251
- db .store [storeName ] = Store {Bitcask : c }
286
+
287
+ aclosed := & atomic.Bool {}
288
+ aclosed .Store (false )
289
+ db .store [storeName ] = & Store {Bitcask : c , closed : aclosed }
252
290
return nil
253
291
}
254
292
293
+ // Destroy will remove the bitcask store and all data associated with it.
294
+ func (db * DB ) Destroy (storeName string ) error {
295
+ db .mu .Lock ()
296
+ defer db .mu .Unlock ()
297
+ st , ok := db .store [storeName ]
298
+ if ! ok {
299
+ return ErrBogusStore
300
+ }
301
+ err := st .Close ()
302
+ if err != nil {
303
+ return err
304
+ }
305
+ delete (db .store , storeName )
306
+ return os .RemoveAll (filepath .Join (db .path , storeName ))
307
+ }
308
+
255
309
// With calls the given underlying bitcask instance.
256
310
func (db * DB ) With (storeName string ) database.Store {
257
311
if err := db .init (); err != nil {
258
312
panic (err )
259
313
}
260
314
db .mu .RLock ()
261
- defer db .mu .RUnlock ()
262
315
d , ok := db .store [storeName ]
263
- if ok {
316
+ if ok && ! d .closed .Load () {
317
+ db .mu .RUnlock ()
264
318
return d
265
319
}
320
+ if ok && d .closed .Load () {
321
+ db .mu .RUnlock ()
322
+ db .mu .Lock ()
323
+ delete (db .store , storeName )
324
+ db .mu .Unlock ()
325
+ return nil
326
+ }
327
+ db .mu .RUnlock ()
266
328
return nil
267
329
}
268
330
@@ -271,42 +333,54 @@ func (db *DB) WithNew(storeName string, opts ...any) database.Filer {
271
333
if err := db .init (); err != nil {
272
334
panic (err )
273
335
}
274
- db . mu . RLock ()
275
- defer db . mu . RUnlock ( )
336
+
337
+ newOpts := make ([]bitcask. Option , 0 )
276
338
for _ , opt := range opts {
277
339
if _ , ok := opt .(bitcask.Option ); ! ok {
278
340
fmt .Println ("invalid bitcask option type: " , opt )
279
341
continue
280
342
}
281
- defaultBitcaskOptions = append (defaultBitcaskOptions , opt .(bitcask.Option ))
343
+ newOpts = append (newOpts , opt .(bitcask.Option ))
282
344
}
345
+
346
+ db .mu .Lock ()
347
+ defer db .mu .Unlock ()
348
+
283
349
d , ok := db .store [storeName ]
284
350
if ok {
351
+ if d .Bitcask == nil || d .closed == nil || d .closed .Load () {
352
+ delete (db .store , storeName )
353
+ if err := db .initStore (storeName , newOpts ... ); err != nil {
354
+ fmt .Println ("error re-initializing bitcask store: " , err .Error ())
355
+ }
356
+ return db .store [storeName ]
357
+ }
285
358
return d
286
359
}
287
- db .mu .RUnlock ()
288
- err := db .Init (storeName )
289
- db .mu .RLock ()
360
+
361
+ err := db .initStore (storeName , newOpts ... )
290
362
if err != nil {
291
363
fmt .Println ("error creating bitcask store: " , err )
292
-
293
364
}
294
365
return db .store [storeName ]
295
366
}
296
367
297
368
// Close is a simple shim for bitcask's Close function.
298
369
func (db * DB ) Close (storeName string ) error {
299
- db .mu .Lock ()
300
- defer db .mu .Unlock ()
370
+ db .mu .RLock ()
301
371
st , ok := db .store [storeName ]
302
372
if ! ok {
373
+ db .mu .RUnlock ()
303
374
return ErrBogusStore
304
375
}
376
+ db .mu .RUnlock ()
305
377
err := st .Close ()
306
378
if err != nil {
307
379
return err
308
380
}
381
+ db .mu .Lock ()
309
382
delete (db .store , storeName )
383
+ db .mu .Unlock ()
310
384
return nil
311
385
}
312
386
@@ -334,7 +408,10 @@ const (
334
408
// In the case of an error, withAll will continue and return a compound form of any errors that occurred.
335
409
// For now this is just for Close and Sync, thusly it does a hard lock on the Keeper.
336
410
func (db * DB ) withAll (action withAllAction ) error {
337
- if db == nil || db .store == nil || len (db .store ) < 1 {
411
+ if db == nil || db .store == nil {
412
+ panic ("bitcask: nil db or db.store" )
413
+ }
414
+ if len (db .store ) < 1 {
338
415
return ErrNoStores
339
416
}
340
417
var errs = make ([]error , len (db .store ))
@@ -344,13 +421,19 @@ func (db *DB) withAll(action withAllAction) error {
344
421
errs = append (errs , namedErr (name , ErrBogusStore ))
345
422
continue
346
423
}
424
+ if store .closed .Load () {
425
+ continue
426
+ }
347
427
switch action {
348
428
case dclose :
429
+ db .mu .Lock ()
349
430
closeErr := store .Close ()
350
431
if errors .Is (closeErr , fs .ErrClosed ) {
351
432
continue
352
433
}
353
434
err = namedErr (name , closeErr )
435
+ delete (db .store , name )
436
+ db .mu .Unlock ()
354
437
case dsync :
355
438
err = namedErr (name , store .Sync ())
356
439
default :
@@ -391,6 +474,8 @@ func (db *DB) SyncAll() error {
391
474
return db .withAll (dsync )
392
475
}
393
476
477
+ // Type returns the type of keeper, in this case "bitcask".
478
+ // This is in order to implement [database.Keeper].
394
479
func (db * DB ) Type () string {
395
480
return "bitcask"
396
481
}
0 commit comments