Chris Lu
4 years ago
8 changed files with 540 additions and 2 deletions
-
1go.mod
-
8go.sum
-
27weed/filer/leveldb/leveldb_store_test.go
-
38weed/filer/rocksdb/README.md
-
293weed/filer/rocksdb/rocksdb_store.go
-
51weed/filer/rocksdb/rocksdb_store_kv.go
-
117weed/filer/rocksdb/rocksdb_store_test.go
-
7weed/server/filer_server_rocksdb.go
@ -0,0 +1,38 @@ |
|||
# Prepare the compilation environment on linux |
|||
- sudo add-apt-repository -y ppa:ubuntu-toolchain-r/test |
|||
- sudo apt-get update -qq |
|||
- sudo apt-get install gcc-6 g++-6 libsnappy-dev zlib1g-dev libbz2-dev -qq |
|||
- export CXX="g++-6" CC="gcc-6" |
|||
|
|||
- wget https://launchpad.net/ubuntu/+archive/primary/+files/libgflags2_2.0-1.1ubuntu1_amd64.deb |
|||
- sudo dpkg -i libgflags2_2.0-1.1ubuntu1_amd64.deb |
|||
- wget https://launchpad.net/ubuntu/+archive/primary/+files/libgflags-dev_2.0-1.1ubuntu1_amd64.deb |
|||
- sudo dpkg -i libgflags-dev_2.0-1.1ubuntu1_amd64.deb |
|||
|
|||
# Prepare the compilation environment on mac os |
|||
``` |
|||
brew install snappy |
|||
``` |
|||
|
|||
# install rocksdb: |
|||
``` |
|||
|
|||
export ROCKSDB_HOME=/Users/chris/dev/rocksdb |
|||
|
|||
git clone https://github.com/facebook/rocksdb.git $ROCKSDB_HOME |
|||
pushd $ROCKSDB_HOME |
|||
make clean |
|||
make install-static |
|||
popd |
|||
|
|||
``` |
|||
|
|||
# install gorocksdb |
|||
|
|||
``` |
|||
export CGO_CFLAGS="-I$ROCKSDB_HOME/include" |
|||
export CGO_LDFLAGS="-L$ROCKSDB_HOME -lrocksdb -lstdc++ -lm -lz -lbz2 -lsnappy -llz4 -lzstd" |
|||
|
|||
go get github.com/tecbot/gorocksdb |
|||
|
|||
``` |
|||
@ -0,0 +1,293 @@ |
|||
// +build rocksdb
|
|||
|
|||
package rocksdb |
|||
|
|||
import ( |
|||
"bytes" |
|||
"context" |
|||
"crypto/md5" |
|||
"fmt" |
|||
"github.com/chrislusf/seaweedfs/weed/filer" |
|||
"github.com/chrislusf/seaweedfs/weed/glog" |
|||
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb" |
|||
weed_util "github.com/chrislusf/seaweedfs/weed/util" |
|||
"github.com/tecbot/gorocksdb" |
|||
"io" |
|||
) |
|||
|
|||
func init() { |
|||
filer.Stores = append(filer.Stores, &RocksDBStore{}) |
|||
} |
|||
|
|||
type RocksDBStore struct { |
|||
path string |
|||
db *gorocksdb.DB |
|||
} |
|||
|
|||
func (store *RocksDBStore) GetName() string { |
|||
return "rocksdb" |
|||
} |
|||
|
|||
func (store *RocksDBStore) Initialize(configuration weed_util.Configuration, prefix string) (err error) { |
|||
dir := configuration.GetString(prefix + "dir") |
|||
return store.initialize(dir) |
|||
} |
|||
|
|||
func (store *RocksDBStore) initialize(dir string) (err error) { |
|||
glog.Infof("filer store rocksdb dir: %s", dir) |
|||
if err := weed_util.TestFolderWritable(dir); err != nil { |
|||
return fmt.Errorf("Check Level Folder %s Writable: %s", dir, err) |
|||
} |
|||
|
|||
options := gorocksdb.NewDefaultOptions() |
|||
options.SetCreateIfMissing(true) |
|||
store.db, err = gorocksdb.OpenDb(options, dir) |
|||
|
|||
return |
|||
} |
|||
|
|||
func (store *RocksDBStore) BeginTransaction(ctx context.Context) (context.Context, error) { |
|||
return ctx, nil |
|||
} |
|||
func (store *RocksDBStore) CommitTransaction(ctx context.Context) error { |
|||
return nil |
|||
} |
|||
func (store *RocksDBStore) RollbackTransaction(ctx context.Context) error { |
|||
return nil |
|||
} |
|||
|
|||
func (store *RocksDBStore) InsertEntry(ctx context.Context, entry *filer.Entry) (err error) { |
|||
dir, name := entry.DirAndName() |
|||
key := genKey(dir, name) |
|||
|
|||
value, err := entry.EncodeAttributesAndChunks() |
|||
if err != nil { |
|||
return fmt.Errorf("encoding %s %+v: %v", entry.FullPath, entry.Attr, err) |
|||
} |
|||
|
|||
wo := gorocksdb.NewDefaultWriteOptions() |
|||
err = store.db.Put(wo, key, value) |
|||
|
|||
if err != nil { |
|||
return fmt.Errorf("persisting %s : %v", entry.FullPath, err) |
|||
} |
|||
|
|||
// println("saved", entry.FullPath, "chunks", len(entry.Chunks))
|
|||
|
|||
return nil |
|||
} |
|||
|
|||
func (store *RocksDBStore) UpdateEntry(ctx context.Context, entry *filer.Entry) (err error) { |
|||
|
|||
return store.InsertEntry(ctx, entry) |
|||
} |
|||
|
|||
func (store *RocksDBStore) FindEntry(ctx context.Context, fullpath weed_util.FullPath) (entry *filer.Entry, err error) { |
|||
dir, name := fullpath.DirAndName() |
|||
key := genKey(dir, name) |
|||
|
|||
ro := gorocksdb.NewDefaultReadOptions() |
|||
data, err := store.db.GetBytes(ro, key) |
|||
|
|||
if data == nil { |
|||
return nil, filer_pb.ErrNotFound |
|||
} |
|||
if err != nil { |
|||
return nil, fmt.Errorf("get %s : %v", entry.FullPath, err) |
|||
} |
|||
|
|||
entry = &filer.Entry{ |
|||
FullPath: fullpath, |
|||
} |
|||
err = entry.DecodeAttributesAndChunks(weed_util.MaybeDecompressData(data)) |
|||
if err != nil { |
|||
return entry, fmt.Errorf("decode %s : %v", entry.FullPath, err) |
|||
} |
|||
|
|||
// println("read", entry.FullPath, "chunks", len(entry.Chunks), "data", len(data), string(data))
|
|||
|
|||
return entry, nil |
|||
} |
|||
|
|||
func (store *RocksDBStore) DeleteEntry(ctx context.Context, fullpath weed_util.FullPath) (err error) { |
|||
dir, name := fullpath.DirAndName() |
|||
key := genKey(dir, name) |
|||
|
|||
wo := gorocksdb.NewDefaultWriteOptions() |
|||
err = store.db.Delete(wo, key) |
|||
if err != nil { |
|||
return fmt.Errorf("delete %s : %v", fullpath, err) |
|||
} |
|||
|
|||
return nil |
|||
} |
|||
|
|||
func (store *RocksDBStore) DeleteFolderChildren(ctx context.Context, fullpath weed_util.FullPath) (err error) { |
|||
directoryPrefix := genDirectoryKeyPrefix(fullpath, "") |
|||
|
|||
batch := new(gorocksdb.WriteBatch) |
|||
|
|||
ro := gorocksdb.NewDefaultReadOptions() |
|||
ro.SetFillCache(false) |
|||
iter := store.db.NewIterator(ro) |
|||
defer iter.Close() |
|||
err = enumerate(iter, directoryPrefix, nil, false, -1, func(key, value []byte) bool { |
|||
batch.Delete(key) |
|||
return true |
|||
}) |
|||
if err != nil { |
|||
return fmt.Errorf("delete list %s : %v", fullpath, err) |
|||
} |
|||
|
|||
wo := gorocksdb.NewDefaultWriteOptions() |
|||
err = store.db.Write(wo, batch) |
|||
|
|||
if err != nil { |
|||
return fmt.Errorf("delete %s : %v", fullpath, err) |
|||
} |
|||
|
|||
return nil |
|||
} |
|||
|
|||
func enumerate(iter *gorocksdb.Iterator, prefix, lastKey []byte, includeLastKey bool, limit int, fn func(key, value []byte) bool) error { |
|||
|
|||
if len(lastKey) == 0 { |
|||
iter.Seek(prefix) |
|||
} else { |
|||
iter.Seek(lastKey) |
|||
|
|||
if !includeLastKey { |
|||
k := iter.Key() |
|||
v := iter.Value() |
|||
key := k.Data() |
|||
defer k.Free() |
|||
defer v.Free() |
|||
|
|||
if !bytes.HasPrefix(key, prefix) { |
|||
return nil |
|||
} |
|||
|
|||
if bytes.Equal(key, lastKey) { |
|||
iter.Next() |
|||
} |
|||
|
|||
} |
|||
} |
|||
|
|||
i := 0 |
|||
for ; iter.Valid(); iter.Next() { |
|||
|
|||
if limit > 0 { |
|||
i++ |
|||
if i > limit { |
|||
break |
|||
} |
|||
} |
|||
|
|||
k := iter.Key() |
|||
v := iter.Value() |
|||
key := k.Data() |
|||
value := v.Data() |
|||
|
|||
if !bytes.HasPrefix(key, prefix) { |
|||
k.Free() |
|||
v.Free() |
|||
break |
|||
} |
|||
|
|||
ret := fn(key, value) |
|||
|
|||
k.Free() |
|||
v.Free() |
|||
|
|||
if !ret { |
|||
break |
|||
} |
|||
|
|||
} |
|||
|
|||
if err := iter.Err(); err != nil { |
|||
return fmt.Errorf("prefix scan iterator: %v", err) |
|||
} |
|||
return nil |
|||
} |
|||
|
|||
func (store *RocksDBStore) ListDirectoryEntries(ctx context.Context, fullpath weed_util.FullPath, startFileName string, inclusive bool, |
|||
limit int) (entries []*filer.Entry, err error) { |
|||
return store.ListDirectoryPrefixedEntries(ctx, fullpath, startFileName, inclusive, limit, "") |
|||
} |
|||
|
|||
func (store *RocksDBStore) ListDirectoryPrefixedEntries(ctx context.Context, fullpath weed_util.FullPath, startFileName string, inclusive bool, limit int, prefix string) (entries []*filer.Entry, err error) { |
|||
|
|||
directoryPrefix := genDirectoryKeyPrefix(fullpath, prefix) |
|||
lastFileStart := directoryPrefix |
|||
if startFileName != "" { |
|||
lastFileStart = genDirectoryKeyPrefix(fullpath, startFileName) |
|||
} |
|||
|
|||
ro := gorocksdb.NewDefaultReadOptions() |
|||
ro.SetFillCache(false) |
|||
iter := store.db.NewIterator(ro) |
|||
defer iter.Close() |
|||
err = enumerate(iter, directoryPrefix, lastFileStart, inclusive, limit, func(key, value []byte) bool { |
|||
fileName := getNameFromKey(key) |
|||
if fileName == "" { |
|||
return true |
|||
} |
|||
limit-- |
|||
if limit < 0 { |
|||
return false |
|||
} |
|||
entry := &filer.Entry{ |
|||
FullPath: weed_util.NewFullPath(string(fullpath), fileName), |
|||
} |
|||
|
|||
// println("list", entry.FullPath, "chunks", len(entry.Chunks))
|
|||
if decodeErr := entry.DecodeAttributesAndChunks(weed_util.MaybeDecompressData(value)); decodeErr != nil { |
|||
err = decodeErr |
|||
glog.V(0).Infof("list %s : %v", entry.FullPath, err) |
|||
return false |
|||
} |
|||
entries = append(entries, entry) |
|||
return true |
|||
}) |
|||
if err != nil { |
|||
return entries, fmt.Errorf("prefix list %s : %v", fullpath, err) |
|||
} |
|||
|
|||
return entries, err |
|||
} |
|||
|
|||
func genKey(dirPath, fileName string) (key []byte) { |
|||
key = hashToBytes(dirPath) |
|||
key = append(key, []byte(fileName)...) |
|||
return key |
|||
} |
|||
|
|||
func genDirectoryKeyPrefix(fullpath weed_util.FullPath, startFileName string) (keyPrefix []byte) { |
|||
keyPrefix = hashToBytes(string(fullpath)) |
|||
if len(startFileName) > 0 { |
|||
keyPrefix = append(keyPrefix, []byte(startFileName)...) |
|||
} |
|||
return keyPrefix |
|||
} |
|||
|
|||
func getNameFromKey(key []byte) string { |
|||
|
|||
return string(key[md5.Size:]) |
|||
|
|||
} |
|||
|
|||
// hash directory, and use last byte for partitioning
|
|||
func hashToBytes(dir string) []byte { |
|||
h := md5.New() |
|||
io.WriteString(h, dir) |
|||
|
|||
b := h.Sum(nil) |
|||
|
|||
return b |
|||
} |
|||
|
|||
func (store *RocksDBStore) Shutdown() { |
|||
store.db.Close() |
|||
} |
|||
@ -0,0 +1,51 @@ |
|||
// +build rocksdb
|
|||
|
|||
package rocksdb |
|||
|
|||
import ( |
|||
"context" |
|||
"fmt" |
|||
"github.com/tecbot/gorocksdb" |
|||
|
|||
"github.com/chrislusf/seaweedfs/weed/filer" |
|||
) |
|||
|
|||
func (store *RocksDBStore) KvPut(ctx context.Context, key []byte, value []byte) (err error) { |
|||
|
|||
wo := gorocksdb.NewDefaultWriteOptions() |
|||
err = store.db.Put(wo, key, value) |
|||
|
|||
if err != nil { |
|||
return fmt.Errorf("kv put: %v", err) |
|||
} |
|||
|
|||
return nil |
|||
} |
|||
|
|||
func (store *RocksDBStore) KvGet(ctx context.Context, key []byte) (value []byte, err error) { |
|||
|
|||
ro := gorocksdb.NewDefaultReadOptions() |
|||
value, err = store.db.GetBytes(ro, key) |
|||
|
|||
if value == nil { |
|||
return nil, filer.ErrKvNotFound |
|||
} |
|||
|
|||
if err != nil { |
|||
return nil, fmt.Errorf("kv get: %v", err) |
|||
} |
|||
|
|||
return |
|||
} |
|||
|
|||
func (store *RocksDBStore) KvDelete(ctx context.Context, key []byte) (err error) { |
|||
|
|||
wo := gorocksdb.NewDefaultWriteOptions() |
|||
err = store.db.Delete(wo, key) |
|||
|
|||
if err != nil { |
|||
return fmt.Errorf("kv delete: %v", err) |
|||
} |
|||
|
|||
return nil |
|||
} |
|||
@ -0,0 +1,117 @@ |
|||
// +build rocksdb
|
|||
|
|||
package rocksdb |
|||
|
|||
import ( |
|||
"context" |
|||
"fmt" |
|||
"io/ioutil" |
|||
"os" |
|||
"testing" |
|||
"time" |
|||
|
|||
"github.com/chrislusf/seaweedfs/weed/filer" |
|||
"github.com/chrislusf/seaweedfs/weed/util" |
|||
) |
|||
|
|||
func TestCreateAndFind(t *testing.T) { |
|||
testFiler := filer.NewFiler(nil, nil, "", 0, "", "", "", nil) |
|||
dir, _ := ioutil.TempDir("", "seaweedfs_filer_test") |
|||
defer os.RemoveAll(dir) |
|||
store := &RocksDBStore{} |
|||
store.initialize(dir) |
|||
testFiler.SetStore(store) |
|||
|
|||
fullpath := util.FullPath("/home/chris/this/is/one/file1.jpg") |
|||
|
|||
ctx := context.Background() |
|||
|
|||
entry1 := &filer.Entry{ |
|||
FullPath: fullpath, |
|||
Attr: filer.Attr{ |
|||
Mode: 0440, |
|||
Uid: 1234, |
|||
Gid: 5678, |
|||
}, |
|||
} |
|||
|
|||
if err := testFiler.CreateEntry(ctx, entry1, false, false, nil); err != nil { |
|||
t.Errorf("create entry %v: %v", entry1.FullPath, err) |
|||
return |
|||
} |
|||
|
|||
entry, err := testFiler.FindEntry(ctx, fullpath) |
|||
|
|||
if err != nil { |
|||
t.Errorf("find entry: %v", err) |
|||
return |
|||
} |
|||
|
|||
if entry.FullPath != entry1.FullPath { |
|||
t.Errorf("find wrong entry: %v", entry.FullPath) |
|||
return |
|||
} |
|||
|
|||
// checking one upper directory
|
|||
entries, _ := testFiler.ListDirectoryEntries(ctx, util.FullPath("/home/chris/this/is/one"), "", false, 100, "", "") |
|||
if len(entries) != 1 { |
|||
t.Errorf("list entries count: %v", len(entries)) |
|||
return |
|||
} |
|||
|
|||
// checking one upper directory
|
|||
entries, _ = testFiler.ListDirectoryEntries(ctx, util.FullPath("/"), "", false, 100, "", "") |
|||
if len(entries) != 1 { |
|||
t.Errorf("list entries count: %v", len(entries)) |
|||
return |
|||
} |
|||
|
|||
} |
|||
|
|||
func TestEmptyRoot(t *testing.T) { |
|||
testFiler := filer.NewFiler(nil, nil, "", 0, "", "", "", nil) |
|||
dir, _ := ioutil.TempDir("", "seaweedfs_filer_test2") |
|||
defer os.RemoveAll(dir) |
|||
store := &RocksDBStore{} |
|||
store.initialize(dir) |
|||
testFiler.SetStore(store) |
|||
|
|||
ctx := context.Background() |
|||
|
|||
// checking one upper directory
|
|||
entries, err := testFiler.ListDirectoryEntries(ctx, util.FullPath("/"), "", false, 100, "", "") |
|||
if err != nil { |
|||
t.Errorf("list entries: %v", err) |
|||
return |
|||
} |
|||
if len(entries) != 0 { |
|||
t.Errorf("list entries count: %v", len(entries)) |
|||
return |
|||
} |
|||
|
|||
} |
|||
|
|||
func BenchmarkInsertEntry(b *testing.B) { |
|||
testFiler := filer.NewFiler(nil, nil, "", 0, "", "", "", nil) |
|||
dir, _ := ioutil.TempDir("", "seaweedfs_filer_bench") |
|||
defer os.RemoveAll(dir) |
|||
store := &RocksDBStore{} |
|||
store.initialize(dir) |
|||
testFiler.SetStore(store) |
|||
|
|||
ctx := context.Background() |
|||
|
|||
b.ReportAllocs() |
|||
|
|||
for i := 0; i < b.N; i++ { |
|||
entry := &filer.Entry{ |
|||
FullPath: util.FullPath(fmt.Sprintf("/file%d.txt", i)), |
|||
Attr: filer.Attr{ |
|||
Crtime: time.Now(), |
|||
Mtime: time.Now(), |
|||
Mode: os.FileMode(0644), |
|||
}, |
|||
} |
|||
store.InsertEntry(ctx, entry) |
|||
} |
|||
} |
|||
@ -0,0 +1,7 @@ |
|||
// +build rocksdb
|
|||
|
|||
package weed_server |
|||
|
|||
import ( |
|||
_ "github.com/chrislusf/seaweedfs/weed/filer/rocksdb" |
|||
) |
|||
Write
Preview
Loading…
Cancel
Save
Reference in new issue