taigrr/bitcask
1
0
Fork 0
mirror of https://github.com/taigrr/bitcask synced 2025-01-18 04:03:17 -08:00
Code Issues Projects Releases Wiki Activity

Use []byte byte slices as keys directly avoiding serialing string(s) (#46)

Browse Source
This commit is contained in:
James Mills 2019-08-08 07:59:11 +10:00 committed by GitHub
parent 5d1dd6657a
commit 3c1808cad3
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
13 changed files with 145 additions and 97 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
bitcask.go
View File

@ -112,7 +112,7 @@ func (b *Bitcask) Sync() error {
// Get retrieves the value of the given key. If the key is not found or an/I/O
// error occurs a null byte slice is returned along with the error.
func (b *Bitcask) Get(key string) ([]byte, error) {
func (b *Bitcask) Get(key []byte) ([]byte, error) {
var df *internal.Datafile
item, ok := b.keydir.Get(key)
@ -140,13 +140,13 @@ func (b *Bitcask) Get(key string) ([]byte, error) {
}
// Has returns true if the key exists in the database, false otherwise.
func (b *Bitcask) Has(key string) bool {
func (b *Bitcask) Has(key []byte) bool {
_, ok := b.keydir.Get(key)
return ok
}
// Put stores the key and value in the database.
func (b *Bitcask) Put(key string, value []byte) error {
func (b *Bitcask) Put(key, value []byte) error {
if len(key) > b.config.maxKeySize {
return ErrKeyTooLarge
}
@ -160,21 +160,21 @@ func (b *Bitcask) Put(key string, value []byte) error {
}
item := b.keydir.Add(key, b.curr.FileID(), offset, n)
b.trie.Add(key, item)
b.trie.Add(string(key), item)
return nil
}
// Delete deletes the named key. If the key doesn't exist or an I/O error
// occurs the error is returned.
func (b *Bitcask) Delete(key string) error {
func (b *Bitcask) Delete(key []byte) error {
_, _, err := b.put(key, []byte{})
if err != nil {
return err
}
b.keydir.Delete(key)
b.trie.Remove(key)
b.trie.Remove(string(key))
return nil
}
@ -182,10 +182,10 @@ func (b *Bitcask) Delete(key string) error {
// Scan performs a prefix scan of keys matching the given prefix and calling
// the function `f` with the keys found. If the function returns an error
// no further keys are processed and the first error returned.
func (b *Bitcask) Scan(prefix string, f func(key string) error) error {
keys := b.trie.PrefixSearch(prefix)
func (b *Bitcask) Scan(prefix []byte, f func(key []byte) error) error {
keys := b.trie.PrefixSearch(string(prefix))
for _, key := range keys {
if err := f(key); err != nil {
if err := f([]byte(key)); err != nil {
return err
}
}
@ -198,14 +198,14 @@ func (b *Bitcask) Len() int {
}
// Keys returns all keys in the database as a channel of string(s)
func (b *Bitcask) Keys() chan string {
func (b *Bitcask) Keys() chan []byte {
return b.keydir.Keys()
}
// Fold iterates over all keys in the database calling the function `f` for
// each key. If the function returns an error, no further keys are processed
// and the error returned.
func (b *Bitcask) Fold(f func(key string) error) error {
func (b *Bitcask) Fold(f func(key []byte) error) error {
for key := range b.keydir.Keys() {
if err := f(key); err != nil {
return err
@ -214,7 +214,7 @@ func (b *Bitcask) Fold(f func(key string) error) error {
return nil
}
func (b *Bitcask) put(key string, value []byte) (int64, int64, error) {
func (b *Bitcask) put(key, value []byte) (int64, int64, error) {
b.mu.Lock()
defer b.mu.Unlock()
@ -301,7 +301,7 @@ func (b *Bitcask) reopen() error {
}
for key := range keydir.Keys() {
item, _ := keydir.Get(key)
trie.Add(key, item)
trie.Add(string(key), item)
}
} else {
for i, df := range datafiles {
@ -321,7 +321,7 @@ func (b *Bitcask) reopen() error {
}
item := keydir.Add(e.Key, ids[i], e.Offset, n)
trie.Add(e.Key, item)
trie.Add(string(e.Key), item)
}
}
}
@ -366,7 +366,7 @@ func (b *Bitcask) Merge() error {
// Rewrite all key/value pairs into merged database
// Doing this automatically strips deleted keys and
// old key/value pairs
err = b.Fold(func(key string) error {
err = b.Fold(func(key []byte) error {
value, err := b.Get(key)
if err != nil {
return err

113
bitcask_test.go
View File

@ -1,6 +1,7 @@
package bitcask
import (
"bytes"
"fmt"
"io/ioutil"
"os"
@ -13,6 +14,32 @@ import (
"github.com/stretchr/testify/assert"
)
type sortByteArrays [][]byte
func (b sortByteArrays) Len() int {
return len(b)
}
func (b sortByteArrays) Less(i, j int) bool {
switch bytes.Compare(b[i], b[j]) {
case -1:
return true
case 0, 1:
return false
}
return false
}
func (b sortByteArrays) Swap(i, j int) {
b[j], b[i] = b[i], b[j]
}
func SortByteArrays(src [][]byte) [][]byte {
sorted := sortByteArrays(src)
sort.Sort(sorted)
return sorted
}
func TestAll(t *testing.T) {
var (
db *Bitcask
@ -31,12 +58,12 @@ func TestAll(t *testing.T) {
})
t.Run("Put", func(t *testing.T) {
err = db.Put("foo", []byte("bar"))
err = db.Put([]byte([]byte("foo")), []byte("bar"))
assert.NoError(err)
})
t.Run("Get", func(t *testing.T) {
val, err := db.Get("foo")
val, err := db.Get([]byte("foo"))
assert.NoError(err)
assert.Equal([]byte("bar"), val)
})
@ -46,24 +73,24 @@ func TestAll(t *testing.T) {
})
t.Run("Has", func(t *testing.T) {
assert.True(db.Has("foo"))
assert.True(db.Has([]byte("foo")))
})
t.Run("Keys", func(t *testing.T) {
keys := make([]string, 0)
keys := make([][]byte, 0)
for key := range db.Keys() {
keys = append(keys, key)
}
assert.Equal([]string{"foo"}, keys)
assert.Equal([][]byte{[]byte("foo")}, keys)
})
t.Run("Fold", func(t *testing.T) {
var (
keys []string
keys [][]byte
values [][]byte
)
err := db.Fold(func(key string) error {
err := db.Fold(func(key []byte) error {
value, err := db.Get(key)
if err != nil {
return err
@ -73,14 +100,14 @@ func TestAll(t *testing.T) {
return nil
})
assert.NoError(err)
assert.Equal([]string{"foo"}, keys)
assert.Equal([][]byte{[]byte("foo")}, keys)
assert.Equal([][]byte{[]byte("bar")}, values)
})
t.Run("Delete", func(t *testing.T) {
err := db.Delete("foo")
err := db.Delete([]byte("foo"))
assert.NoError(err)
_, err = db.Get("foo")
_, err = db.Get([]byte("foo"))
assert.Error(err)
assert.Equal(ErrKeyNotFound, err)
})
@ -114,20 +141,20 @@ func TestDeletedKeys(t *testing.T) {
})
t.Run("Put", func(t *testing.T) {
err = db.Put("foo", []byte("bar"))
err = db.Put([]byte("foo"), []byte("bar"))
assert.NoError(err)
})
t.Run("Get", func(t *testing.T) {
val, err := db.Get("foo")
val, err := db.Get([]byte("foo"))
assert.NoError(err)
assert.Equal([]byte("bar"), val)
})
t.Run("Delete", func(t *testing.T) {
err := db.Delete("foo")
err := db.Delete([]byte("foo"))
assert.NoError(err)
_, err = db.Get("foo")
_, err = db.Get([]byte("foo"))
assert.Error(err)
assert.Equal(ErrKeyNotFound, err)
})
@ -155,7 +182,7 @@ func TestDeletedKeys(t *testing.T) {
})
t.Run("Get", func(t *testing.T) {
_, err = db.Get("foo")
_, err = db.Get([]byte("foo"))
assert.Error(err)
assert.Equal(ErrKeyNotFound, err)
})
@ -181,7 +208,7 @@ func TestMaxKeySize(t *testing.T) {
})
t.Run("Put", func(t *testing.T) {
key := strings.Repeat(" ", 17)
key := []byte(strings.Repeat(" ", 17))
value := []byte("foobar")
err = db.Put(key, value)
assert.Error(err)
@ -203,7 +230,7 @@ func TestMaxValueSize(t *testing.T) {
})
t.Run("Put", func(t *testing.T) {
key := "foo"
key := []byte("foo")
value := []byte(strings.Repeat(" ", 17))
err = db.Put(key, value)
assert.Error(err)
@ -229,12 +256,12 @@ func TestStats(t *testing.T) {
})
t.Run("Put", func(t *testing.T) {
err := db.Put("foo", []byte("bar"))
err := db.Put([]byte("foo"), []byte("bar"))
assert.NoError(err)
})
t.Run("Get", func(t *testing.T) {
val, err := db.Get("foo")
val, err := db.Get([]byte("foo"))
assert.NoError(err)
assert.Equal([]byte("bar"), val)
})
@ -276,7 +303,7 @@ func TestMerge(t *testing.T) {
})
t.Run("Put", func(t *testing.T) {
err := db.Put("foo", []byte("bar"))
err := db.Put([]byte("foo"), []byte("bar"))
assert.NoError(err)
})
@ -287,7 +314,7 @@ func TestMerge(t *testing.T) {
t.Run("Put", func(t *testing.T) {
for i := 0; i < 10; i++ {
err := db.Put("foo", []byte("bar"))
err := db.Put([]byte("foo"), []byte("bar"))
assert.NoError(err)
}
})
@ -340,7 +367,7 @@ func TestConcurrent(t *testing.T) {
})
t.Run("Put", func(t *testing.T) {
err = db.Put("foo", []byte("bar"))
err = db.Put([]byte("foo"), []byte("bar"))
assert.NoError(err)
})
})
@ -353,7 +380,7 @@ func TestConcurrent(t *testing.T) {
}()
for i := 0; i <= 100; i++ {
if i%x == 0 {
key := fmt.Sprintf("k%d", i)
key := []byte(fmt.Sprintf("k%d", i))
value := []byte(fmt.Sprintf("v%d", i))
err := db.Put(key, value)
assert.NoError(err)
@ -377,7 +404,7 @@ func TestConcurrent(t *testing.T) {
wg.Done()
}()
for i := 0; i <= N; i++ {
value, err := db.Get("foo")
value, err := db.Get([]byte("foo"))
assert.NoError(err)
assert.Equal([]byte("bar"), value)
}
@ -420,12 +447,12 @@ func TestScan(t *testing.T) {
"2": []byte("2"),
"3": []byte("3"),
"food": []byte("pizza"),
"foo": []byte("foo"),
"foo": []byte([]byte("foo")),
"fooz": []byte("fooz ball"),
"hello": []byte("world"),
}
for k, v := range items {
err = db.Put(k, v)
err = db.Put([]byte(k), v)
assert.NoError(err)
}
})
@ -433,21 +460,21 @@ func TestScan(t *testing.T) {
t.Run("Scan", func(t *testing.T) {
var (
vals []string
expected = []string{
"foo",
"fooz ball",
"pizza",
vals [][]byte
expected = [][]byte{
[]byte("foo"),
[]byte("fooz ball"),
[]byte("pizza"),
}
)
err = db.Scan("fo", func(key string) error {
err = db.Scan([]byte("fo"), func(key []byte) error {
val, err := db.Get(key)
assert.NoError(err)
vals = append(vals, string(val))
vals = append(vals, val)
return nil
})
sort.Strings(vals)
vals = SortByteArrays(vals)
assert.Equal(expected, vals)
})
}
@ -510,7 +537,7 @@ func BenchmarkGet(b *testing.B) {
b.Run(tt.name, func(b *testing.B) {
b.SetBytes(int64(tt.size))
key := "foo"
key := []byte("foo")
value := []byte(strings.Repeat(" ", tt.size))
options := []Option{
@ -536,7 +563,7 @@ func BenchmarkGet(b *testing.B) {
if err != nil {
b.Fatal(err)
}
if string(val) != string(value) {
if !bytes.Equal(val, value) {
b.Errorf("unexpected value")
}
}
@ -580,7 +607,7 @@ func BenchmarkPut(b *testing.B) {
b.Run(tt.name, func(b *testing.B) {
b.SetBytes(int64(tt.size))
key := "foo"
key := []byte("foo")
value := []byte(strings.Repeat(" ", tt.size))
b.ResetTimer()
for i := 0; i < b.N; i++ {
@ -616,30 +643,30 @@ func BenchmarkScan(b *testing.B) {
"2": []byte("2"),
"3": []byte("3"),
"food": []byte("pizza"),
"foo": []byte("foo"),
"foo": []byte([]byte("foo")),
"fooz": []byte("fooz ball"),
"hello": []byte("world"),
}
for k, v := range items {
err := db.Put(k, v)
err := db.Put([]byte(k), v)
if err != nil {
b.Fatal(err)
}
}
var expected = []string{"foo", "food", "fooz"}
var expected = [][]byte{[]byte("foo"), []byte("food"), []byte("fooz")}
b.ResetTimer()
for i := 0; i < b.N; i++ {
var keys []string
err = db.Scan("fo", func(key string) error {
var keys [][]byte
err = db.Scan([]byte("fo"), func(key []byte) error {
keys = append(keys, key)
return nil
})
if err != nil {
b.Fatal(err)
}
sort.Strings(keys)
keys = SortByteArrays(keys)
if !reflect.DeepEqual(expected, keys) {
b.Fatal(fmt.Errorf("expected keys=#%v got=%#v", expected, keys))
}

2
cmd/bitcask/del.go
View File

@ -37,7 +37,7 @@ func del(path, key string) int {
}
defer db.Close()
err = db.Delete(key)
err = db.Delete([]byte(key))
if err != nil {
log.WithError(err).Error("error deleting key")
return 1

2
cmd/bitcask/get.go
View File

@ -38,7 +38,7 @@ func get(path, key string) int {
}
defer db.Close()
value, err := db.Get(key)
value, err := db.Get([]byte(key))
if err != nil {
log.WithError(err).Error("error reading key")
return 1

4
cmd/bitcask/keys.go
View File

@ -36,8 +36,8 @@ func keys(path string) int {
}
defer db.Close()
err = db.Fold(func(key string) error {
fmt.Printf("%s\n", key)
err = db.Fold(func(key []byte) error {
fmt.Printf("%s\n", string(key))
return nil
})
if err != nil {

2
cmd/bitcask/put.go
View File

@ -55,7 +55,7 @@ func put(path, key string, value io.Reader) int {
return 1
}
err = db.Put(key, data)
err = db.Put([]byte(key), data)
if err != nil {
log.WithError(err).Error("error writing key")
return 1

2
cmd/bitcask/scan.go
View File

@ -40,7 +40,7 @@ func scan(path, prefix string) int {
}
defer db.Close()
err = db.Scan(prefix, func(key string) error {
err = db.Scan([]byte(prefix), func(key []byte) error {
value, err := db.Get(key)
if err != nil {
log.WithError(err).Error("error reading key")

8
cmd/bitcaskd/main.go
View File

@ -76,7 +76,7 @@ func main() {
conn.WriteError("ERR wrong number of arguments for '" + string(cmd.Args[0]) + "' command")
return
}
key := string(cmd.Args[1])
key := cmd.Args[1]
value := cmd.Args[2]
err = db.Put(key, value)
if err != nil {
@ -89,7 +89,7 @@ func main() {
conn.WriteError("ERR wrong number of arguments for '" + string(cmd.Args[0]) + "' command")
return
}
key := string(cmd.Args[1])
key := cmd.Args[1]
value, err := db.Get(key)
if err != nil {
conn.WriteNull()
@ -106,7 +106,7 @@ func main() {
conn.WriteError("ERR wrong number of arguments for '" + string(cmd.Args[0]) + "' command")
return
}
key := string(cmd.Args[1])
key := cmd.Args[1]
if db.Has(key) {
conn.WriteInt(1)
} else {
@ -117,7 +117,7 @@ func main() {
conn.WriteError("ERR wrong number of arguments for '" + string(cmd.Args[0]) + "' command")
return
}
key := string(cmd.Args[1])
key := cmd.Args[1]
err := db.Delete(key)
if err != nil {
conn.WriteInt(0)

2
internal/entry.go
View File

@ -6,7 +6,7 @@ import (
pb "github.com/prologic/bitcask/internal/proto"
)
func NewEntry(key string, value []byte) pb.Entry {
func NewEntry(key, value []byte) pb.Entry {
checksum := crc32.ChecksumIEEE(value)
return pb.Entry{

48
internal/keydir.go
View File

@ -17,51 +17,58 @@ type Item struct {
type Keydir struct {
sync.RWMutex
kv map[string]Item
keys map[uint64][]byte
items map[uint64]Item
}
func NewKeydir() *Keydir {
return &Keydir{
kv: make(map[string]Item),
keys: make(map[uint64][]byte),
items: make(map[uint64]Item),
}
}
func (k *Keydir) Add(key string, fileid int, offset, size int64) Item {
func (k *Keydir) Add(key []byte, fileid int, offset, size int64) Item {
item := Item{
FileID: fileid,
Offset: offset,
Size: size,
}
hash := Hash(key)
k.Lock()
k.kv[key] = item
k.keys[hash] = key
k.items[hash] = item
k.Unlock()
return item
}
func (k *Keydir) Get(key string) (Item, bool) {
func (k *Keydir) Get(key []byte) (Item, bool) {
k.RLock()
item, ok := k.kv[key]
item, ok := k.items[Hash(key)]
k.RUnlock()
return item, ok
}
func (k *Keydir) Delete(key string) {
func (k *Keydir) Delete(key []byte) {
hash := Hash(key)
k.Lock()
delete(k.kv, key)
delete(k.keys, hash)
delete(k.items, hash)
k.Unlock()
}
func (k *Keydir) Len() int {
return len(k.kv)
return len(k.keys)
}
func (k *Keydir) Keys() chan string {
ch := make(chan string)
func (k *Keydir) Keys() chan []byte {
ch := make(chan []byte)
go func() {
k.RLock()
for key := range k.kv {
for _, key := range k.keys {
ch <- key
}
close(ch)
@ -73,8 +80,10 @@ func (k *Keydir) Keys() chan string {
func (k *Keydir) Bytes() ([]byte, error) {
var buf bytes.Buffer
enc := gob.NewEncoder(&buf)
err := enc.Encode(k.kv)
if err != nil {
if err := enc.Encode(k.keys); err != nil {
return nil, err
}
if err := enc.Encode(k.items); err != nil {
return nil, err
}
return buf.Bytes(), nil
@ -88,7 +97,10 @@ func (k *Keydir) Load(fn string) error {
defer f.Close()
dec := gob.NewDecoder(f)
if err := dec.Decode(&k.kv); err != nil {
if err := dec.Decode(&k.keys); err != nil {
return err
}
if err := dec.Decode(&k.items); err != nil {
return err
}
@ -107,8 +119,10 @@ func (k *Keydir) Save(fn string) error {
func NewKeydirFromBytes(r io.Reader) (*Keydir, error) {
k := NewKeydir()
dec := gob.NewDecoder(r)
err := dec.Decode(&k.kv)
if err != nil {
if err := dec.Decode(&k.keys); err != nil {
return nil, err
}
if err := dec.Decode(&k.items); err != nil {
return nil, err
}
return k, nil

20
internal/proto/entry.pb.go
View File

@ -20,7 +20,7 @@ const _ = proto.ProtoPackageIsVersion2 // please upgrade the proto package
type Entry struct {
Checksum uint32 `protobuf:"varint,1,opt,name=Checksum,proto3" json:"Checksum,omitempty"`
Key string `protobuf:"bytes,2,opt,name=Key,proto3" json:"Key,omitempty"`
Key []byte `protobuf:"bytes,2,opt,name=Key,proto3" json:"Key,omitempty"`
Offset int64 `protobuf:"varint,3,opt,name=Offset,proto3" json:"Offset,omitempty"`
Value []byte `protobuf:"bytes,4,opt,name=Value,proto3" json:"Value,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
@ -32,7 +32,7 @@ func (m *Entry) Reset() { *m = Entry{} }
func (m *Entry) String() string { return proto.CompactTextString(m) }
func (*Entry) ProtoMessage() {}
func (*Entry) Descriptor() ([]byte, []int) {
return fileDescriptor_entry_db5b99f271e6b4b6, []int{0}
return fileDescriptor_entry_085f82c8520d7cd0, []int{0}
}
func (m *Entry) XXX_Unmarshal(b []byte) error {
return xxx_messageInfo_Entry.Unmarshal(m, b)
@ -59,11 +59,11 @@ func (m *Entry) GetChecksum() uint32 {
return 0
}
func (m *Entry) GetKey() string {
func (m *Entry) GetKey() []byte {
if m != nil {
return m.Key
}
return ""
return nil
}
func (m *Entry) GetOffset() int64 {
@ -84,16 +84,16 @@ func init() {
proto.RegisterType((*Entry)(nil), "proto.Entry")
}
func init() { proto.RegisterFile("entry.proto", fileDescriptor_entry_db5b99f271e6b4b6) }
func init() { proto.RegisterFile("entry.proto", fileDescriptor_entry_085f82c8520d7cd0) }
var fileDescriptor_entry_db5b99f271e6b4b6 = []byte{
// 126 bytes of a gzipped FileDescriptorProto
var fileDescriptor_entry_085f82c8520d7cd0 = []byte{
// 123 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xe2, 0xe2, 0x4e, 0xcd, 0x2b, 0x29,
0xaa, 0xd4, 0x2b, 0x28, 0xca, 0x2f, 0xc9, 0x17, 0x62, 0x05, 0x53, 0x4a, 0xc9, 0x5c, 0xac, 0xae,
0x20, 0x51, 0x21, 0x29, 0x2e, 0x0e, 0xe7, 0x8c, 0xd4, 0xe4, 0xec, 0xe2, 0xd2, 0x5c, 0x09, 0x46,
0x05, 0x46, 0x0d, 0xde, 0x20, 0x38, 0x5f, 0x48, 0x80, 0x8b, 0xd9, 0x3b, 0xb5, 0x52, 0x82, 0x49,
0x81, 0x51, 0x83, 0x33, 0x08, 0xc4, 0x14, 0x12, 0xe3, 0x62, 0xf3, 0x4f, 0x4b, 0x2b, 0x4e, 0x2d,
0x81, 0x51, 0x83, 0x27, 0x08, 0xc4, 0x14, 0x12, 0xe3, 0x62, 0xf3, 0x4f, 0x4b, 0x2b, 0x4e, 0x2d,
0x91, 0x60, 0x56, 0x60, 0xd4, 0x60, 0x0e, 0x82, 0xf2, 0x84, 0x44, 0xb8, 0x58, 0xc3, 0x12, 0x73,
0x4a, 0x53, 0x25, 0x58, 0x14, 0x18, 0x35, 0x78, 0x82, 0x20, 0x9c, 0x24, 0x36, 0xb0, 0x5d, 0xc6,
0x80, 0x00, 0x00, 0x00, 0xff, 0xff, 0x76, 0xd2, 0x3e, 0x83, 0x81, 0x00, 0x00, 0x00,
0x4a, 0x53, 0x25, 0x58, 0xc0, 0x6a, 0x21, 0x9c, 0x24, 0x36, 0xb0, 0x5d, 0xc6, 0x80, 0x00, 0x00,
0x00, 0xff, 0xff, 0x47, 0x6a, 0x41, 0xd4, 0x81, 0x00, 0x00, 0x00,
}

2
internal/proto/entry.proto
View File

@ -4,7 +4,7 @@ package proto;
message Entry {
uint32 Checksum = 1;
string Key = 2;
bytes Key = 2;
int64 Offset = 3;
bytes Value = 4;
}

7
internal/utils.go
View File

@ -2,6 +2,7 @@ package internal
import (
"fmt"
"hash/fnv"
"os"
"path/filepath"
"sort"
@ -9,6 +10,12 @@ import (
"strings"
)
func Hash(key []byte) uint64 {
h := fnv.New64a()
h.Write(key)
return h.Sum64()
}
func Exists(path string) bool {
_, err := os.Stat(path)
return err == nil