nxshock/zkv
1
0
Fork 0
mirror of https://github.com/nxshock/zkv.git synced 2024-11-27 11:21:02 +05:00
Code Issues Projects Releases Wiki Activity

Speedup small writes by using write buffer

Browse Source
This commit is contained in:
nxshock 2022-12-09 20:05:30 +05:00
parent e166e07daa
commit 8dfc73af1d
4 changed files with 22 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
defaults.go
View File

@ -9,5 +9,6 @@ import (
var defaultOptions = Options{ var defaultOptions = Options{
MaxParallelReads: runtime.NumCPU(), MaxParallelReads: runtime.NumCPU(),
CompressionLevel: zstd.SpeedDefault, CompressionLevel: zstd.SpeedDefault,
BufferSize: 4 * 1024 * 1024, MemoryBufferSize: 4 * 1024 * 1024,
DiskBufferSize: 1 * 1024 * 1024,
} }

7
options.go
View File

@ -9,8 +9,11 @@ type Options struct {
// Compression level // Compression level
CompressionLevel zstd.EncoderLevel CompressionLevel zstd.EncoderLevel
// Write buffer size in bytes // Memory write buffer size in bytes
BufferSize int MemoryBufferSize int
// Diwk write buffer size in bytes
DiskBufferSize int
} }
func (o *Options) setDefaults() { func (o *Options) setDefaults() {

17
zkv.go
View File

@ -1,6 +1,7 @@
package zkv package zkv
import ( import (
"bufio"
"bytes" "bytes"
"crypto/sha256" "crypto/sha256"
"encoding/base64" "encoding/base64"
@ -124,7 +125,7 @@ func (s *Store) Delete(key interface{}) error {
return err return err
} }
if s.buffer.Len() > s.options.BufferSize { if s.buffer.Len() > s.options.MemoryBufferSize {
err = s.flush() err = s.flush()
if err != nil { if err != nil {
@ -209,7 +210,7 @@ func (s *Store) setBytes(keyHash [sha256.Size224]byte, valueBytes []byte) error
return err return err
} }
if s.buffer.Len() > s.options.BufferSize { if s.buffer.Len() > s.options.MemoryBufferSize {
err = s.flush() err = s.flush()
if err != nil { if err != nil {
@ -238,7 +239,7 @@ func (s *Store) set(key, value interface{}) error {
return err return err
} }
if s.buffer.Len() > s.options.BufferSize { if s.buffer.Len() > s.options.MemoryBufferSize {
err = s.flush() err = s.flush()
if err != nil { if err != nil {
@ -377,7 +378,9 @@ func (s *Store) flush() error {
return fmt.Errorf("open store file: %v", err) return fmt.Errorf("open store file: %v", err)
} }
encoder, err := zstd.NewWriter(f, zstd.WithEncoderLevel(s.options.CompressionLevel)) diskWriteBuffer := bufio.NewWriterSize(f, s.options.DiskBufferSize)
encoder, err := zstd.NewWriter(diskWriteBuffer, zstd.WithEncoderLevel(s.options.CompressionLevel))
if err != nil { if err != nil {
f.Close() f.Close()
return fmt.Errorf("open store file: %v", err) return fmt.Errorf("open store file: %v", err)
@ -402,6 +405,12 @@ func (s *Store) flush() error {
return err return err
} }
err = diskWriteBuffer.Flush()
if err != nil {
// TODO: truncate file to previous state
return err
}
err = f.Close() err = f.Close()
if err != nil { if err != nil {
return err return err

4
zkv_test.go
View File

@ -165,7 +165,7 @@ func TestBufferBasic(t *testing.T) {
const filePath = "TestBuffer.zkv" const filePath = "TestBuffer.zkv"
defer os.Remove(filePath) defer os.Remove(filePath)
db, err := OpenWithOptions(filePath, Options{BufferSize: 100}) db, err := OpenWithOptions(filePath, Options{MemoryBufferSize: 100})
assert.NoError(t, err) assert.NoError(t, err)
err = db.Set(1, make([]byte, 100)) err = db.Set(1, make([]byte, 100))
@ -190,7 +190,7 @@ func TestBufferRead(t *testing.T) {
const recordCount = 100 const recordCount = 100
defer os.Remove(filePath) defer os.Remove(filePath)
db, err := OpenWithOptions(filePath, Options{BufferSize: 100}) db, err := OpenWithOptions(filePath, Options{MemoryBufferSize: 100})
assert.NoError(t, err) assert.NoError(t, err)
for i := 1; i <= recordCount; i++ { for i := 1; i <= recordCount; i++ {