mtcontrib/mtsatellite
1
0
Fork 0
mirror of https://bitbucket.org/s_l_teichmann/mtsatellite synced 2025-01-11 09:20:17 +01:00
Code Issues Projects Releases Wiki Activity

Free memory of write batches in LevelDB backend. Use an extra r/w mutex.

Browse Source
This commit is contained in:
Sascha L. Teichmann 2014-08-07 15:16:12 +02:00
parent 954dddf795
commit c09d74ae0f
1 changed files with 48 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

66
leveldb.go
View File

@ -6,6 +6,7 @@ package main
import ( import (
"log" "log"
"sync"
leveldb "github.com/jmhodges/levigo" leveldb "github.com/jmhodges/levigo"
) )
@ -13,6 +14,7 @@ import (
type LevelDBBackend struct { type LevelDBBackend struct {
cache *leveldb.Cache cache *leveldb.Cache
db *leveldb.DB db *leveldb.DB
mutex sync.RWMutex
} }
type LevelDBSession struct { type LevelDBSession struct {
@ -36,6 +38,18 @@ func NewLeveDBBackend(path string, cacheSize int) (ldb *LevelDBBackend, err erro
return return
} }
func (ldb *LevelDBBackend) doRead(f func(db *leveldb.DB)) {
ldb.mutex.RLock()
f(ldb.db)
ldb.mutex.RUnlock()
}
func (ldb *LevelDBBackend) doWrite(f func(db *leveldb.DB)) {
ldb.mutex.Lock()
f(ldb.db)
ldb.mutex.Unlock()
}
func (ldb *LevelDBBackend) NewSession() (Session, error) { func (ldb *LevelDBBackend) NewSession() (Session, error) {
return &LevelDBSession{ldb, nil}, nil return &LevelDBSession{ldb, nil}, nil
} }
@ -54,9 +68,19 @@ func (ldb *LevelDBBackend) Shutdown() error {
} }
func (ldbs *LevelDBSession) Fetch(hash, key []byte) (value []byte, err error) { func (ldbs *LevelDBSession) Fetch(hash, key []byte) (value []byte, err error) {
ro := leveldb.NewReadOptions()
value, err = ldbs.backend.db.Get(ro, key) var pos int64
ro.Close() if pos, err = bytes2pos(key); err != nil {
return
}
// Re-code it to make LevelDB happy.
key = pos2bytes(pos)
ldbs.backend.doRead(func(db *leveldb.DB) {
ro := leveldb.NewReadOptions()
value, err = ldbs.backend.db.Get(ro, key)
ro.Close()
})
return return
} }
@ -64,11 +88,11 @@ func (ldbs *LevelDBSession) InTransaction() bool {
return ldbs.tx != nil return ldbs.tx != nil
} }
func (ldbs *LevelDBSession) keyExists(key []byte) (exists bool, err error) { func keyExists(db *leveldb.DB, key []byte) (exists bool, err error) {
ro := leveldb.NewReadOptions() ro := leveldb.NewReadOptions()
defer ro.Close() defer ro.Close()
var data []byte var data []byte
if data, err = ldbs.backend.db.Get(ro, key); err != nil { if data, err = db.Get(ro, key); err != nil {
return return
} }
exists = data != nil exists = data != nil
@ -84,18 +108,20 @@ func (ldbs *LevelDBSession) Store(hash, key, value []byte) (exists bool, err err
// Re-code it to make LevelDB happy. // Re-code it to make LevelDB happy.
key = pos2bytes(pos) key = pos2bytes(pos)
if exists, err = ldbs.keyExists(key); err != nil { ldbs.backend.doWrite(func(db *leveldb.DB) {
return if exists, err = keyExists(db, key); err != nil {
} return
}
if ldbs.tx != nil { if ldbs.tx != nil {
ldbs.tx.Put(key, value) ldbs.tx.Put(key, value)
return return
} }
wo := leveldb.NewWriteOptions() wo := leveldb.NewWriteOptions()
err = ldbs.backend.db.Put(wo, key, value) err = ldbs.backend.db.Put(wo, key, value)
wo.Close() wo.Close()
})
return return
} }
@ -112,8 +138,12 @@ func (ldbs *LevelDBSession) CommitTransaction() (err error) {
return return
} }
ldbs.tx = nil ldbs.tx = nil
wo := leveldb.NewWriteOptions() ldbs.backend.doWrite(func(db *leveldb.DB) {
err = ldbs.backend.db.Write(wo, tx) wo := leveldb.NewWriteOptions()
wo.Close() wo.SetSync(true)
err = db.Write(wo, tx)
wo.Close()
tx.Close()
})
return return
} }