Browse Source
weed volume: remove boltdb, btree options, add options for leveldb medium, large
pull/925/head
weed volume: remove boltdb, btree options, add options for leveldb medium, large
pull/925/head
Chris Lu
6 years ago
6 changed files with 34 additions and 205 deletions
-
2weed/command/server.go
-
8weed/command/volume.go
-
6weed/storage/needle_map.go
-
185weed/storage/needle_map_boltdb.go
-
7weed/storage/needle_map_leveldb.go
-
31weed/storage/volume_loading.go
@ -1,185 +0,0 @@ |
|||||
package storage |
|
||||
|
|
||||
import ( |
|
||||
"fmt" |
|
||||
"os" |
|
||||
|
|
||||
"github.com/boltdb/bolt" |
|
||||
|
|
||||
"errors" |
|
||||
"github.com/chrislusf/seaweedfs/weed/glog" |
|
||||
"github.com/chrislusf/seaweedfs/weed/storage/needle" |
|
||||
. "github.com/chrislusf/seaweedfs/weed/storage/types" |
|
||||
"github.com/chrislusf/seaweedfs/weed/util" |
|
||||
) |
|
||||
|
|
||||
type BoltDbNeedleMap struct { |
|
||||
dbFileName string |
|
||||
db *bolt.DB |
|
||||
baseNeedleMapper |
|
||||
} |
|
||||
|
|
||||
var boltdbBucket = []byte("weed") |
|
||||
|
|
||||
var NotFound = errors.New("not found") |
|
||||
|
|
||||
func NewBoltDbNeedleMap(dbFileName string, indexFile *os.File) (m *BoltDbNeedleMap, err error) { |
|
||||
m = &BoltDbNeedleMap{dbFileName: dbFileName} |
|
||||
m.indexFile = indexFile |
|
||||
if !isBoltDbFresh(dbFileName, indexFile) { |
|
||||
glog.V(0).Infof("Start to Generate %s from %s", dbFileName, indexFile.Name()) |
|
||||
generateBoltDbFile(dbFileName, indexFile) |
|
||||
glog.V(0).Infof("Finished Generating %s from %s", dbFileName, indexFile.Name()) |
|
||||
} |
|
||||
glog.V(1).Infof("Opening %s...", dbFileName) |
|
||||
if m.db, err = bolt.Open(dbFileName, 0644, nil); err != nil { |
|
||||
return |
|
||||
} |
|
||||
glog.V(1).Infof("Loading %s...", indexFile.Name()) |
|
||||
mm, indexLoadError := newNeedleMapMetricFromIndexFile(indexFile) |
|
||||
if indexLoadError != nil { |
|
||||
return nil, indexLoadError |
|
||||
} |
|
||||
m.mapMetric = *mm |
|
||||
return |
|
||||
} |
|
||||
|
|
||||
func isBoltDbFresh(dbFileName string, indexFile *os.File) bool { |
|
||||
// normally we always write to index file first
|
|
||||
dbLogFile, err := os.Open(dbFileName) |
|
||||
if err != nil { |
|
||||
return false |
|
||||
} |
|
||||
defer dbLogFile.Close() |
|
||||
dbStat, dbStatErr := dbLogFile.Stat() |
|
||||
indexStat, indexStatErr := indexFile.Stat() |
|
||||
if dbStatErr != nil || indexStatErr != nil { |
|
||||
glog.V(0).Infof("Can not stat file: %v and %v", dbStatErr, indexStatErr) |
|
||||
return false |
|
||||
} |
|
||||
|
|
||||
return dbStat.ModTime().After(indexStat.ModTime()) |
|
||||
} |
|
||||
|
|
||||
func generateBoltDbFile(dbFileName string, indexFile *os.File) error { |
|
||||
db, err := bolt.Open(dbFileName, 0644, nil) |
|
||||
if err != nil { |
|
||||
return err |
|
||||
} |
|
||||
defer db.Close() |
|
||||
return WalkIndexFile(indexFile, func(key NeedleId, offset Offset, size uint32) error { |
|
||||
if !offset.IsZero() && size != TombstoneFileSize { |
|
||||
boltDbWrite(db, key, offset, size) |
|
||||
} else { |
|
||||
boltDbDelete(db, key) |
|
||||
} |
|
||||
return nil |
|
||||
}) |
|
||||
} |
|
||||
|
|
||||
func (m *BoltDbNeedleMap) Get(key NeedleId) (element *needle.NeedleValue, ok bool) { |
|
||||
var offset Offset |
|
||||
var size uint32 |
|
||||
bytes := make([]byte, NeedleIdSize) |
|
||||
NeedleIdToBytes(bytes, key) |
|
||||
err := m.db.View(func(tx *bolt.Tx) error { |
|
||||
bucket := tx.Bucket(boltdbBucket) |
|
||||
if bucket == nil { |
|
||||
return fmt.Errorf("Bucket %q not found!", boltdbBucket) |
|
||||
} |
|
||||
|
|
||||
data := bucket.Get(bytes) |
|
||||
|
|
||||
if len(data) == 0 { |
|
||||
return NotFound |
|
||||
} |
|
||||
|
|
||||
if len(data) != OffsetSize+SizeSize { |
|
||||
glog.V(0).Infof("key:%v has wrong data length: %d", key, len(data)) |
|
||||
return fmt.Errorf("key:%v has wrong data length: %d", key, len(data)) |
|
||||
} |
|
||||
|
|
||||
offset = BytesToOffset(data[0:OffsetSize]) |
|
||||
size = util.BytesToUint32(data[OffsetSize : OffsetSize+SizeSize]) |
|
||||
|
|
||||
return nil |
|
||||
}) |
|
||||
|
|
||||
if err != nil { |
|
||||
return nil, false |
|
||||
} |
|
||||
return &needle.NeedleValue{Key: key, Offset: offset, Size: size}, true |
|
||||
} |
|
||||
|
|
||||
func (m *BoltDbNeedleMap) Put(key NeedleId, offset Offset, size uint32) error { |
|
||||
var oldSize uint32 |
|
||||
if oldNeedle, ok := m.Get(key); ok { |
|
||||
oldSize = oldNeedle.Size |
|
||||
} |
|
||||
m.logPut(key, oldSize, size) |
|
||||
// write to index file first
|
|
||||
if err := m.appendToIndexFile(key, offset, size); err != nil { |
|
||||
return fmt.Errorf("cannot write to indexfile %s: %v", m.indexFile.Name(), err) |
|
||||
} |
|
||||
return boltDbWrite(m.db, key, offset, size) |
|
||||
} |
|
||||
|
|
||||
func boltDbWrite(db *bolt.DB, |
|
||||
key NeedleId, offset Offset, size uint32) error { |
|
||||
|
|
||||
bytes := make([]byte, NeedleIdSize+OffsetSize+SizeSize) |
|
||||
NeedleIdToBytes(bytes[0:NeedleIdSize], key) |
|
||||
OffsetToBytes(bytes[NeedleIdSize:NeedleIdSize+OffsetSize], offset) |
|
||||
util.Uint32toBytes(bytes[NeedleIdSize+OffsetSize:NeedleIdSize+OffsetSize+SizeSize], size) |
|
||||
|
|
||||
return db.Update(func(tx *bolt.Tx) error { |
|
||||
bucket, err := tx.CreateBucketIfNotExists(boltdbBucket) |
|
||||
if err != nil { |
|
||||
return err |
|
||||
} |
|
||||
|
|
||||
err = bucket.Put(bytes[0:NeedleIdSize], bytes[NeedleIdSize:NeedleIdSize+OffsetSize+SizeSize]) |
|
||||
if err != nil { |
|
||||
return err |
|
||||
} |
|
||||
return nil |
|
||||
}) |
|
||||
} |
|
||||
func boltDbDelete(db *bolt.DB, key NeedleId) error { |
|
||||
bytes := make([]byte, NeedleIdSize) |
|
||||
NeedleIdToBytes(bytes, key) |
|
||||
return db.Update(func(tx *bolt.Tx) error { |
|
||||
bucket, err := tx.CreateBucketIfNotExists(boltdbBucket) |
|
||||
if err != nil { |
|
||||
return err |
|
||||
} |
|
||||
|
|
||||
err = bucket.Delete(bytes) |
|
||||
if err != nil { |
|
||||
return err |
|
||||
} |
|
||||
return nil |
|
||||
}) |
|
||||
} |
|
||||
|
|
||||
func (m *BoltDbNeedleMap) Delete(key NeedleId, offset Offset) error { |
|
||||
if oldNeedle, ok := m.Get(key); ok { |
|
||||
m.logDelete(oldNeedle.Size) |
|
||||
} |
|
||||
// write to index file first
|
|
||||
if err := m.appendToIndexFile(key, offset, TombstoneFileSize); err != nil { |
|
||||
return err |
|
||||
} |
|
||||
return boltDbDelete(m.db, key) |
|
||||
} |
|
||||
|
|
||||
func (m *BoltDbNeedleMap) Close() { |
|
||||
m.indexFile.Close() |
|
||||
m.db.Close() |
|
||||
} |
|
||||
|
|
||||
func (m *BoltDbNeedleMap) Destroy() error { |
|
||||
m.Close() |
|
||||
os.Remove(m.indexFile.Name()) |
|
||||
return os.Remove(m.dbFileName) |
|
||||
} |
|
||||
Write
Preview
Loading…
Cancel
Save
Reference in new issue