Compare commits

..

43 Commits

Author SHA1 Message Date
Ignacio Hagopian
0338755f8c fix readme typos (#81) 2019-09-02 10:22:08 +10:00
Ignacio Hagopian
877bf982b1 fix go vet (#80) 2019-09-02 10:20:56 +10:00
James Mills
abbbeb8e1d Replace keydir with ART trie (#75)
* Replace keydir with ART trie

* Address some review feedback

* Address review feedback (consts)
2019-09-02 08:38:56 +10:00
James Mills
36bc134b22 Fix a bug wit the decoder passing the wrong value for the value's offset into the buffer (#77) 2019-08-31 08:35:17 +10:00
James Mills
ea96b8afc0 Update README.md 2019-08-30 13:28:17 +10:00
James Mills
b3d6f734b6 Use an Adaptive Radix Tree (#71) 2019-08-30 08:13:24 +10:00
Ignacio Hagopian
55459a5c93 readme: fix stargazers link (#73) 2019-08-30 08:05:08 +10:00
James Mills
a20ee3e3d4 Update README.md 2019-08-26 08:45:15 +10:00
James Mills
cd27b84069 Create FUNDING.yml 2019-08-25 12:01:15 +10:00
James Mills
b28353de02 Update README.md 2019-08-18 22:34:14 +10:00
Awn
e8bee948bc Make optimised scan functionality optional (#68) 2019-08-16 10:51:59 +10:00
Xin Zhang
156d29e344 Fix typo (#65) 2019-08-14 20:53:08 +10:00
James Mills
c5a565cd82 Adds WithSync(...) option to turn on sync after write durability (#63)
* Added WithSync(...) option to turn  on sync after write durability

* Add Sync/NoSync benchmark variants for Put()
2019-08-12 06:47:46 +10:00
Ignacio Hagopian
8f56cffd86 codec: collapse write and save extra alloc (#64) 2019-08-12 06:47:26 +10:00
James Mills
7204a33512 Fix and cleanup some unnecessary internal sub-packages and duplication 2019-08-08 22:28:25 +10:00
James Mills
c7d101d34f Fixed missing model import for codec with undefined Entry 2019-08-08 20:14:27 +10:00
Awn
af43cfa8f1 Remove merge function (#60)
* tidy: clean up some leftovers

Fixes #56
Fixes #57
Fixes #58

* api: remove standalone merge function

Fixes #55
2019-08-08 19:51:45 +10:00
James Mills
110c5024ee Update README.md 2019-08-08 12:17:42 +10:00
Ignacio Hagopian
1f10b4026d inline hash function to save extra alloc (#53) 2019-08-08 09:52:31 +10:00
Ignacio Hagopian
fd179b4a86 custom high-performance encoder implementation (#52) 2019-08-08 09:21:46 +10:00
James Mills
755b1879b5 Use []byte byte slices as keys directly avoiding serialing string(s) (#46) (#51) 2019-08-08 08:14:48 +10:00
James Mills
d0c913ccee Revert "Use []byte byte slices as keys directly avoiding serialing string(s) (#46)" (#50)
This reverts commit 3c1808cad3.
2019-08-08 08:06:38 +10:00
James Mills
6b372d8334 Added export/import sub-commands to backup/resotre a database (#48) 2019-08-08 08:00:29 +10:00
James Mills
3c1808cad3 Use []byte byte slices as keys directly avoiding serialing string(s) (#46) 2019-08-08 07:59:11 +10:00
James Mills
5d1dd6657a Fixed handling of missing config.json from cli behavior 2019-08-07 21:47:51 +10:00
James Mills
1ba9ca46e3 Rename set command to put and cleanup the command's docs 2019-08-07 21:44:33 +10:00
James Mills
2a419c46d2 Update README.md 2019-08-07 13:24:38 +10:00
James Mills
e543fc38fb Added AUTHORS file to record contributors beyond the scope of Github metadata (#41) 2019-08-07 13:21:09 +10:00
James Mills
82e26449fa Added the same functional options to the bitcask CLI and persist options to the db store (#40) 2019-08-07 10:23:10 +10:00
James Mills
bce2721be4 Update README.md 2019-08-06 08:15:03 +10:00
Ignacio Hagopian
f2b5515e03 update trie dependency to take advantage of improvements (#45) 2019-08-06 08:05:41 +10:00
James Mills
8b684b635d Update CONTRIBUTING.md 2019-08-05 19:48:32 +10:00
Ignacio Hagopian
a407905ae2 Improve Get/Put performance with optional mempooling (#36)
* avoid unnecessary use of encoder/decoder to decrease memory allocations

* add an optional configurable mempool to avoid extra allocs

* add doc.go with examples
2019-08-05 07:23:07 +10:00
James Mills
6ceeccfd64 Update README.md 2019-08-03 19:49:15 +10:00
James Mills
35dc7e70d2 Update README.md 2019-08-03 19:47:23 +10:00
James Mills
6cc1154611 Update README.md 2019-08-03 19:46:18 +10:00
Ignacio Hagopian
8aa66c66da keydir: avoid defers (#34) 2019-08-01 19:18:05 +10:00
Ignacio Hagopian
e3242c8426 README: typos (#35) 2019-08-01 13:48:36 +10:00
James Mills
912371645d Fixed an off-by-one bug with managing datafiles (#31) 2019-07-29 23:49:37 +10:00
James Mills
bc782a3083 Update README.md 2019-07-27 07:57:38 +10:00
James Mills
a2161179ef Update README.md 2019-07-27 07:56:34 +10:00
James Mills
51bac21c0a Improves Merge() operation by also pruning old key/value pairs (#29)
* Added new API Stats() and Prune()

* Improved Merge() logic to also prune old key/values and actually reclaim disk space

* Added backward compat for the old Merge() function

* Refactor indexing of keys to items (hints)

* Remove redundant TestOpenMerge

* Add unit test for Stats()

* Improve TestMerge()
2019-07-27 07:52:25 +10:00
Awn
b7ac95d66a remove merge folder after merge completes (#26) 2019-07-25 08:31:44 +10:00
33 changed files with 1278 additions and 776 deletions

2
.github/FUNDING.yml vendored Normal file
View File

@@ -0,0 +1,2 @@
github: prologic
patreon: prologic

13
AUTHORS Normal file
View File

@@ -0,0 +1,13 @@
# Entries should be added alphabetically in the form:
# Name or Organization <email address>
# The email address is not required for organizations.
Awn Umar <awn@spacetime.dev>
Christian Muehlhaeuser <muesli@gmail.com>
Ignacio Hagopian <jsign.uy@gmail.com>
James Mills <prologic@shortcircuit.net.au>
Jesse Donat <donatj@gmail.com>
Kebert Xela kebertxela
panyun panyun
Whemoon Jang <palindrom615@gmail.com>
Yury Fedorov orlangure

View File

@@ -1,8 +1,12 @@
# Contributing
No preference. If you know hot to use Github and contributed to open source projects before then:
No preference. If you know how to use Github and have contributed to open source projects before then:
* File an issue
* Submit a pull request
* File an issue + Submit a pull request
* Use this project somewhere :)
Be sure to add yourself to the [AUTHORS](/AUTHORS) file when you submit your PR(s). Every contribution counts no how big or small!
Thanks for using Bitcask!

121
README.md
View File

@@ -4,14 +4,10 @@
[![CodeCov](https://codecov.io/gh/prologic/bitcask/branch/master/graph/badge.svg)](https://codecov.io/gh/prologic/bitcask)
[![Go Report Card](https://goreportcard.com/badge/prologic/bitcask)](https://goreportcard.com/report/prologic/bitcask)
[![GoDoc](https://godoc.org/github.com/prologic/bitcask?status.svg)](https://godoc.org/github.com/prologic/bitcask)
[![Sourcegraph](https://sourcegraph.com/github.com/prologic/bitcask/-/badge.svg)](https://sourcegraph.com/github.com/prologic/bitcask?badge)
[![Github all releases](https://img.shields.io/github/downloads/prologic/bitcask/total.svg)](https://github.com/prologic/bitcask/releases)
[![GitHub license](https://img.shields.io/github/license/prologic/bitcask.svg)](https://github.com/prologic/bitcask)
[![Sourcegraph](https://sourcegraph.com/github.com/prologic/bitcask/-/badge.svg)](https://sourcegraph.com/github.com/prologic/bitcask?badge)
[![](https://images.microbadger.com/badges/version/prologic/bitcask.svg)](https://microbadger.com/images/prologic/bitcask)
[![](https://images.microbadger.com/badges/image/prologic/bitcask.svg)](https://microbadger.com/images/prologic/bitcask)
A high performance Key/Value store written in [Go](https://golang.org) with a predictable read/write performance and high throughput. Uses a [Bitcask](https://en.wikipedia.org/wiki/Bitcask) on-disk layout (LSM+WAL) similar to [Riak](https://riak.com/). 🗃️
A high performance Key/Value store written in [Go](https://golang.org) with a predictable read/write performance and high throughput. Uses a [Bitcask](https://en.wikipedia.org/wiki/Bitcask) on-disk layout (LSM+WAL) similar to [Riak](https://riak.com/)
For a more feature-complete Redis-compatible server, distributed key/value store have a look at [Bitraft](https://github.com/prologic/bitraft) which uses this library as its backend. Use [Bitcask](https://github.com/prologic/bitcask) as a starting point or if you want to embed in your application, use [Bitraft](https://github.com/prologic/bitraft) if you need a complete server/client solution with high availability with a Redis-compatible API.
@@ -21,36 +17,19 @@ For a more feature-complete Redis-compatible server, distributed key/value store
* Builtin CLI (`bitcask`)
* Builtin Redis-compatible server (`bitcaskd`)
* Predictable read/write performance
* Low latecny
* Low latency
* High throughput (See: [Performance](README.md#Performance) )
## Development
1. Get the source
```#!bash
```#!sh
$ git clone https://github.com/prologic/bitcask.git
```
2. Install required tools
This library uses [Protobuf](https://github.com/protocolbuffers/protobuf) to serialize data on disk. Please follow the
instructions for installing `protobuf` on your system. You will also need the
following Go libraries/tools to generate Go code from Protobuf defs:
3. Build the project
```#!bash
$ make
```
This will invoke `go generate` and `go build`.
- [protoc-gen-go](https://github.com/golang/protobuf)
## Install
```#!bash
```#!sh
$ go get github.com/prologic/bitcask
```
@@ -58,7 +37,7 @@ $ go get github.com/prologic/bitcask
Install the package into your project:
```#!bash
```#!sh
$ go get github.com/prologic/bitcask
```
@@ -80,7 +59,7 @@ documentation and other examples.
## Usage (tool)
```#!bash
```#!sh
$ bitcask -p /tmp/db set Hello World
$ bitcask -p /tmp/db get Hello
World
@@ -90,14 +69,14 @@ World
There is also a builtin very simple Redis-compatible server called `bitcaskd`:
```#!bash
```#!sh
$ ./bitcaskd ./tmp
INFO[0000] starting bitcaskd v0.0.7@146f777 bind=":6379" path=./tmp
```
Example session:
```
```#!sh
$ telnet localhost 6379
Trying ::1...
Connected to localhost.
@@ -122,7 +101,7 @@ Connection closed by foreign host.
You can also use the [Bitcask Docker Image](https://cloud.docker.com/u/prologic/repository/docker/prologic/bitcask):
```#!bash
```#!sh
$ docker pull prologic/bitcask
$ docker run -d -p 6379:6379 prologic/bitcask
```
@@ -131,39 +110,71 @@ $ docker run -d -p 6379:6379 prologic/bitcask
Benchmarks run on a 11" Macbook with a 1.4Ghz Intel Core i7:
```
```#!sh
$ make bench
...
BenchmarkGet/128B-4 500000 2537 ns/op 672 B/op 7 allocs/op
BenchmarkGet/256B-4 500000 2629 ns/op 1056 B/op 7 allocs/op
BenchmarkGet/512B-4 500000 2773 ns/op 1888 B/op 7 allocs/op
BenchmarkGet/1K-4 500000 3202 ns/op 3552 B/op 7 allocs/op
BenchmarkGet/2K-4 300000 3904 ns/op 6880 B/op 7 allocs/op
BenchmarkGet/4K-4 300000 5678 ns/op 14048 B/op 7 allocs/op
BenchmarkGet/8K-4 200000 8948 ns/op 27360 B/op 7 allocs/op
BenchmarkGet/16K-4 100000 14635 ns/op 53472 B/op 7 allocs/op
BenchmarkGet/32K-4 50000 28292 ns/op 114912 B/op 7 allocs/op
goos: darwin
goarch: amd64
pkg: github.com/prologic/bitcask
BenchmarkPut/128B-4 200000 8173 ns/op 409 B/op 6 allocs/op
BenchmarkPut/256B-4 200000 8404 ns/op 538 B/op 6 allocs/op
BenchmarkPut/512B-4 200000 9741 ns/op 829 B/op 6 allocs/op
BenchmarkPut/1K-4 100000 13118 ns/op 1411 B/op 6 allocs/op
BenchmarkPut/2K-4 100000 17982 ns/op 2573 B/op 6 allocs/op
BenchmarkPut/4K-4 50000 35477 ns/op 5154 B/op 6 allocs/op
BenchmarkPut/8K-4 30000 54021 ns/op 9804 B/op 6 allocs/op
BenchmarkPut/16K-4 20000 96551 ns/op 18849 B/op 6 allocs/op
BenchmarkPut/32K-4 10000 129957 ns/op 41561 B/op 7 allocs/op
BenchmarkGet/128B-4 300000 3913 ns/op 32.71 MB/s 387 B/op 4 allocs/op
BenchmarkGet/128BWithPool-4 300000 4143 ns/op 30.89 MB/s 227 B/op 3 allocs/op
BenchmarkGet/256B-4 300000 3919 ns/op 65.31 MB/s 643 B/op 4 allocs/op
BenchmarkGet/256BWithPool-4 300000 4270 ns/op 59.95 MB/s 355 B/op 3 allocs/op
BenchmarkGet/512B-4 300000 4248 ns/op 120.52 MB/s 1187 B/op 4 allocs/op
BenchmarkGet/512BWithPool-4 300000 4676 ns/op 109.48 MB/s 611 B/op 3 allocs/op
BenchmarkGet/1K-4 200000 5248 ns/op 195.10 MB/s 2275 B/op 4 allocs/op
BenchmarkGet/1KWithPool-4 200000 5270 ns/op 194.28 MB/s 1123 B/op 3 allocs/op
BenchmarkGet/2K-4 200000 6229 ns/op 328.74 MB/s 4451 B/op 4 allocs/op
BenchmarkGet/2KWithPool-4 200000 6282 ns/op 325.99 MB/s 2147 B/op 3 allocs/op
BenchmarkGet/4K-4 200000 9027 ns/op 453.74 MB/s 9059 B/op 4 allocs/op
BenchmarkGet/4KWithPool-4 200000 8906 ns/op 459.87 MB/s 4195 B/op 3 allocs/op
BenchmarkGet/8K-4 100000 12024 ns/op 681.28 MB/s 17763 B/op 4 allocs/op
BenchmarkGet/8KWithPool-4 200000 11103 ns/op 737.79 MB/s 8291 B/op 3 allocs/op
BenchmarkGet/16K-4 100000 16844 ns/op 972.65 MB/s 34915 B/op 4 allocs/op
BenchmarkGet/16KWithPool-4 100000 14575 ns/op 1124.10 MB/s 16483 B/op 3 allocs/op
BenchmarkGet/32K-4 50000 27770 ns/op 1179.97 MB/s 73827 B/op 4 allocs/op
BenchmarkGet/32KWithPool-4 100000 24495 ns/op 1337.74 MB/s 32867 B/op 3 allocs/op
BenchmarkScan-4 1000000 2011 ns/op 493 B/op 25 allocs/op
BenchmarkPut/128B-4 100000 17492 ns/op 7.32 MB/s 441 B/op 6 allocs/op
BenchmarkPut/256B-4 100000 17234 ns/op 14.85 MB/s 571 B/op 6 allocs/op
BenchmarkPut/512B-4 100000 22837 ns/op 22.42 MB/s 861 B/op 6 allocs/op
BenchmarkPut/1K-4 50000 30333 ns/op 33.76 MB/s 1443 B/op 6 allocs/op
BenchmarkPut/2K-4 30000 45304 ns/op 45.21 MB/s 2606 B/op 6 allocs/op
BenchmarkPut/4K-4 20000 83953 ns/op 48.79 MB/s 5187 B/op 6 allocs/op
BenchmarkPut/8K-4 10000 142142 ns/op 57.63 MB/s 9845 B/op 6 allocs/op
BenchmarkPut/16K-4 5000 206722 ns/op 79.26 MB/s 18884 B/op 6 allocs/op
BenchmarkPut/32K-4 5000 361108 ns/op 90.74 MB/s 41582 B/op 7 allocs/op
BenchmarkScan-4 1000000 1679 ns/op 408 B/op 16 allocs/op
PASS
```
For 128B values:
* ~400,000 reads/sec
* ~130,000 writes/sec
* ~200,000 reads/sec
* ~50,000 writes/sec
The full benchmark above shows linear performance as you increase key/value sizes.
The full benchmark above shows linear performance as you increase key/value sizes. Memory pooling starts to become advantageous for larger values.
## Stargazers over time
[![Stargazers over time](https://starcharts.herokuapp.com/prologic/bitcask.svg)](https://starcharts.herokuapp.com/prologic/bitcask)
## Support
Support the ongoing development of Bitcask!
**Sponser**
- Become a [Sponser](https://www.patreon.com/prologic)
## Contributors
Thank you to all those that have contributed to this project, battle-tested it, used it in their own projects or products, fixed bugs, improved performance and even fix tiny typos in documentation! Thank you and keep contributing!
You can find an [AUTHORS](/AUTHORS) file where we keep a list of contributors to the project. If you contriibute a PR please consider adding your name there. There is also Github's own [Contributors](https://github.com/prologic/bitcask/graphs/contributors) statistics.
## License
bitcask is licensed under the [MIT License](https://github.com/prologic/bitcask/blob/master/LICENSE)
bitcask is licensed under the term of the [MIT License](https://github.com/prologic/bitcask/blob/master/LICENSE)

View File

@@ -1,18 +1,18 @@
package bitcask
import (
"encoding/json"
"errors"
"hash/crc32"
"io"
"io/ioutil"
"os"
"path"
"path/filepath"
"strings"
"sync"
"github.com/gofrs/flock"
"github.com/prologic/trie"
art "github.com/plar/go-adaptive-radix-tree"
"github.com/prologic/bitcask/internal"
)
@@ -46,11 +46,37 @@ type Bitcask struct {
*flock.Flock
config *config
options []Option
path string
curr *internal.Datafile
keydir *internal.Keydir
datafiles []*internal.Datafile
trie *trie.Trie
datafiles map[int]*internal.Datafile
trie art.Tree
}
// Stats is a struct returned by Stats() on an open Bitcask instance
type Stats struct {
Datafiles int
Keys int
Size int64
}
// Stats returns statistics about the database including the number of
// data files, keys and overall size on disk of the data
func (b *Bitcask) Stats() (stats Stats, err error) {
var size int64
size, err = internal.DirSize(b.path)
if err != nil {
return
}
stats.Datafiles = len(b.datafiles)
b.mu.RLock()
stats.Keys = b.trie.Size()
b.mu.RUnlock()
stats.Size = size
return
}
// Close closes the database and removes the lock. It is important to call
@@ -62,9 +88,25 @@ func (b *Bitcask) Close() error {
os.Remove(b.Flock.Path())
}()
for _, df := range b.datafiles {
df.Close()
f, err := os.OpenFile(filepath.Join(b.path, "index"), os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0600)
if err != nil {
return err
}
defer f.Close()
if err := internal.WriteIndex(b.trie, f); err != nil {
return err
}
if err := f.Sync(); err != nil {
return err
}
for _, df := range b.datafiles {
if err := df.Close(); err != nil {
return err
}
}
return b.curr.Close()
}
@@ -75,14 +117,18 @@ 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)
if !ok {
b.mu.RLock()
value, found := b.trie.Search(key)
b.mu.RUnlock()
if !found {
return nil, ErrKeyNotFound
}
item := value.(internal.Item)
if item.FileID == b.curr.FileID() {
df = b.curr
} else {
@@ -103,13 +149,15 @@ 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 {
_, ok := b.keydir.Get(key)
return ok
func (b *Bitcask) Has(key []byte) bool {
b.mu.RLock()
_, found := b.trie.Search(key)
b.mu.RUnlock()
return found
}
// 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
}
@@ -122,22 +170,31 @@ func (b *Bitcask) Put(key string, value []byte) error {
return err
}
item := b.keydir.Add(key, b.curr.FileID(), offset, n)
b.trie.Add(key, item)
if b.config.sync {
if err := b.curr.Sync(); err != nil {
return err
}
}
item := internal.Item{FileID: b.curr.FileID(), Offset: offset, Size: n}
b.mu.Lock()
b.trie.Insert(key, item)
b.mu.Unlock()
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.mu.Lock()
b.trie.Delete(key)
b.mu.Unlock()
return nil
}
@@ -145,39 +202,69 @@ 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)
for _, key := range keys {
if err := f(key); err != nil {
return err
func (b *Bitcask) Scan(prefix []byte, f func(key []byte) error) (err error) {
b.trie.ForEachPrefix(prefix, func(node art.Node) bool {
// Skip the root node
if len(node.Key()) == 0 {
return true
}
}
return nil
if err = f(node.Key()); err != nil {
return false
}
return true
})
return
}
// Len returns the total number of keys in the database
func (b *Bitcask) Len() int {
return b.keydir.Len()
b.mu.RLock()
defer b.mu.RUnlock()
return b.trie.Size()
}
// Keys returns all keys in the database as a channel of string(s)
func (b *Bitcask) Keys() chan string {
return b.keydir.Keys()
// Keys returns all keys in the database as a channel of keys
func (b *Bitcask) Keys() chan []byte {
ch := make(chan []byte)
go func() {
b.mu.RLock()
defer b.mu.RUnlock()
for it := b.trie.Iterator(); it.HasNext(); {
node, _ := it.Next()
// Skip the root node
if len(node.Key()) == 0 {
continue
}
ch <- node.Key()
}
close(ch)
}()
return ch
}
// 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 {
for key := range b.keydir.Keys() {
if err := f(key); err != nil {
return err
func (b *Bitcask) Fold(f func(key []byte) error) error {
b.mu.RLock()
defer b.mu.RUnlock()
b.trie.ForEach(func(node art.Node) bool {
if err := f(node.Key()); err != nil {
return false
}
}
return true
})
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()
@@ -188,14 +275,16 @@ func (b *Bitcask) put(key string, value []byte) (int64, int64, error) {
return -1, 0, err
}
df, err := internal.NewDatafile(b.path, b.curr.FileID(), true)
id := b.curr.FileID()
df, err := internal.NewDatafile(b.path, id, true)
if err != nil {
return -1, 0, err
}
b.datafiles = append(b.datafiles, df)
b.datafiles[id] = df
id := b.curr.FileID() + 1
id = b.curr.FileID() + 1
curr, err := internal.NewDatafile(b.path, id, false)
if err != nil {
return -1, 0, err
@@ -207,185 +296,83 @@ func (b *Bitcask) put(key string, value []byte) (int64, int64, error) {
return b.curr.Write(e)
}
// Merge merges all datafiles in the database creating hint files for faster
// startup. Old keys are squashed and deleted keys removes. Call this function
// periodically to reclaim disk space.
func Merge(path string, force bool) error {
fns, err := internal.GetDatafiles(path)
if err != nil {
return err
}
ids, err := internal.ParseIds(fns)
if err != nil {
return err
}
// Do not merge if we only have 1 Datafile
if len(ids) <= 1 {
return nil
}
// Don't merge the Active Datafile (the last one)
fns = fns[:len(fns)-1]
ids = ids[:len(ids)-1]
temp, err := ioutil.TempDir(path, "merge")
if err != nil {
return err
}
for i, fn := range fns {
// Don't merge Datafiles whose .hint files we've already generated
// (they are already merged); unless we set the force flag to true
// (forcing a re-merge).
if filepath.Ext(fn) == ".hint" && !force {
// Already merged
continue
}
id := ids[i]
keydir := internal.NewKeydir()
df, err := internal.NewDatafile(path, id, true)
if err != nil {
return err
}
defer df.Close()
for {
e, n, err := df.Read()
if err != nil {
if err == io.EOF {
break
}
return err
}
// Tombstone value (deleted key)
if len(e.Value) == 0 {
keydir.Delete(e.Key)
continue
}
keydir.Add(e.Key, ids[i], e.Offset, n)
}
tempdf, err := internal.NewDatafile(temp, id, false)
if err != nil {
return err
}
defer tempdf.Close()
for key := range keydir.Keys() {
item, _ := keydir.Get(key)
e, err := df.ReadAt(item.Offset, item.Size)
if err != nil {
return err
}
_, _, err = tempdf.Write(e)
if err != nil {
return err
}
}
err = tempdf.Close()
func (b *Bitcask) readConfig() error {
if internal.Exists(filepath.Join(b.path, "config.json")) {
data, err := ioutil.ReadFile(filepath.Join(b.path, "config.json"))
if err != nil {
return err
}
err = df.Close()
if err != nil {
return err
}
err = os.Rename(tempdf.Name(), df.Name())
if err != nil {
return err
}
hint := strings.TrimSuffix(df.Name(), ".data") + ".hint"
err = keydir.Save(hint)
if err != nil {
if err := json.Unmarshal(data, &b.config); err != nil {
return err
}
}
return nil
}
// Open opens the database at the given path with optional options.
// Options can be provided with the `WithXXX` functions that provide
// configuration options as functions.
func Open(path string, options ...Option) (*Bitcask, error) {
if err := os.MkdirAll(path, 0755); err != nil {
return nil, err
}
err := Merge(path, false)
func (b *Bitcask) writeConfig() error {
data, err := json.Marshal(b.config)
if err != nil {
return nil, err
return err
}
return ioutil.WriteFile(filepath.Join(b.path, "config.json"), data, 0600)
}
fns, err := internal.GetDatafiles(path)
func (b *Bitcask) reopen() error {
b.mu.Lock()
defer b.mu.Unlock()
fns, err := internal.GetDatafiles(b.path)
if err != nil {
return nil, err
return err
}
ids, err := internal.ParseIds(fns)
if err != nil {
return nil, err
return err
}
var datafiles []*internal.Datafile
datafiles := make(map[int]*internal.Datafile, len(ids))
keydir := internal.NewKeydir()
trie := trie.New()
for i, fn := range fns {
df, err := internal.NewDatafile(path, ids[i], true)
for _, id := range ids {
df, err := internal.NewDatafile(b.path, id, true)
if err != nil {
return nil, err
return err
}
datafiles = append(datafiles, df)
datafiles[id] = df
}
if filepath.Ext(fn) == ".hint" {
f, err := os.Open(filepath.Join(path, fn))
if err != nil {
return nil, err
}
defer f.Close()
t := art.New()
hint, err := internal.NewKeydirFromBytes(f)
if err != nil {
return nil, err
}
if internal.Exists(path.Join(b.path, "index")) {
f, err := os.Open(path.Join(b.path, "index"))
if err != nil {
return err
}
defer f.Close()
for key := range hint.Keys() {
item, _ := hint.Get(key)
_ = keydir.Add(key, item.FileID, item.Offset, item.Size)
trie.Add(key, item)
}
} else {
if err := internal.ReadIndex(f, t); err != nil {
return err
}
} else {
for i, df := range datafiles {
for {
e, n, err := df.Read()
if err != nil {
if err == io.EOF {
break
}
return nil, err
return err
}
// Tombstone value (deleted key)
if len(e.Value) == 0 {
keydir.Delete(e.Key)
t.Delete(e.Key)
continue
}
item := keydir.Add(e.Key, ids[i], e.Offset, n)
trie.Add(e.Key, item)
item := internal.Item{FileID: ids[i], Offset: e.Offset, Size: n}
t.Insert(e.Key, item)
}
}
}
@@ -395,24 +382,125 @@ func Open(path string, options ...Option) (*Bitcask, error) {
id = ids[(len(ids) - 1)]
}
curr, err := internal.NewDatafile(path, id, false)
curr, err := internal.NewDatafile(b.path, id, false)
if err != nil {
return err
}
b.trie = t
b.curr = curr
b.datafiles = datafiles
return nil
}
// Merge merges all datafiles in the database. Old keys are squashed
// and deleted keys removes. Duplicate key/value pairs are also removed.
// Call this function periodically to reclaim disk space.
func (b *Bitcask) Merge() error {
// Temporary merged database path
temp, err := ioutil.TempDir(b.path, "merge")
if err != nil {
return err
}
defer os.RemoveAll(temp)
// Create a merged database
mdb, err := Open(temp, b.options...)
if err != nil {
return err
}
// Rewrite all key/value pairs into merged database
// Doing this automatically strips deleted keys and
// old key/value pairs
err = b.Fold(func(key []byte) error {
value, err := b.Get(key)
if err != nil {
return err
}
if err := mdb.Put(key, value); err != nil {
return err
}
return nil
})
if err != nil {
return err
}
err = mdb.Close()
if err != nil {
return err
}
// Close the database
err = b.Close()
if err != nil {
return err
}
// Remove all data files
files, err := ioutil.ReadDir(b.path)
if err != nil {
return err
}
for _, file := range files {
if !file.IsDir() {
err := os.RemoveAll(path.Join([]string{b.path, file.Name()}...))
if err != nil {
return err
}
}
}
// Rename all merged data files
files, err = ioutil.ReadDir(mdb.path)
if err != nil {
return err
}
for _, file := range files {
err := os.Rename(
path.Join([]string{mdb.path, file.Name()}...),
path.Join([]string{b.path, file.Name()}...),
)
if err != nil {
return err
}
}
// And finally reopen the database
return b.reopen()
}
// Open opens the database at the given path with optional options.
// Options can be provided with the `WithXXX` functions that provide
// configuration options as functions.
func Open(path string, options ...Option) (*Bitcask, error) {
var (
cfg *config
err error
)
if err := os.MkdirAll(path, 0755); err != nil {
return nil, err
}
cfg, err = getConfig(path)
if err != nil {
cfg = newDefaultConfig()
}
bitcask := &Bitcask{
Flock: flock.New(filepath.Join(path, "lock")),
config: newDefaultConfig(),
path: path,
curr: curr,
keydir: keydir,
datafiles: datafiles,
trie: trie,
Flock: flock.New(filepath.Join(path, "lock")),
config: cfg,
options: options,
path: path,
}
for _, opt := range options {
err = opt(bitcask.config)
if err != nil {
if err := opt(bitcask.config); err != nil {
return nil, err
}
}
@@ -426,5 +514,13 @@ func Open(path string, options ...Option) (*Bitcask, error) {
return nil, ErrDatabaseLocked
}
if err := bitcask.writeConfig(); err != nil {
return nil, err
}
if err := bitcask.reopen(); err != nil {
return nil, err
}
return bitcask, nil
}

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)
@@ -211,38 +238,39 @@ func TestMaxValueSize(t *testing.T) {
})
}
func TestOpenMerge(t *testing.T) {
func TestStats(t *testing.T) {
var (
db *Bitcask
err error
)
assert := assert.New(t)
testdir, err := ioutil.TempDir("", "bitcask")
assert.NoError(err)
t.Run("Setup", func(t *testing.T) {
var (
db *Bitcask
err error
)
t.Run("Open", func(t *testing.T) {
db, err = Open(testdir, WithMaxDatafileSize(32))
db, err = Open(testdir)
assert.NoError(err)
})
t.Run("Put", func(t *testing.T) {
for i := 0; i < 1024; i++ {
err = db.Put(string(i), []byte(strings.Repeat(" ", 1024)))
assert.NoError(err)
}
err := db.Put([]byte("foo"), []byte("bar"))
assert.NoError(err)
})
t.Run("Get", func(t *testing.T) {
for i := 0; i < 32; i++ {
err = db.Put(string(i), []byte(strings.Repeat(" ", 1024)))
assert.NoError(err)
val, err := db.Get(string(i))
assert.NoError(err)
assert.Equal([]byte(strings.Repeat(" ", 1024)), val)
}
val, err := db.Get([]byte("foo"))
assert.NoError(err)
assert.Equal([]byte("bar"), val)
})
t.Run("Stats", func(t *testing.T) {
stats, err := db.Stats()
assert.NoError(err)
assert.Equal(stats.Datafiles, 0)
assert.Equal(stats.Keys, 1)
})
t.Run("Sync", func(t *testing.T) {
@@ -255,34 +283,9 @@ func TestOpenMerge(t *testing.T) {
assert.NoError(err)
})
})
t.Run("Merge", func(t *testing.T) {
var (
db *Bitcask
err error
)
t.Run("Open", func(t *testing.T) {
db, err = Open(testdir)
assert.NoError(err)
})
t.Run("Get", func(t *testing.T) {
for i := 0; i < 32; i++ {
val, err := db.Get(string(i))
assert.NoError(err)
assert.Equal([]byte(strings.Repeat(" ", 1024)), val)
}
})
t.Run("Close", func(t *testing.T) {
err = db.Close()
assert.NoError(err)
})
})
}
func TestMergeOpen(t *testing.T) {
func TestMerge(t *testing.T) {
var (
db *Bitcask
err error
@@ -300,22 +303,40 @@ func TestMergeOpen(t *testing.T) {
})
t.Run("Put", func(t *testing.T) {
for i := 0; i < 1024; i++ {
err = db.Put(string(i), []byte(strings.Repeat(" ", 1024)))
err := db.Put([]byte("foo"), []byte("bar"))
assert.NoError(err)
})
s1, err := db.Stats()
assert.NoError(err)
assert.Equal(0, s1.Datafiles)
assert.Equal(1, s1.Keys)
t.Run("Put", func(t *testing.T) {
for i := 0; i < 10; i++ {
err := db.Put([]byte("foo"), []byte("bar"))
assert.NoError(err)
}
})
t.Run("Get", func(t *testing.T) {
for i := 0; i < 32; i++ {
err = db.Put(string(i), []byte(strings.Repeat(" ", 1024)))
assert.NoError(err)
val, err := db.Get(string(i))
assert.NoError(err)
assert.Equal([]byte(strings.Repeat(" ", 1024)), val)
}
s2, err := db.Stats()
assert.NoError(err)
assert.Equal(5, s2.Datafiles)
assert.Equal(1, s2.Keys)
assert.True(s2.Size > s1.Size)
t.Run("Merge", func(t *testing.T) {
err := db.Merge()
assert.NoError(err)
})
s3, err := db.Stats()
assert.NoError(err)
assert.Equal(1, s3.Datafiles)
assert.Equal(1, s3.Keys)
assert.True(s3.Size > s1.Size)
assert.True(s3.Size < s2.Size)
t.Run("Sync", func(t *testing.T) {
err = db.Sync()
assert.NoError(err)
@@ -326,31 +347,6 @@ func TestMergeOpen(t *testing.T) {
assert.NoError(err)
})
})
t.Run("Merge", func(t *testing.T) {
t.Run("Merge", func(t *testing.T) {
err = Merge(testdir, true)
assert.NoError(err)
})
t.Run("Open", func(t *testing.T) {
db, err = Open(testdir)
assert.NoError(err)
})
t.Run("Get", func(t *testing.T) {
for i := 0; i < 32; i++ {
val, err := db.Get(string(i))
assert.NoError(err)
assert.Equal([]byte(strings.Repeat(" ", 1024)), val)
}
})
t.Run("Close", func(t *testing.T) {
err = db.Close()
assert.NoError(err)
})
})
}
func TestConcurrent(t *testing.T) {
@@ -371,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)
})
})
@@ -384,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)
@@ -408,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)
}
@@ -451,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)
}
})
@@ -464,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)
})
}
@@ -515,12 +511,6 @@ func BenchmarkGet(b *testing.B) {
}
defer os.RemoveAll(testdir)
db, err := Open(testdir)
if err != nil {
b.Fatal(err)
}
defer db.Close()
tests := []benchmarkTestCase{
{"128B", 128},
{"256B", 256},
@@ -537,9 +527,18 @@ 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{
WithMaxKeySize(len(key)),
WithMaxValueSize(tt.size),
}
db, err := Open(testdir, options...)
if err != nil {
b.Fatal(err)
}
err = db.Put(key, value)
if err != nil {
b.Fatal(err)
@@ -551,10 +550,12 @@ 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")
}
}
b.StopTimer()
db.Close()
})
}
}
@@ -565,22 +566,9 @@ func BenchmarkPut(b *testing.B) {
b.Fatal(err)
}
testdir, err := ioutil.TempDir(currentDir, "bitcask_bench")
if err != nil {
b.Fatal(err)
}
defer os.RemoveAll(testdir)
db, err := Open(testdir)
if err != nil {
b.Fatal(err)
}
defer db.Close()
tests := []benchmarkTestCase{
{"128B", 128},
{"256B", 256},
{"512B", 512},
{"1K", 1024},
{"2K", 2048},
{"4K", 4096},
@@ -589,20 +577,43 @@ func BenchmarkPut(b *testing.B) {
{"32K", 32768},
}
for _, tt := range tests {
b.Run(tt.name, func(b *testing.B) {
b.SetBytes(int64(tt.size))
variants := map[string][]Option{
"NoSync": []Option{
WithSync(false),
},
"Sync": []Option{
WithSync(true),
},
}
key := "foo"
value := []byte(strings.Repeat(" ", tt.size))
b.ResetTimer()
for i := 0; i < b.N; i++ {
err := db.Put(key, value)
if err != nil {
b.Fatal(err)
for name, options := range variants {
testdir, err := ioutil.TempDir(currentDir, "bitcask_bench")
if err != nil {
b.Fatal(err)
}
defer os.RemoveAll(testdir)
db, err := Open(testdir, options...)
if err != nil {
b.Fatal(err)
}
defer db.Close()
for _, tt := range tests {
b.Run(tt.name+name, func(b *testing.B) {
b.SetBytes(int64(tt.size))
key := []byte("foo")
value := []byte(strings.Repeat(" ", tt.size))
b.ResetTimer()
for i := 0; i < b.N; i++ {
err := db.Put(key, value)
if err != nil {
b.Fatal(err)
}
}
}
})
})
}
}
}
@@ -629,30 +640,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))
}

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

142
cmd/bitcask/export.go Normal file
View File

@@ -0,0 +1,142 @@
package main
import (
"encoding/base64"
"encoding/json"
"errors"
"io"
"os"
log "github.com/sirupsen/logrus"
"github.com/spf13/cobra"
"github.com/spf13/viper"
"github.com/prologic/bitcask"
)
var errNotAllDataWritten = errors.New("error: not all data written")
var exportCmd = &cobra.Command{
Use: "export",
Aliases: []string{"backup", "dump"},
Short: "Export a database",
Long: `This command allows you to export or dump/backup a database's
key/values into a long-term portable archival format suitable for backup and
restore purposes or migrating from older on-disk formats of Bitcask.
All key/value pairs are base64 encoded and serialized as JSON one pair per
line to form an output stream to either standard output or a file. You can
optionally compress the output with standard compression tools such as gzip.`,
Args: cobra.RangeArgs(0, 1),
Run: func(cmd *cobra.Command, args []string) {
var output string
path := viper.GetString("path")
if len(args) == 1 {
output = args[0]
} else {
output = "-"
}
os.Exit(export(path, output))
},
}
func init() {
RootCmd.AddCommand(exportCmd)
exportCmd.PersistentFlags().IntP(
"with-max-datafile-size", "", bitcask.DefaultMaxDatafileSize,
"Maximum size of each datafile",
)
exportCmd.PersistentFlags().IntP(
"with-max-key-size", "", bitcask.DefaultMaxKeySize,
"Maximum size of each key",
)
exportCmd.PersistentFlags().IntP(
"with-max-value-size", "", bitcask.DefaultMaxValueSize,
"Maximum size of each value",
)
}
type kvPair struct {
Key string `json:"key"`
Value string `json:"value"`
}
func export(path, output string) int {
var (
err error
w io.WriteCloser
)
db, err := bitcask.Open(path)
if err != nil {
log.WithError(err).Error("error opening database")
return 1
}
defer db.Close()
if output == "-" {
w = os.Stdout
} else {
w, err = os.OpenFile(output, os.O_WRONLY|os.O_CREATE|os.O_EXCL|os.O_TRUNC, 0755)
if err != nil {
log.WithError(err).
WithField("output", output).
Error("error opening output for writing")
return 1
}
}
err = db.Fold(func(key []byte) error {
value, err := db.Get(key)
if err != nil {
log.WithError(err).
WithField("key", key).
Error("error reading key")
return err
}
kv := kvPair{
Key: base64.StdEncoding.EncodeToString([]byte(key)),
Value: base64.StdEncoding.EncodeToString(value),
}
data, err := json.Marshal(&kv)
if err != nil {
log.WithError(err).
WithField("key", key).
Error("error serialzing key")
return err
}
if n, err := w.Write(data); err != nil || n != len(data) {
if err == nil && n != len(data) {
err = errNotAllDataWritten
}
log.WithError(err).
WithField("key", key).
WithField("n", n).
Error("error writing key")
return err
}
if _, err := w.Write([]byte("\n")); err != nil {
log.WithError(err).Error("error writing newline")
return err
}
return nil
})
if err != nil {
log.WithError(err).
WithField("path", path).
WithField("output", output).
Error("error exporting keys")
return 2
}
return 0
}

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

106
cmd/bitcask/import.go Normal file
View File

@@ -0,0 +1,106 @@
package main
import (
"bufio"
"encoding/base64"
"encoding/json"
"io"
"os"
log "github.com/sirupsen/logrus"
"github.com/spf13/cobra"
"github.com/spf13/viper"
"github.com/prologic/bitcask"
)
var importCmd = &cobra.Command{
Use: "import",
Aliases: []string{"restore", "read"},
Short: "Import a database",
Long: `This command allows you to import or restore a database from a
previous export/dump using the export command either creating a new database
or adding additional key/value pairs to an existing one.`,
Args: cobra.RangeArgs(0, 1),
Run: func(cmd *cobra.Command, args []string) {
var input string
path := viper.GetString("path")
if len(args) == 1 {
input = args[0]
} else {
input = "-"
}
os.Exit(_import(path, input))
},
}
func init() {
RootCmd.AddCommand(importCmd)
}
func _import(path, input string) int {
var (
err error
r io.ReadCloser
)
db, err := bitcask.Open(path)
if err != nil {
log.WithError(err).Error("error opening database")
return 1
}
defer db.Close()
if input == "-" {
r = os.Stdin
} else {
r, err = os.Open(input)
if err != nil {
log.WithError(err).
WithField("input", input).
Error("error opening input for reading")
return 1
}
}
var kv kvPair
scanner := bufio.NewScanner(r)
for scanner.Scan() {
if err := json.Unmarshal(scanner.Bytes(), &kv); err != nil {
log.WithError(err).
WithField("input", input).
Error("error reading input")
return 2
}
key, err := base64.StdEncoding.DecodeString(kv.Key)
if err != nil {
log.WithError(err).Error("error decoding key")
return 2
}
value, err := base64.StdEncoding.DecodeString(kv.Value)
if err != nil {
log.WithError(err).Error("error decoding value")
return 2
}
if err := db.Put(key, value); err != nil {
log.WithError(err).Error("error writing key/value")
return 2
}
}
if err := scanner.Err(); err != nil {
log.WithError(err).
WithField("input", input).
Error("error reading input")
return 2
}
return 0
}

67
cmd/bitcask/initdb.go Normal file
View File

@@ -0,0 +1,67 @@
package main
import (
"os"
log "github.com/sirupsen/logrus"
"github.com/spf13/cobra"
"github.com/spf13/viper"
"github.com/prologic/bitcask"
)
var initdbCmd = &cobra.Command{
Use: "initdb",
Aliases: []string{"create", "init"},
Short: "Initialize a new database",
Long: `This initializes a new database with persisted options`,
Args: cobra.ExactArgs(0),
PreRun: func(cmd *cobra.Command, args []string) {
viper.BindPFlag("with-max-datafile-size", cmd.Flags().Lookup("with-max-datafile-size"))
viper.SetDefault("with-max-datafile-size", bitcask.DefaultMaxDatafileSize)
viper.BindPFlag("with-max-key-size", cmd.Flags().Lookup("with-max-key-size"))
viper.SetDefault("with-max-key-size", bitcask.DefaultMaxKeySize)
viper.BindPFlag("with-max-value-size", cmd.Flags().Lookup("with-max-value-size"))
viper.SetDefault("with-max-value-size", bitcask.DefaultMaxValueSize)
},
Run: func(cmd *cobra.Command, args []string) {
path := viper.GetString("path")
maxDatafileSize := viper.GetInt("with-max-datafile-size")
maxKeySize := viper.GetInt("with-max-key-size")
maxValueSize := viper.GetInt("with-max-value-size")
db, err := bitcask.Open(
path,
bitcask.WithMaxDatafileSize(maxDatafileSize),
bitcask.WithMaxKeySize(maxKeySize),
bitcask.WithMaxValueSize(maxValueSize),
)
if err != nil {
log.WithError(err).Error("error opening database")
os.Exit(1)
}
defer db.Close()
os.Exit(0)
},
}
func init() {
RootCmd.AddCommand(initdbCmd)
initdbCmd.PersistentFlags().IntP(
"with-max-datafile-size", "", bitcask.DefaultMaxDatafileSize,
"Maximum size of each datafile",
)
initdbCmd.PersistentFlags().IntP(
"with-max-key-size", "", bitcask.DefaultMaxKeySize,
"Maximum size of each key",
)
initdbCmd.PersistentFlags().IntP(
"with-max-value-size", "", bitcask.DefaultMaxValueSize,
"Maximum size of each value",
)
}

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 {

View File

@@ -20,28 +20,23 @@ keys.`,
Args: cobra.ExactArgs(0),
Run: func(cmd *cobra.Command, args []string) {
path := viper.GetString("path")
force, err := cmd.Flags().GetBool("force")
if err != nil {
log.WithError(err).Error("error parsing force flag")
os.Exit(1)
}
os.Exit(merge(path, force))
os.Exit(merge(path))
},
}
func init() {
RootCmd.AddCommand(mergeCmd)
mergeCmd.Flags().BoolP(
"force", "f", false,
"Force a re-merge even if .hint files exist",
)
}
func merge(path string, force bool) int {
err := bitcask.Merge(path, force)
func merge(path string) int {
db, err := bitcask.Open(path)
if err != nil {
log.WithError(err).Error("error opening database")
return 1
}
if err = db.Merge(); err != nil {
log.WithError(err).Error("error merging database")
return 1
}

View File

@@ -13,11 +13,11 @@ import (
"github.com/prologic/bitcask"
)
var setCmd = &cobra.Command{
Use: "set <key> [<value>]",
Aliases: []string{"add"},
Short: "Add/Set a new Key/Value pair",
Long: `This adds or sets a new key/value pair.
var putCmd = &cobra.Command{
Use: "put <key> [<value>]",
Aliases: []string{"add", "set", "store"},
Short: "Adds a new Key/Value pair",
Long: `This adds a new key/value pair or modifies an existing one.
If the value is not specified as an argument it is read from standard input.`,
Args: cobra.MinimumNArgs(1),
@@ -33,15 +33,15 @@ If the value is not specified as an argument it is read from standard input.`,
value = os.Stdin
}
os.Exit(set(path, key, value))
os.Exit(put(path, key, value))
},
}
func init() {
RootCmd.AddCommand(setCmd)
RootCmd.AddCommand(putCmd)
}
func set(path, key string, value io.Reader) int {
func put(path, key string, value io.Reader) int {
db, err := bitcask.Open(path)
if err != nil {
log.WithError(err).Error("error opening database")
@@ -55,7 +55,7 @@ func set(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

View File

@@ -14,7 +14,7 @@ import (
var scanCmd = &cobra.Command{
Use: "scan <prefix>",
Aliases: []string{"search", "find"},
Short: "Perform a prefis scan for keys",
Short: "Perform a prefix scan for keys",
Long: `This performa a prefix scan for keys starting with the given
prefix. This uses a Trie to search for matching keys and returns all matched
keys.`,
@@ -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")

55
cmd/bitcask/stats.go Normal file
View File

@@ -0,0 +1,55 @@
package main
import (
"encoding/json"
"fmt"
"os"
log "github.com/sirupsen/logrus"
"github.com/spf13/cobra"
"github.com/spf13/viper"
"github.com/prologic/bitcask"
)
var statsCmd = &cobra.Command{
Use: "stats",
Aliases: []string{},
Short: "Display statis about the Database",
Long: `This displays statistics about the Database"`,
Args: cobra.ExactArgs(0),
Run: func(cmd *cobra.Command, args []string) {
path := viper.GetString("path")
os.Exit(stats(path))
},
}
func init() {
RootCmd.AddCommand(statsCmd)
}
func stats(path string) int {
db, err := bitcask.Open(path)
if err != nil {
log.WithError(err).Error("error opening database")
return 1
}
defer db.Close()
stats, err := db.Stats()
if err != nil {
log.WithError(err).Error("error getting stats")
return 1
}
data, err := json.MarshalIndent(stats, "", " ")
if err != nil {
log.WithError(err).Error("error marshalling stats")
return 1
}
fmt.Println(string(data))
return 0
}

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)

3
doc.go Normal file
View File

@@ -0,0 +1,3 @@
// Package bitcask implements a high-performance key-value store based on a
// WAL and LSM.
package bitcask

13
doc_test.go Normal file
View File

@@ -0,0 +1,13 @@
package bitcask
func Example() {
_, _ = Open("path/to/db")
}
func Example_withOptions() {
opts := []Option{
WithMaxKeySize(1024),
WithMaxValueSize(4096),
}
_, _ = Open("path/to/db", opts...)
}

10
go.mod
View File

@@ -1,14 +1,14 @@
module github.com/prologic/bitcask
go 1.12
require (
github.com/gofrs/flock v0.7.1
github.com/gogo/protobuf v1.2.1
github.com/golang/protobuf v1.3.2
github.com/konsorten/go-windows-terminal-sequences v1.0.2 // indirect
github.com/magiconair/properties v1.8.1 // indirect
github.com/pelletier/go-toml v1.4.0 // indirect
github.com/pkg/errors v0.8.1
github.com/prologic/trie v0.0.0-20190322091023-3972df81f9b5
github.com/plar/go-adaptive-radix-tree v1.0.1
github.com/sirupsen/logrus v1.4.2
github.com/spf13/afero v1.2.2 // indirect
github.com/spf13/cobra v0.0.5
@@ -17,7 +17,7 @@ require (
github.com/spf13/viper v1.4.0
github.com/stretchr/testify v1.3.0
github.com/tidwall/redcon v1.0.0
golang.org/x/exp v0.0.0-20190718202018-cfdd5522f6f6
golang.org/x/sys v0.0.0-20190712062909-fae7ac547cb7 // indirect
golang.org/x/exp v0.0.0-20190731235908-ec7cb31e5a56
golang.org/x/sys v0.0.0-20190804053845-51ab0e2deafa // indirect
golang.org/x/text v0.3.2 // indirect
)

23
go.sum
View File

@@ -17,7 +17,6 @@ github.com/coreos/go-semver v0.2.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3Ee
github.com/coreos/go-systemd v0.0.0-20190321100706-95778dfbb74e/go.mod h1:F5haX7vjVVG0kc13fIWeqUViNPyEJxv/OmvnBo0Yme4=
github.com/coreos/pkg v0.0.0-20180928190104-399ea9e2e55f/go.mod h1:E3G3o1h8I7cfcXa63jLwjI0eiQQMgzzUDFVpN/nH/eA=
github.com/cpuguy83/go-md2man v1.0.10/go.mod h1:SmD6nW6nTyfqj6ABTjUi3V3JVMnlJmwcJI5acqYI6dE=
github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -33,16 +32,12 @@ github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/me
github.com/gofrs/flock v0.7.1 h1:DP+LD/t0njgoPBvT5MJLeliUIVQR03hiKR6vezdwHlc=
github.com/gofrs/flock v0.7.1/go.mod h1:F1TvTiK9OcQqauNUHlbJvyl9Qa1QvF/gOUDKA14jxHU=
github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
github.com/gogo/protobuf v1.2.1 h1:/s5zKNz0uPFCZ5hddgPdo2TK2TVrUNMn0OOX8/aZMTE=
github.com/gogo/protobuf v1.2.1/go.mod h1:hp+jE20tsWTFYpLwKvXlhS1hjn+gTNwPg2I6zVXpSg4=
github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
github.com/golang/groupcache v0.0.0-20190129154638-5b532d6fd5ef/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
github.com/golang/mock v1.1.1/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A=
github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
github.com/golang/protobuf v1.3.1 h1:YF8+flBXS5eO826T4nzqPrxfhQThhXl0YzfuUPu4SBg=
github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
github.com/golang/protobuf v1.3.2 h1:6nsPYzhq5kReh6QImI3k5qWzO4PEbvbIW2cwSfR/6xs=
github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
github.com/google/btree v1.0.0/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
github.com/gorilla/websocket v1.4.0/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoAMk2YaljkQ=
@@ -57,7 +52,6 @@ github.com/jonboulle/clockwork v0.1.0/go.mod h1:Ii8DK3G1RaLaWxj9trq07+26W01tbo22
github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w=
github.com/kisielk/errcheck v1.1.0/go.mod h1:EZBBE59ingxPouuu3KfxchcWSUPOHkagtvWXihfKN4Q=
github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
github.com/konsorten/go-windows-terminal-sequences v1.0.1 h1:mweAR1A6xJ3oS2pRaGiHgQ4OO8tzTaLawm8vnODuwDk=
github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
github.com/konsorten/go-windows-terminal-sequences v1.0.2 h1:DB17ag19krx9CFsz4o3enTrPXyIXCl+2iCXH/aMAp9s=
github.com/konsorten/go-windows-terminal-sequences v1.0.2/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
@@ -84,10 +78,10 @@ github.com/pelletier/go-toml v1.4.0/go.mod h1:PN7xzY2wHTK0K9p34ErDQMlFxa51Fk0OUr
github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=
github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
github.com/plar/go-adaptive-radix-tree v1.0.1 h1:J+2qrXaKWLACw59s8SlTVYYxWjlUr/BlCsfkAzn96/0=
github.com/plar/go-adaptive-radix-tree v1.0.1/go.mod h1:Ot8d28EII3i7Lv4PSvBlF8ejiD/CtRYDuPsySJbSaK8=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/prologic/trie v0.0.0-20190322091023-3972df81f9b5 h1:H8dTZzU3aWNQnuRyiT45J9szv7EFakAhFzsFq27t3Uo=
github.com/prologic/trie v0.0.0-20190322091023-3972df81f9b5/go.mod h1:LFuDmpHJGmciXd8Rl5YMhVlLMps9gz2GtYLzwxrFhzs=
github.com/prometheus/client_golang v0.9.1/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw=
github.com/prometheus/client_golang v0.9.3/go.mod h1:/TN21ttK/J9q6uSwhBd54HahCDft0ttaMvbicHlPoso=
github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo=
@@ -118,7 +112,6 @@ github.com/spf13/jwalterweatherman v1.1.0 h1:ue6voC5bR5F8YxI5S67j9i582FU4Qvo2bmq
github.com/spf13/jwalterweatherman v1.1.0/go.mod h1:aNWZUN0dPAAO/Ljvb5BEdw96iTZ0EXowPYD95IqWIGo=
github.com/spf13/pflag v1.0.3 h1:zPAT6CGy6wXeQ7NtTnaTerfKOsV6V6F8agHXFiazDkg=
github.com/spf13/pflag v1.0.3/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnInEg4=
github.com/spf13/viper v1.3.2 h1:VUFqw5KcqRf7i70GOzW7N+Q7+gxVBkSSqiXB12+JQ4M=
github.com/spf13/viper v1.3.2/go.mod h1:ZiWeW+zYFKm7srdB9IoDzzZXaJaI5eL9QjNiN/DMA2s=
github.com/spf13/viper v1.4.0 h1:yXHLWeravcrgGyFSyCgdYpXQ9dR9c/WED3pg1RhxqEU=
github.com/spf13/viper v1.4.0/go.mod h1:PTJ7Z/lr49W6bUbkmS1V3by4uWynFiR9p7+dSq/yZzE=
@@ -139,12 +132,11 @@ go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
go.uber.org/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0=
go.uber.org/zap v1.10.0/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q=
golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
golang.org/x/crypto v0.0.0-20181203042331-505ab145d0a9 h1:mKdxBk7AujPs8kU4m80U72y/zjbZ3UcXC7dClwKbUI0=
golang.org/x/crypto v0.0.0-20181203042331-505ab145d0a9/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
golang.org/x/crypto v0.0.0-20190510104115-cbcb75029529/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
golang.org/x/exp v0.0.0-20190718202018-cfdd5522f6f6 h1:1xlTfQXUyQvvbx4TFsXRgoj9r0YUwcqq9XGX9u9OODU=
golang.org/x/exp v0.0.0-20190718202018-cfdd5522f6f6/go.mod h1:JhuoJpWY28nO4Vef9tZUw9qufEGTyX1+7lmHxV5q5G4=
golang.org/x/exp v0.0.0-20190731235908-ec7cb31e5a56 h1:estk1glOnSVeJ9tdEZZc5mAMDZk5lNJNyJ6DvrBkTEU=
golang.org/x/exp v0.0.0-20190731235908-ec7cb31e5a56/go.mod h1:JhuoJpWY28nO4Vef9tZUw9qufEGTyX1+7lmHxV5q5G4=
golang.org/x/image v0.0.0-20190227222117-0694c2d4d067/go.mod h1:kZ7UVZpmo3dzQBMxlp+ypCbDeSB+sBbTgSJuh5dn5js=
golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
@@ -164,13 +156,13 @@ golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5h
golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20181107165924-66b7b1311ac8/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20181116152217-5ac8a444bdc5/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20181205085412-a5c9d58dba9a h1:1n5lsVfiQW3yfsRGu98756EH1YthsFqr/5mxHduZW2A=
golang.org/x/sys v0.0.0-20181205085412-a5c9d58dba9a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20190422165155-953cdadca894 h1:Cz4ceDQGXuKRnVBDTS23GTn/pU5OE2C0WrNTOYK1Uuc=
golang.org/x/sys v0.0.0-20190422165155-953cdadca894/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20190712062909-fae7ac547cb7 h1:LepdCS8Gf/MVejFIt8lsiexZATdoGVyp5bcyS+rYoUI=
golang.org/x/sys v0.0.0-20190712062909-fae7ac547cb7/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20190804053845-51ab0e2deafa h1:KIDDMLT1O0Nr7TSxp8xM5tJcdn8tgyAONntO829og1M=
golang.org/x/sys v0.0.0-20190804053845-51ab0e2deafa/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/text v0.3.0 h1:g61tztE5qeGQ89tm6NTjjM9VPIm088od1l6aSorWRWg=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/text v0.3.2 h1:tW2bmiBqwgJj/UpqtC8EpXEZVYOwU0yG4iWbprSVAcs=
@@ -186,7 +178,6 @@ google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8/go.mod h1:JiN7NxoA
google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=
google.golang.org/grpc v1.21.0/go.mod h1:oYelfM1adQP15Ek0mdvEgi9Df8B9CZIaU1084ijfRaM=
gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY=
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

105
internal/codec.go Normal file
View File

@@ -0,0 +1,105 @@
package internal
import (
"bufio"
"encoding/binary"
"io"
"github.com/pkg/errors"
)
const (
KeySize = 4
ValueSize = 8
checksumSize = 4
)
// NewEncoder creates a streaming Entry encoder.
func NewEncoder(w io.Writer) *Encoder {
return &Encoder{w: bufio.NewWriter(w)}
}
// Encoder wraps an underlying io.Writer and allows you to stream
// Entry encodings on it.
type Encoder struct {
w *bufio.Writer
}
// Encode takes any Entry and streams it to the underlying writer.
// Messages are framed with a key-length and value-length prefix.
func (e *Encoder) Encode(msg Entry) (int64, error) {
var bufKeyValue = make([]byte, KeySize+ValueSize)
binary.BigEndian.PutUint32(bufKeyValue[:KeySize], uint32(len(msg.Key)))
binary.BigEndian.PutUint64(bufKeyValue[KeySize:KeySize+ValueSize], uint64(len(msg.Value)))
if _, err := e.w.Write(bufKeyValue); err != nil {
return 0, errors.Wrap(err, "failed writing key & value length prefix")
}
if _, err := e.w.Write(msg.Key); err != nil {
return 0, errors.Wrap(err, "failed writing key data")
}
if _, err := e.w.Write(msg.Value); err != nil {
return 0, errors.Wrap(err, "failed writing value data")
}
bufChecksumSize := bufKeyValue[:checksumSize]
binary.BigEndian.PutUint32(bufChecksumSize, msg.Checksum)
if _, err := e.w.Write(bufChecksumSize); err != nil {
return 0, errors.Wrap(err, "failed writing checksum data")
}
if err := e.w.Flush(); err != nil {
return 0, errors.Wrap(err, "failed flushing data")
}
return int64(KeySize + ValueSize + len(msg.Key) + len(msg.Value) + checksumSize), nil
}
// NewDecoder creates a streaming Entry decoder.
func NewDecoder(r io.Reader) *Decoder {
return &Decoder{r: r}
}
// Decoder wraps an underlying io.Reader and allows you to stream
// Entry decodings on it.
type Decoder struct {
r io.Reader
}
func (d *Decoder) Decode(v *Entry) (int64, error) {
prefixBuf := make([]byte, KeySize+ValueSize)
_, err := io.ReadFull(d.r, prefixBuf)
if err != nil {
return 0, err
}
actualKeySize, actualValueSize := GetKeyValueSizes(prefixBuf)
buf := make([]byte, actualKeySize+actualValueSize+checksumSize)
if _, err = io.ReadFull(d.r, buf); err != nil {
return 0, errors.Wrap(translateError(err), "failed reading saved data")
}
DecodeWithoutPrefix(buf, actualKeySize, v)
return int64(KeySize + ValueSize + actualKeySize + actualValueSize + checksumSize), nil
}
func GetKeyValueSizes(buf []byte) (uint64, uint64) {
actualKeySize := binary.BigEndian.Uint32(buf[:KeySize])
actualValueSize := binary.BigEndian.Uint64(buf[KeySize:])
return uint64(actualKeySize), actualValueSize
}
func DecodeWithoutPrefix(buf []byte, valueOffset uint64, v *Entry) {
v.Key = buf[:valueOffset]
v.Value = buf[valueOffset : len(buf)-checksumSize]
v.Checksum = binary.BigEndian.Uint32(buf[len(buf)-checksumSize:])
}
func translateError(err error) error {
if err == io.EOF {
return io.ErrUnexpectedEOF
}
return err
}

110
internal/codec_index.go Normal file
View File

@@ -0,0 +1,110 @@
package internal
import (
"encoding/binary"
"io"
art "github.com/plar/go-adaptive-radix-tree"
)
const (
Int32Size = 4
Int64Size = 8
FileIDSize = Int32Size
OffsetSize = Int64Size
SizeSize = Int64Size
)
func ReadBytes(r io.Reader) ([]byte, error) {
s := make([]byte, Int32Size)
_, err := io.ReadFull(r, s)
if err != nil {
return nil, err
}
size := binary.BigEndian.Uint32(s)
b := make([]byte, size)
_, err = io.ReadFull(r, b)
if err != nil {
return nil, err
}
return b, nil
}
func WriteBytes(b []byte, w io.Writer) (int, error) {
s := make([]byte, Int32Size)
binary.BigEndian.PutUint32(s, uint32(len(b)))
n, err := w.Write(s)
if err != nil {
return n, err
}
m, err := w.Write(b)
if err != nil {
return (n + m), err
}
return (n + m), nil
}
func ReadItem(r io.Reader) (Item, error) {
buf := make([]byte, (FileIDSize + OffsetSize + SizeSize))
_, err := io.ReadFull(r, buf)
if err != nil {
return Item{}, err
}
return Item{
FileID: int(binary.BigEndian.Uint32(buf[:FileIDSize])),
Offset: int64(binary.BigEndian.Uint64(buf[FileIDSize:(FileIDSize + OffsetSize)])),
Size: int64(binary.BigEndian.Uint64(buf[(FileIDSize + OffsetSize):])),
}, nil
}
func WriteItem(item Item, w io.Writer) (int, error) {
buf := make([]byte, (FileIDSize + OffsetSize + SizeSize))
binary.BigEndian.PutUint32(buf[:FileIDSize], uint32(item.FileID))
binary.BigEndian.PutUint64(buf[FileIDSize:(FileIDSize+OffsetSize)], uint64(item.Offset))
binary.BigEndian.PutUint64(buf[(FileIDSize+OffsetSize):], uint64(item.Size))
n, err := w.Write(buf)
if err != nil {
return 0, err
}
return n, nil
}
func ReadIndex(r io.Reader, t art.Tree) error {
for {
key, err := ReadBytes(r)
if err != nil {
if err == io.EOF {
break
}
return err
}
item, err := ReadItem(r)
if err != nil {
return err
}
t.Insert(key, item)
}
return nil
}
func WriteIndex(t art.Tree, w io.Writer) (err error) {
t.ForEach(func(node art.Node) bool {
_, err = WriteBytes(node.Key(), w)
if err != nil {
return false
}
item := node.Value().(Item)
_, err := WriteItem(item, w)
if err != nil {
return false
}
return true
})
return
}

View File

@@ -1,7 +1,6 @@
package internal
import (
"bytes"
"fmt"
"os"
"path/filepath"
@@ -9,9 +8,6 @@ import (
"github.com/pkg/errors"
"golang.org/x/exp/mmap"
pb "github.com/prologic/bitcask/internal/proto"
"github.com/prologic/bitcask/internal/streampb"
)
const (
@@ -21,6 +17,8 @@ const (
var (
ErrReadonly = errors.New("error: read only datafile")
ErrReadError = errors.New("error: read error")
mxMemPool sync.RWMutex
)
type Datafile struct {
@@ -31,8 +29,8 @@ type Datafile struct {
ra *mmap.ReaderAt
w *os.File
offset int64
dec *streampb.Decoder
enc *streampb.Encoder
dec *Decoder
enc *Encoder
}
func NewDatafile(path string, id int, readonly bool) (*Datafile, error) {
@@ -68,8 +66,8 @@ func NewDatafile(path string, id int, readonly bool) (*Datafile, error) {
offset := stat.Size()
dec := streampb.NewDecoder(r)
enc := streampb.NewEncoder(w)
dec := NewDecoder(r)
enc := NewEncoder(w)
return &Datafile{
id: id,
@@ -121,7 +119,7 @@ func (df *Datafile) Size() int64 {
return df.offset
}
func (df *Datafile) Read() (e pb.Entry, n int64, err error) {
func (df *Datafile) Read() (e Entry, n int64, err error) {
df.Lock()
defer df.Unlock()
@@ -133,7 +131,7 @@ func (df *Datafile) Read() (e pb.Entry, n int64, err error) {
return
}
func (df *Datafile) ReadAt(index, size int64) (e pb.Entry, err error) {
func (df *Datafile) ReadAt(index, size int64) (e Entry, err error) {
var n int
b := make([]byte, size)
@@ -151,13 +149,13 @@ func (df *Datafile) ReadAt(index, size int64) (e pb.Entry, err error) {
return
}
buf := bytes.NewBuffer(b)
dec := streampb.NewDecoder(buf)
_, err = dec.Decode(&e)
valueOffset, _ := GetKeyValueSizes(b)
DecodeWithoutPrefix(b[KeySize+ValueSize:], valueOffset, &e)
return
}
func (df *Datafile) Write(e pb.Entry) (int64, int64, error) {
func (df *Datafile) Write(e Entry) (int64, int64, error) {
if df.w == nil {
return -1, 0, ErrReadonly
}
@@ -167,7 +165,7 @@ func (df *Datafile) Write(e pb.Entry) (int64, int64, error) {
e.Offset = df.offset
n, err := df.enc.Encode(&e)
n, err := df.enc.Encode(e)
if err != nil {
return -1, 0, err
}

View File

@@ -2,14 +2,20 @@ package internal
import (
"hash/crc32"
pb "github.com/prologic/bitcask/internal/proto"
)
func NewEntry(key string, value []byte) pb.Entry {
// Entry represents a key/value in the database
type Entry struct {
Checksum uint32
Key []byte
Offset int64
Value []byte
}
func NewEntry(key, value []byte) Entry {
checksum := crc32.ChecksumIEEE(value)
return pb.Entry{
return Entry{
Checksum: checksum,
Key: key,
Value: value,

7
internal/item.go Normal file
View File

@@ -0,0 +1,7 @@
package internal
type Item struct {
FileID int `json:"fileid"`
Offset int64 `json:"offset"`
Size int64 `json:"size"`
}

View File

@@ -1,101 +0,0 @@
package internal
import (
"bytes"
"encoding/gob"
"io"
"io/ioutil"
"sync"
)
type Item struct {
FileID int
Offset int64
Size int64
}
type Keydir struct {
sync.RWMutex
kv map[string]Item
}
func NewKeydir() *Keydir {
return &Keydir{
kv: make(map[string]Item),
}
}
func (k *Keydir) Add(key string, fileid int, offset, size int64) Item {
item := Item{
FileID: fileid,
Offset: offset,
Size: size,
}
k.Lock()
k.kv[key] = item
k.Unlock()
return item
}
func (k *Keydir) Get(key string) (Item, bool) {
k.RLock()
defer k.RUnlock()
item, ok := k.kv[key]
return item, ok
}
func (k *Keydir) Delete(key string) {
k.Lock()
defer k.Unlock()
delete(k.kv, key)
}
func (k *Keydir) Len() int {
return len(k.kv)
}
func (k *Keydir) Keys() chan string {
ch := make(chan string)
go func() {
k.RLock()
defer k.RUnlock()
for key := range k.kv {
ch <- key
}
close(ch)
}()
return ch
}
func (k *Keydir) Bytes() ([]byte, error) {
var buf bytes.Buffer
enc := gob.NewEncoder(&buf)
err := enc.Encode(k.kv)
if err != nil {
return nil, err
}
return buf.Bytes(), nil
}
func (k *Keydir) Save(fn string) error {
data, err := k.Bytes()
if err != nil {
return err
}
return ioutil.WriteFile(fn, data, 0644)
}
func NewKeydirFromBytes(r io.Reader) (*Keydir, error) {
k := NewKeydir()
dec := gob.NewDecoder(r)
err := dec.Decode(&k.kv)
if err != nil {
return nil, err
}
return k, nil
}

View File

@@ -1,3 +0,0 @@
package proto
//go:generate protoc --go_out=. entry.proto

View File

@@ -1,99 +0,0 @@
// Code generated by protoc-gen-go. DO NOT EDIT.
// source: entry.proto
package proto
import proto "github.com/golang/protobuf/proto"
import fmt "fmt"
import math "math"
// Reference imports to suppress errors if they are not otherwise used.
var _ = proto.Marshal
var _ = fmt.Errorf
var _ = math.Inf
// This is a compile-time assertion to ensure that this generated file
// is compatible with the proto package it is being compiled against.
// A compilation error at this line likely means your copy of the
// proto package needs to be updated.
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"`
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:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
}
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}
}
func (m *Entry) XXX_Unmarshal(b []byte) error {
return xxx_messageInfo_Entry.Unmarshal(m, b)
}
func (m *Entry) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
return xxx_messageInfo_Entry.Marshal(b, m, deterministic)
}
func (dst *Entry) XXX_Merge(src proto.Message) {
xxx_messageInfo_Entry.Merge(dst, src)
}
func (m *Entry) XXX_Size() int {
return xxx_messageInfo_Entry.Size(m)
}
func (m *Entry) XXX_DiscardUnknown() {
xxx_messageInfo_Entry.DiscardUnknown(m)
}
var xxx_messageInfo_Entry proto.InternalMessageInfo
func (m *Entry) GetChecksum() uint32 {
if m != nil {
return m.Checksum
}
return 0
}
func (m *Entry) GetKey() string {
if m != nil {
return m.Key
}
return ""
}
func (m *Entry) GetOffset() int64 {
if m != nil {
return m.Offset
}
return 0
}
func (m *Entry) GetValue() []byte {
if m != nil {
return m.Value
}
return nil
}
func init() {
proto.RegisterType((*Entry)(nil), "proto.Entry")
}
func init() { proto.RegisterFile("entry.proto", fileDescriptor_entry_db5b99f271e6b4b6) }
var fileDescriptor_entry_db5b99f271e6b4b6 = []byte{
// 126 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,
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,
}

View File

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

View File

@@ -1,97 +0,0 @@
package streampb
import (
"bufio"
"encoding/binary"
"io"
"github.com/gogo/protobuf/proto"
"github.com/pkg/errors"
)
const (
// prefixSize is the number of bytes we preallocate for storing
// our big endian lenth prefix buffer.
prefixSize = 8
)
// NewEncoder creates a streaming protobuf encoder.
func NewEncoder(w io.Writer) *Encoder {
return &Encoder{w: bufio.NewWriter(w)}
}
// Encoder wraps an underlying io.Writer and allows you to stream
// proto encodings on it.
type Encoder struct {
w *bufio.Writer
}
// Encode takes any proto.Message and streams it to the underlying writer.
// Messages are framed with a length prefix.
func (e *Encoder) Encode(msg proto.Message) (int64, error) {
prefixBuf := make([]byte, prefixSize)
buf, err := proto.Marshal(msg)
if err != nil {
return 0, err
}
binary.BigEndian.PutUint64(prefixBuf, uint64(len(buf)))
if _, err := e.w.Write(prefixBuf); err != nil {
return 0, errors.Wrap(err, "failed writing length prefix")
}
n, err := e.w.Write(buf)
if err != nil {
return 0, errors.Wrap(err, "failed writing marshaled data")
}
if err = e.w.Flush(); err != nil {
return 0, errors.Wrap(err, "failed flushing data")
}
return int64(n + prefixSize), nil
}
// NewDecoder creates a streaming protobuf decoder.
func NewDecoder(r io.Reader) *Decoder {
return &Decoder{r: r}
}
// Decoder wraps an underlying io.Reader and allows you to stream
// proto decodings on it.
type Decoder struct {
r io.Reader
}
// Decode takes a proto.Message and unmarshals the next payload in the
// underlying io.Reader. It returns an EOF when it's done.
func (d *Decoder) Decode(v proto.Message) (int64, error) {
prefixBuf := make([]byte, prefixSize)
_, err := io.ReadFull(d.r, prefixBuf)
if err != nil {
return 0, err
}
n := binary.BigEndian.Uint64(prefixBuf)
buf := make([]byte, n)
idx := uint64(0)
for idx < n {
m, err := d.r.Read(buf[idx:n])
if err != nil {
return 0, errors.Wrap(translateError(err), "failed reading marshaled data")
}
idx += uint64(m)
}
return int64(idx + prefixSize), proto.Unmarshal(buf[:n], v)
}
func translateError(err error) error {
if err == io.EOF {
return io.ErrUnexpectedEOF
}
return err
}

View File

@@ -2,12 +2,32 @@ package internal
import (
"fmt"
"os"
"path/filepath"
"sort"
"strconv"
"strings"
)
func Exists(path string) bool {
_, err := os.Stat(path)
return err == nil
}
func DirSize(path string) (int64, error) {
var size int64
err := filepath.Walk(path, func(_ string, info os.FileInfo, err error) error {
if err != nil {
return err
}
if !info.IsDir() {
size += info.Size()
}
return err
})
return size, err
}
func GetDatafiles(path string) ([]string, error) {
fns, err := filepath.Glob(fmt.Sprintf("%s/*.data", path))
if err != nil {

View File

@@ -1,5 +1,11 @@
package bitcask
import (
"encoding/json"
"io/ioutil"
"path/filepath"
)
const (
// DefaultMaxDatafileSize is the default maximum datafile size in bytes
DefaultMaxDatafileSize = 1 << 20 // 1MB
@@ -18,6 +24,48 @@ type config struct {
maxDatafileSize int
maxKeySize int
maxValueSize int
sync bool
}
func (c *config) MarshalJSON() ([]byte, error) {
return json.Marshal(struct {
MaxDatafileSize int `json:"max_datafile_size"`
MaxKeySize int `json:"max_key_size"`
MaxValueSize int `json:"max_value_size"`
Sync bool `json:"sync"`
}{
MaxDatafileSize: c.maxDatafileSize,
MaxKeySize: c.maxKeySize,
MaxValueSize: c.maxValueSize,
Sync: c.sync,
})
}
func getConfig(path string) (*config, error) {
type Config struct {
MaxDatafileSize int `json:"max_datafile_size"`
MaxKeySize int `json:"max_key_size"`
MaxValueSize int `json:"max_value_size"`
Sync bool `json:"sync"`
}
var cfg Config
data, err := ioutil.ReadFile(filepath.Join(path, "config.json"))
if err != nil {
return nil, err
}
if err := json.Unmarshal(data, &cfg); err != nil {
return nil, err
}
return &config{
maxDatafileSize: cfg.MaxDatafileSize,
maxKeySize: cfg.MaxKeySize,
maxValueSize: cfg.MaxValueSize,
sync: cfg.Sync,
}, nil
}
func newDefaultConfig() *config {
@@ -51,3 +99,12 @@ func WithMaxValueSize(size int) Option {
return nil
}
}
// WithSync causes Sync() to be called on every key/value written increasing
// durability and safety at the expense of performance
func WithSync(sync bool) Option {
return func(cfg *config) error {
cfg.sync = sync
return nil
}
}