mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-08-17 09:23:48 +00:00
swarm/storage: apply changes from swarm-network-rewrite-syncer branch
Only changes from swarm/storage from swarm-network-rewrite-syncer branch are merged. Changes to other packages are not merged.
This commit is contained in:
Janos Guljas
Balint Gabor
parent
aa9a9ff818
commit
9129b6bf44
8 changed files with 643 additions and 406 deletions
|
|
@ -19,12 +19,15 @@ package storage
|
|||
import (
|
||||
"bytes"
|
||||
"crypto/rand"
|
||||
"encoding/binary"
|
||||
"fmt"
|
||||
"hash"
|
||||
"io"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/ethereum/go-ethereum/log"
|
||||
"github.com/ethereum/go-ethereum/crypto/sha3"
|
||||
)
|
||||
|
||||
type brokenLimitedReader struct {
|
||||
|
|
@ -42,16 +45,101 @@ func brokenLimitReader(data io.Reader, size int, errAt int) *brokenLimitedReader
|
|||
}
|
||||
}
|
||||
|
||||
func mputChunks(store ChunkStore, processors int, n int, chunksize int, hash hash.Hash) (hs []Key) {
|
||||
f := func(int) *Chunk {
|
||||
data := make([]byte, chunksize)
|
||||
rand.Reader.Read(data)
|
||||
hash.Reset()
|
||||
hash.Write(data)
|
||||
h := hash.Sum(nil)
|
||||
chunk := NewChunk(Key(h), nil)
|
||||
chunk.SData = data
|
||||
return chunk
|
||||
}
|
||||
return mput(store, processors, n, f)
|
||||
}
|
||||
|
||||
func mputRandomKey(store ChunkStore, processors int, n int, chunksize int) (hs []Key) {
|
||||
data := make([]byte, chunksize+8)
|
||||
binary.LittleEndian.PutUint64(data[0:8], uint64(chunksize))
|
||||
|
||||
f := func(int) *Chunk {
|
||||
h := make([]byte, 32)
|
||||
rand.Reader.Read(h)
|
||||
chunk := NewChunk(Key(h), nil)
|
||||
chunk.SData = data
|
||||
return chunk
|
||||
}
|
||||
return mput(store, processors, n, f)
|
||||
}
|
||||
|
||||
func mput(store ChunkStore, processors int, n int, f func(i int) *Chunk) (hs []Key) {
|
||||
wg := sync.WaitGroup{}
|
||||
wg.Add(processors)
|
||||
c := make(chan *Chunk)
|
||||
for i := 0; i < processors; i++ {
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
for chunk := range c {
|
||||
store.Put(chunk)
|
||||
}
|
||||
}()
|
||||
}
|
||||
for i := 0; i < n; i++ {
|
||||
chunk := f(i)
|
||||
hs = append(hs, chunk.Key)
|
||||
c <- chunk
|
||||
}
|
||||
close(c)
|
||||
wg.Wait()
|
||||
return hs
|
||||
}
|
||||
|
||||
func mget(store ChunkStore, hs []Key, f func(h Key, chunk *Chunk) error) error {
|
||||
wg := sync.WaitGroup{}
|
||||
wg.Add(len(hs))
|
||||
errc := make(chan error)
|
||||
|
||||
for _, k := range hs {
|
||||
go func(h Key) {
|
||||
defer wg.Done()
|
||||
chunk, err := store.Get(h)
|
||||
if err != nil {
|
||||
errc <- err
|
||||
return
|
||||
}
|
||||
if f != nil {
|
||||
err = f(h, chunk)
|
||||
if err != nil {
|
||||
errc <- err
|
||||
return
|
||||
}
|
||||
}
|
||||
}(k)
|
||||
}
|
||||
go func() {
|
||||
wg.Wait()
|
||||
close(errc)
|
||||
}()
|
||||
var err error
|
||||
select {
|
||||
case err = <-errc:
|
||||
case <-time.NewTimer(5 * time.Second).C:
|
||||
err = fmt.Errorf("timed out after 5 seconds")
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
func testDataReader(l int) (r io.Reader) {
|
||||
return io.LimitReader(rand.Reader, int64(l))
|
||||
}
|
||||
|
||||
func (self *brokenLimitedReader) Read(buf []byte) (int, error) {
|
||||
if self.off+len(buf) > self.errAt {
|
||||
func (r *brokenLimitedReader) Read(buf []byte) (int, error) {
|
||||
if r.off+len(buf) > r.errAt {
|
||||
return 0, fmt.Errorf("Broken reader")
|
||||
}
|
||||
self.off += len(buf)
|
||||
return self.lr.Read(buf)
|
||||
r.off += len(buf)
|
||||
return r.lr.Read(buf)
|
||||
}
|
||||
|
||||
func testDataReaderAndSlice(l int) (r io.Reader, slice []byte) {
|
||||
|
|
@ -63,54 +151,50 @@ func testDataReaderAndSlice(l int) (r io.Reader, slice []byte) {
|
|||
return
|
||||
}
|
||||
|
||||
func testStore(m ChunkStore, l int64, branches int64, t *testing.T) {
|
||||
func testStoreRandom(m ChunkStore, processors int, n int, chunksize int, t *testing.T) {
|
||||
hs := mputRandomKey(m, processors, n, chunksize)
|
||||
err := mget(m, hs, nil)
|
||||
if err != nil {
|
||||
t.Fatalf("testStore failed: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
chunkC := make(chan *Chunk)
|
||||
go func() {
|
||||
for chunk := range chunkC {
|
||||
m.Put(chunk)
|
||||
if chunk.wg != nil {
|
||||
chunk.wg.Done()
|
||||
func testStoreCorrect(m ChunkStore, processors int, n int, chunksize int, t *testing.T) {
|
||||
hs := mputChunks(m, processors, n, chunksize, sha3.NewKeccak256())
|
||||
f := func(h Key, chunk *Chunk) error {
|
||||
if !bytes.Equal(h, chunk.Key) {
|
||||
return fmt.Errorf("key does not match retrieved chunk Key")
|
||||
}
|
||||
hasher := sha3.NewKeccak256()
|
||||
hasher.Write(chunk.SData)
|
||||
exp := hasher.Sum(nil)
|
||||
if !bytes.Equal(h, exp) {
|
||||
return fmt.Errorf("key is not hash of chunk data")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
err := mget(m, hs, f)
|
||||
if err != nil {
|
||||
t.Fatalf("testStore failed: %v", err)
|
||||
}
|
||||
}
|
||||
}()
|
||||
chunker := NewTreeChunker(&ChunkerParams{
|
||||
Branches: branches,
|
||||
Hash: SHA3Hash,
|
||||
})
|
||||
swg := &sync.WaitGroup{}
|
||||
key, _ := chunker.Split(rand.Reader, l, chunkC, swg, nil)
|
||||
swg.Wait()
|
||||
close(chunkC)
|
||||
chunkC = make(chan *Chunk)
|
||||
|
||||
quit := make(chan bool)
|
||||
func benchmarkStorePut(store ChunkStore, processors int, n int, chunksize int, b *testing.B) {
|
||||
b.ReportAllocs()
|
||||
b.ResetTimer()
|
||||
for i := 0; i < b.N; i++ {
|
||||
mputRandomKey(store, processors, n, chunksize)
|
||||
}
|
||||
}
|
||||
|
||||
go func() {
|
||||
for ch := range chunkC {
|
||||
go func(chunk *Chunk) {
|
||||
storedChunk, err := m.Get(chunk.Key)
|
||||
if err == notFound {
|
||||
log.Trace(fmt.Sprintf("chunk '%v' not found", chunk.Key.Log()))
|
||||
} else if err != nil {
|
||||
log.Trace(fmt.Sprintf("error retrieving chunk %v: %v", chunk.Key.Log(), err))
|
||||
} else {
|
||||
chunk.SData = storedChunk.SData
|
||||
chunk.Size = storedChunk.Size
|
||||
func benchmarkStoreGet(store ChunkStore, processors int, n int, chunksize int, b *testing.B) {
|
||||
hs := mputRandomKey(store, processors, n, chunksize)
|
||||
b.ReportAllocs()
|
||||
b.ResetTimer()
|
||||
for i := 0; i < b.N; i++ {
|
||||
err := mget(store, hs, nil)
|
||||
if err != nil {
|
||||
b.Fatalf("mget failed: %v", err)
|
||||
}
|
||||
log.Trace(fmt.Sprintf("chunk '%v' not found", chunk.Key.Log()))
|
||||
close(chunk.C)
|
||||
}(ch)
|
||||
}
|
||||
close(quit)
|
||||
}()
|
||||
r := chunker.Join(key, chunkC)
|
||||
|
||||
b := make([]byte, l)
|
||||
n, err := r.ReadAt(b, 0)
|
||||
if err != io.EOF {
|
||||
t.Fatalf("read error (%v/%v) %v", n, l, err)
|
||||
}
|
||||
close(chunkC)
|
||||
<-quit
|
||||
}
|
||||
|
|
|
|||
|
|
@ -23,19 +23,15 @@
|
|||
package storage
|
||||
|
||||
import (
|
||||
"archive/tar"
|
||||
"bytes"
|
||||
"encoding/binary"
|
||||
"encoding/hex"
|
||||
"fmt"
|
||||
"io"
|
||||
"io/ioutil"
|
||||
"sync"
|
||||
|
||||
"github.com/ethereum/go-ethereum/log"
|
||||
"github.com/ethereum/go-ethereum/rlp"
|
||||
"github.com/syndtr/goleveldb/leveldb"
|
||||
"github.com/syndtr/goleveldb/leveldb/iterator"
|
||||
"github.com/syndtr/goleveldb/leveldb/opt"
|
||||
)
|
||||
|
||||
const (
|
||||
|
|
@ -51,10 +47,13 @@ const (
|
|||
)
|
||||
|
||||
var (
|
||||
keyOldData = byte(1)
|
||||
keyAccessCnt = []byte{2}
|
||||
keyEntryCnt = []byte{3}
|
||||
keyDataIdx = []byte{4}
|
||||
keyGCPos = []byte{5}
|
||||
keyData = byte(6)
|
||||
keyDistanceCnt = byte(7)
|
||||
)
|
||||
|
||||
type gcItem struct {
|
||||
|
|
@ -68,25 +67,29 @@ type DbStore struct {
|
|||
|
||||
// this should be stored in db, accessed transactionally
|
||||
entryCnt, accessCnt, dataIdx, capacity uint64
|
||||
bucketCnt []uint64
|
||||
|
||||
gcPos, gcStartPos []byte
|
||||
gcArray []*gcItem
|
||||
|
||||
hashfunc SwarmHasher
|
||||
|
||||
po func(Key) uint8
|
||||
lock sync.Mutex
|
||||
trusted bool // if hash integity check is to be performed (for testing only)
|
||||
}
|
||||
|
||||
func NewDbStore(path string, hash SwarmHasher, capacity uint64, radius int) (s *DbStore, err error) {
|
||||
// TODO: Instead of passing the distance function, just pass the address from which distances are calculated
|
||||
// to avoid the appearance of a pluggable distance metric and opportunities of bugs associated with providing
|
||||
// a function diferent from the one that is actually used.
|
||||
func NewDbStore(path string, hash SwarmHasher, capacity uint64, po func(Key) uint8) (s *DbStore, err error) {
|
||||
s = new(DbStore)
|
||||
|
||||
s.hashfunc = hash
|
||||
|
||||
s.db, err = NewLDBDatabase(path)
|
||||
if err != nil {
|
||||
return
|
||||
return nil, err
|
||||
}
|
||||
|
||||
s.po = po
|
||||
s.setCapacity(capacity)
|
||||
|
||||
s.gcStartPos = make([]byte, 1)
|
||||
|
|
@ -95,15 +98,31 @@ func NewDbStore(path string, hash SwarmHasher, capacity uint64, radius int) (s *
|
|||
|
||||
data, _ := s.db.Get(keyEntryCnt)
|
||||
s.entryCnt = BytesToU64(data)
|
||||
s.bucketCnt = make([]uint64, 0x100)
|
||||
for i := 0; i < 0x100; i++ {
|
||||
k := make([]byte, 2)
|
||||
k[0] = keyDistanceCnt
|
||||
k[1] = byte(uint8(i))
|
||||
cnt, _ := s.db.Get(k)
|
||||
s.bucketCnt[i] = BytesToU64(cnt)
|
||||
}
|
||||
data, _ = s.db.Get(keyAccessCnt)
|
||||
s.accessCnt = BytesToU64(data)
|
||||
//s.accessCnt = BytesToU64(data)
|
||||
if len(data) == 8 {
|
||||
s.accessCnt = binary.LittleEndian.Uint64(data)
|
||||
s.accessCnt++
|
||||
}
|
||||
data, _ = s.db.Get(keyDataIdx)
|
||||
if len(data) == 8 {
|
||||
s.dataIdx = BytesToU64(data)
|
||||
s.dataIdx++
|
||||
}
|
||||
|
||||
s.gcPos, _ = s.db.Get(keyGCPos)
|
||||
if s.gcPos == nil {
|
||||
s.gcPos = s.gcStartPos
|
||||
}
|
||||
return
|
||||
return s, nil
|
||||
}
|
||||
|
||||
type dpaDBIndex struct {
|
||||
|
|
@ -115,12 +134,14 @@ func BytesToU64(data []byte) uint64 {
|
|||
if len(data) < 8 {
|
||||
return 0
|
||||
}
|
||||
return binary.LittleEndian.Uint64(data)
|
||||
//return binary.LittleEndian.Uint64(data)
|
||||
return binary.BigEndian.Uint64(data)
|
||||
}
|
||||
|
||||
func U64ToBytes(val uint64) []byte {
|
||||
data := make([]byte, 8)
|
||||
binary.LittleEndian.PutUint64(data, val)
|
||||
//binary.LittleEndian.PutUint64(data, val)
|
||||
binary.BigEndian.PutUint64(data, val)
|
||||
return data
|
||||
}
|
||||
|
||||
|
|
@ -133,38 +154,52 @@ func (s *DbStore) updateIndexAccess(index *dpaDBIndex) {
|
|||
}
|
||||
|
||||
func getIndexKey(hash Key) []byte {
|
||||
HashSize := len(hash)
|
||||
key := make([]byte, HashSize+1)
|
||||
hashSize := len(hash)
|
||||
key := make([]byte, hashSize+1)
|
||||
key[0] = 0
|
||||
copy(key[1:], hash[:])
|
||||
return key
|
||||
}
|
||||
|
||||
func getDataKey(idx uint64) []byte {
|
||||
func getOldDataKey(idx uint64) []byte {
|
||||
key := make([]byte, 9)
|
||||
key[0] = 1
|
||||
key[0] = keyOldData
|
||||
binary.BigEndian.PutUint64(key[1:9], idx)
|
||||
|
||||
return key
|
||||
}
|
||||
|
||||
func getDataKey(idx uint64, po uint8) []byte {
|
||||
key := make([]byte, 10)
|
||||
key[0] = keyData
|
||||
key[1] = byte(po)
|
||||
binary.BigEndian.PutUint64(key[2:], idx)
|
||||
|
||||
return key
|
||||
}
|
||||
|
||||
func encodeIndex(index *dpaDBIndex) []byte {
|
||||
data, _ := rlp.EncodeToBytes(index)
|
||||
return data
|
||||
}
|
||||
|
||||
func encodeData(chunk *Chunk) []byte {
|
||||
return chunk.SData
|
||||
return append(chunk.Key[:], chunk.SData...)
|
||||
}
|
||||
|
||||
func decodeIndex(data []byte, index *dpaDBIndex) {
|
||||
func decodeIndex(data []byte, index *dpaDBIndex) error {
|
||||
dec := rlp.NewStream(bytes.NewReader(data), 0)
|
||||
dec.Decode(index)
|
||||
return dec.Decode(index)
|
||||
}
|
||||
|
||||
func decodeData(data []byte, chunk *Chunk) {
|
||||
chunk.SData = data[32:]
|
||||
chunk.Size = int64(binary.BigEndian.Uint64(data[32:40]))
|
||||
}
|
||||
|
||||
func decodeOldData(data []byte, chunk *Chunk) {
|
||||
chunk.SData = data
|
||||
chunk.Size = int64(binary.LittleEndian.Uint64(data[0:8]))
|
||||
chunk.Size = int64(binary.BigEndian.Uint64(data[0:8]))
|
||||
}
|
||||
|
||||
func gcListPartition(list []*gcItem, left int, right int, pivotIndex int) int {
|
||||
|
|
@ -250,98 +285,16 @@ func (s *DbStore) collectGarbage(ratio float32) {
|
|||
cutidx := gcListSelect(s.gcArray, 0, gcnt-1, int(float32(gcnt)*ratio))
|
||||
cutval := s.gcArray[cutidx].value
|
||||
|
||||
// fmt.Print(gcnt, " ", s.entryCnt, " ")
|
||||
|
||||
// actual gc
|
||||
for i := 0; i < gcnt; i++ {
|
||||
if s.gcArray[i].value <= cutval {
|
||||
s.delete(s.gcArray[i].idx, s.gcArray[i].idxKey)
|
||||
s.delete(s.gcArray[i].idx, s.gcArray[i].idxKey, s.po(Key(s.gcPos[1:])))
|
||||
}
|
||||
}
|
||||
|
||||
// fmt.Println(s.entryCnt)
|
||||
|
||||
s.db.Put(keyGCPos, s.gcPos)
|
||||
}
|
||||
|
||||
// Export writes all chunks from the store to a tar archive, returning the
|
||||
// number of chunks written.
|
||||
func (s *DbStore) Export(out io.Writer) (int64, error) {
|
||||
tw := tar.NewWriter(out)
|
||||
defer tw.Close()
|
||||
|
||||
it := s.db.NewIterator()
|
||||
defer it.Release()
|
||||
var count int64
|
||||
for ok := it.Seek([]byte{kpIndex}); ok; ok = it.Next() {
|
||||
key := it.Key()
|
||||
if (key == nil) || (key[0] != kpIndex) {
|
||||
break
|
||||
}
|
||||
|
||||
var index dpaDBIndex
|
||||
decodeIndex(it.Value(), &index)
|
||||
|
||||
data, err := s.db.Get(getDataKey(index.Idx))
|
||||
if err != nil {
|
||||
log.Warn(fmt.Sprintf("Chunk %x found but could not be accessed: %v", key[:], err))
|
||||
continue
|
||||
}
|
||||
|
||||
hdr := &tar.Header{
|
||||
Name: hex.EncodeToString(key[1:]),
|
||||
Mode: 0644,
|
||||
Size: int64(len(data)),
|
||||
}
|
||||
if err := tw.WriteHeader(hdr); err != nil {
|
||||
return count, err
|
||||
}
|
||||
if _, err := tw.Write(data); err != nil {
|
||||
return count, err
|
||||
}
|
||||
count++
|
||||
}
|
||||
|
||||
return count, nil
|
||||
}
|
||||
|
||||
// Import reads chunks into the store from a tar archive, returning the number
|
||||
// of chunks read.
|
||||
func (s *DbStore) Import(in io.Reader) (int64, error) {
|
||||
tr := tar.NewReader(in)
|
||||
|
||||
var count int64
|
||||
for {
|
||||
hdr, err := tr.Next()
|
||||
if err == io.EOF {
|
||||
break
|
||||
} else if err != nil {
|
||||
return count, err
|
||||
}
|
||||
|
||||
if len(hdr.Name) != 64 {
|
||||
log.Warn("ignoring non-chunk file", "name", hdr.Name)
|
||||
continue
|
||||
}
|
||||
|
||||
key, err := hex.DecodeString(hdr.Name)
|
||||
if err != nil {
|
||||
log.Warn("ignoring invalid chunk file", "name", hdr.Name, "err", err)
|
||||
continue
|
||||
}
|
||||
|
||||
data, err := ioutil.ReadAll(tr)
|
||||
if err != nil {
|
||||
return count, err
|
||||
}
|
||||
|
||||
s.Put(&Chunk{Key: key, SData: data})
|
||||
count++
|
||||
}
|
||||
|
||||
return count, nil
|
||||
}
|
||||
|
||||
func (s *DbStore) Cleanup() {
|
||||
//Iterates over the database and checks that there are no faulty chunks
|
||||
it := s.db.NewIterator()
|
||||
|
|
@ -356,21 +309,23 @@ func (s *DbStore) Cleanup() {
|
|||
}
|
||||
total++
|
||||
var index dpaDBIndex
|
||||
decodeIndex(it.Value(), &index)
|
||||
|
||||
data, err := s.db.Get(getDataKey(index.Idx))
|
||||
err := decodeIndex(it.Value(), &index)
|
||||
if err != nil {
|
||||
it.Next()
|
||||
continue
|
||||
}
|
||||
data, err := s.db.Get(getDataKey(index.Idx, s.po(Key(key[1:]))))
|
||||
if err != nil {
|
||||
log.Warn(fmt.Sprintf("Chunk %x found but could not be accessed: %v", key[:], err))
|
||||
s.delete(index.Idx, getIndexKey(key[1:]))
|
||||
s.delete(index.Idx, getIndexKey(key[1:]), s.po(Key(key[1:])))
|
||||
errorsFound++
|
||||
} else {
|
||||
hasher := s.hashfunc()
|
||||
hasher.Write(data)
|
||||
hasher.Write(data[32:])
|
||||
hash := hasher.Sum(nil)
|
||||
if !bytes.Equal(hash, key[1:]) {
|
||||
log.Warn(fmt.Sprintf("Found invalid chunk. Hash mismatch. hash=%x, key=%x", hash, key[:]))
|
||||
s.delete(index.Idx, getIndexKey(key[1:]))
|
||||
errorsFound++
|
||||
s.delete(index.Idx, getIndexKey(key[1:]), s.po(Key(key[1:])))
|
||||
}
|
||||
}
|
||||
it.Next()
|
||||
|
|
@ -379,16 +334,90 @@ func (s *DbStore) Cleanup() {
|
|||
log.Warn(fmt.Sprintf("Found %v errors out of %v entries", errorsFound, total))
|
||||
}
|
||||
|
||||
func (s *DbStore) delete(idx uint64, idxKey []byte) {
|
||||
func (s *DbStore) Dump() {
|
||||
//Iterates over the database and checks that there are no faulty chunks
|
||||
it := s.db.NewIterator()
|
||||
startPosition := []byte{kpIndex}
|
||||
it.Seek(startPosition)
|
||||
var key []byte
|
||||
var total int
|
||||
for it.Valid() {
|
||||
key = it.Key()
|
||||
if (key == nil) || (key[0] != kpIndex) {
|
||||
break
|
||||
}
|
||||
total++
|
||||
fmt.Printf("%x\n", key[1:])
|
||||
it.Next()
|
||||
}
|
||||
it.Release()
|
||||
log.Warn(fmt.Sprintf("logged %v chunks", total))
|
||||
}
|
||||
|
||||
func (s *DbStore) ReIndex() {
|
||||
//Iterates over the database and checks that there are no faulty chunks
|
||||
it := s.db.NewIterator()
|
||||
startPosition := []byte{keyOldData}
|
||||
it.Seek(startPosition)
|
||||
var key []byte
|
||||
var errorsFound, total int
|
||||
for it.Valid() {
|
||||
key = it.Key()
|
||||
if (key == nil) || (key[0] != keyOldData) {
|
||||
break
|
||||
}
|
||||
data := it.Value()
|
||||
hasher := s.hashfunc()
|
||||
hasher.Write(data)
|
||||
hash := hasher.Sum(nil)
|
||||
|
||||
newKey := make([]byte, 10)
|
||||
oldCntKey := make([]byte, 2)
|
||||
newCntKey := make([]byte, 2)
|
||||
oldCntKey[0] = keyDistanceCnt
|
||||
newCntKey[0] = keyDistanceCnt
|
||||
key[0] = keyData
|
||||
key[1] = byte(s.po(Key(key[1:])))
|
||||
oldCntKey[1] = key[1]
|
||||
newCntKey[1] = byte(s.po(Key(newKey[1:])))
|
||||
copy(newKey[2:], key[1:])
|
||||
newValue := append(hash, data...)
|
||||
|
||||
batch := new(leveldb.Batch)
|
||||
batch.Delete(key)
|
||||
s.bucketCnt[oldCntKey[1]]--
|
||||
batch.Put(oldCntKey, U64ToBytes(s.bucketCnt[oldCntKey[1]]))
|
||||
batch.Put(newKey, newValue)
|
||||
s.bucketCnt[newCntKey[1]]++
|
||||
batch.Put(newCntKey, U64ToBytes(s.bucketCnt[newCntKey[1]]))
|
||||
s.db.Write(batch)
|
||||
it.Next()
|
||||
}
|
||||
it.Release()
|
||||
log.Warn(fmt.Sprintf("Found %v errors out of %v entries", errorsFound, total))
|
||||
}
|
||||
|
||||
func (s *DbStore) delete(idx uint64, idxKey []byte, po uint8) {
|
||||
batch := new(leveldb.Batch)
|
||||
batch.Delete(idxKey)
|
||||
batch.Delete(getDataKey(idx))
|
||||
batch.Delete(getDataKey(idx, po))
|
||||
s.entryCnt--
|
||||
s.bucketCnt[po]--
|
||||
cntKey := make([]byte, 2)
|
||||
cntKey[0] = keyDistanceCnt
|
||||
cntKey[1] = po
|
||||
batch.Put(keyEntryCnt, U64ToBytes(s.entryCnt))
|
||||
batch.Put(cntKey, U64ToBytes(s.bucketCnt[po]))
|
||||
s.db.Write(batch)
|
||||
}
|
||||
|
||||
func (s *DbStore) Counter() uint64 {
|
||||
func (s *DbStore) Size() uint64 {
|
||||
s.lock.Lock()
|
||||
defer s.lock.Unlock()
|
||||
return s.entryCnt
|
||||
}
|
||||
|
||||
func (s *DbStore) CurrentStorageIndex() uint64 {
|
||||
s.lock.Lock()
|
||||
defer s.lock.Unlock()
|
||||
return s.dataIdx
|
||||
|
|
@ -410,7 +439,6 @@ func (s *DbStore) Put(chunk *Chunk) {
|
|||
}
|
||||
|
||||
data := encodeData(chunk)
|
||||
//data := ethutil.Encode([]interface{}{entry})
|
||||
|
||||
if s.entryCnt >= s.capacity {
|
||||
s.collectGarbage(gcArrayFreeRatio)
|
||||
|
|
@ -418,7 +446,9 @@ func (s *DbStore) Put(chunk *Chunk) {
|
|||
|
||||
batch := new(leveldb.Batch)
|
||||
|
||||
batch.Put(getDataKey(s.dataIdx), data)
|
||||
po := s.po(chunk.Key)
|
||||
t_datakey := getDataKey(s.dataIdx, po)
|
||||
batch.Put(t_datakey, data)
|
||||
|
||||
index.Idx = s.dataIdx
|
||||
s.updateIndexAccess(&index)
|
||||
|
|
@ -430,9 +460,17 @@ func (s *DbStore) Put(chunk *Chunk) {
|
|||
s.entryCnt++
|
||||
batch.Put(keyDataIdx, U64ToBytes(s.dataIdx))
|
||||
s.dataIdx++
|
||||
batch.Put(keyAccessCnt, U64ToBytes(s.accessCnt))
|
||||
accesscnt := make([]byte, 8)
|
||||
binary.LittleEndian.PutUint64(accesscnt, s.accessCnt)
|
||||
batch.Put(keyAccessCnt, accesscnt)
|
||||
s.accessCnt++
|
||||
|
||||
s.bucketCnt[po]++
|
||||
cntKey := make([]byte, 2)
|
||||
cntKey[0] = keyDistanceCnt
|
||||
cntKey[1] = po
|
||||
batch.Put(cntKey, U64ToBytes(s.bucketCnt[po]))
|
||||
|
||||
s.db.Write(batch)
|
||||
if chunk.dbStored != nil {
|
||||
close(chunk.dbStored)
|
||||
|
|
@ -450,7 +488,10 @@ func (s *DbStore) tryAccessIdx(ikey []byte, index *dpaDBIndex) bool {
|
|||
|
||||
batch := new(leveldb.Batch)
|
||||
|
||||
batch.Put(keyAccessCnt, U64ToBytes(s.accessCnt))
|
||||
accesscnt := make([]byte, 8)
|
||||
binary.LittleEndian.PutUint64(accesscnt, s.accessCnt)
|
||||
batch.Put(keyAccessCnt, accesscnt)
|
||||
|
||||
s.accessCnt++
|
||||
s.updateIndexAccess(index)
|
||||
idata = encodeIndex(index)
|
||||
|
|
@ -464,24 +505,34 @@ func (s *DbStore) tryAccessIdx(ikey []byte, index *dpaDBIndex) bool {
|
|||
func (s *DbStore) Get(key Key) (chunk *Chunk, err error) {
|
||||
s.lock.Lock()
|
||||
defer s.lock.Unlock()
|
||||
return s.get(key)
|
||||
}
|
||||
|
||||
var index dpaDBIndex
|
||||
func (s *DbStore) get(key Key) (chunk *Chunk, err error) {
|
||||
var indx dpaDBIndex
|
||||
|
||||
if s.tryAccessIdx(getIndexKey(key), &index) {
|
||||
if s.tryAccessIdx(getIndexKey(key), &indx) {
|
||||
var data []byte
|
||||
data, err = s.db.Get(getDataKey(index.Idx))
|
||||
|
||||
proximity := s.po(key)
|
||||
datakey := getDataKey(indx.Idx, proximity)
|
||||
data, err = s.db.Get(datakey)
|
||||
log.Trace(fmt.Sprintf("DBStore: Chunk %v indexkey %v datakey %x proximity %d", key.Log(), indx.Idx, datakey, proximity))
|
||||
if err != nil {
|
||||
log.Trace(fmt.Sprintf("DBStore: Chunk %v found but could not be accessed: %v", key.Log(), err))
|
||||
s.delete(index.Idx, getIndexKey(key))
|
||||
s.delete(indx.Idx, getIndexKey(key), s.po(key))
|
||||
return
|
||||
}
|
||||
|
||||
if s.hashfunc != nil {
|
||||
if !s.trusted {
|
||||
data_mod := data[32:]
|
||||
hasher := s.hashfunc()
|
||||
hasher.Write(data)
|
||||
hasher.Write(data_mod)
|
||||
hash := hasher.Sum(nil)
|
||||
|
||||
if !bytes.Equal(hash, key) {
|
||||
s.delete(index.Idx, getIndexKey(key))
|
||||
log.Trace(fmt.Sprintf("Apparent key/hash mismatch. Hash %x, key %v", hash, key[:]))
|
||||
s.delete(indx.Idx, getIndexKey(key), s.po(key))
|
||||
log.Warn("Invalid Chunk in Database. Please repair with command: 'swarm cleandb'")
|
||||
}
|
||||
}
|
||||
|
|
@ -516,7 +567,8 @@ func (s *DbStore) setCapacity(c uint64) {
|
|||
s.capacity = c
|
||||
|
||||
if s.entryCnt > c {
|
||||
ratio := float32(1.01) - float32(c)/float32(s.entryCnt)
|
||||
var ratio float32
|
||||
ratio = float32(1.01) - float32(c)/float32(s.entryCnt)
|
||||
if ratio < gcArrayFreeRatio {
|
||||
ratio = gcArrayFreeRatio
|
||||
}
|
||||
|
|
@ -533,62 +585,40 @@ func (s *DbStore) Close() {
|
|||
s.db.Close()
|
||||
}
|
||||
|
||||
// describes a section of the DbStore representing the unsynced
|
||||
// domain relevant to a peer
|
||||
// Start - Stop designate a continuous area Keys in an address space
|
||||
// typically the addresses closer to us than to the peer but not closer
|
||||
// another closer peer in between
|
||||
// From - To designates a time interval typically from the last disconnect
|
||||
// till the latest connection (real time traffic is relayed)
|
||||
type DbSyncState struct {
|
||||
Start, Stop Key
|
||||
First, Last uint64
|
||||
}
|
||||
|
||||
// implements the syncer iterator interface
|
||||
// iterates by storage index (~ time of storage = first entry to db)
|
||||
type dbSyncIterator struct {
|
||||
it iterator.Iterator
|
||||
DbSyncState
|
||||
}
|
||||
|
||||
// initialises a sync iterator from a syncToken (passed in with the handshake)
|
||||
func (self *DbStore) NewSyncIterator(state DbSyncState) (si *dbSyncIterator, err error) {
|
||||
if state.First > state.Last {
|
||||
return nil, fmt.Errorf("no entries found")
|
||||
}
|
||||
si = &dbSyncIterator{
|
||||
it: self.db.NewIterator(),
|
||||
DbSyncState: state,
|
||||
}
|
||||
si.it.Seek(getIndexKey(state.Start))
|
||||
return si, nil
|
||||
func (s *DbStore) SyncIterator(since uint64, until uint64, po uint8, f func(Key, uint64) bool) error {
|
||||
s.lock.Lock()
|
||||
defer s.lock.Unlock()
|
||||
untilkey := getDataKey(until, po)
|
||||
|
||||
it := s.db.NewIterator()
|
||||
seek := getDataKey(since, po)
|
||||
it.Seek(seek)
|
||||
defer it.Release()
|
||||
for it.Valid() {
|
||||
dbkey := it.Key()
|
||||
if dbkey[0] != keyData || dbkey[1] != byte(po) || bytes.Compare(untilkey, dbkey) < 0 {
|
||||
break
|
||||
}
|
||||
|
||||
// walk the area from Start to Stop and returns items within time interval
|
||||
// First to Last
|
||||
func (self *dbSyncIterator) Next() (key Key) {
|
||||
for self.it.Valid() {
|
||||
dbkey := self.it.Key()
|
||||
if dbkey[0] != 0 {
|
||||
key := make([]byte, 32)
|
||||
copy(key, it.Value()[:32])
|
||||
if !f(Key(key), binary.BigEndian.Uint64(dbkey[2:])) {
|
||||
break
|
||||
}
|
||||
key = Key(make([]byte, len(dbkey)-1))
|
||||
copy(key[:], dbkey[1:])
|
||||
if bytes.Compare(key[:], self.Start) <= 0 {
|
||||
self.it.Next()
|
||||
continue
|
||||
it.Next()
|
||||
}
|
||||
if bytes.Compare(key[:], self.Stop) > 0 {
|
||||
break
|
||||
}
|
||||
var index dpaDBIndex
|
||||
decodeIndex(self.it.Value(), &index)
|
||||
self.it.Next()
|
||||
if (index.Idx >= self.First) && (index.Idx < self.Last) {
|
||||
return
|
||||
}
|
||||
}
|
||||
self.it.Release()
|
||||
return nil
|
||||
}
|
||||
|
||||
func databaseExists(path string) bool {
|
||||
o := &opt.Options{
|
||||
ErrorIfMissing: true,
|
||||
}
|
||||
tdb, err := leveldb.OpenFile(path, o)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
defer tdb.Close()
|
||||
return true
|
||||
}
|
||||
|
|
|
|||
|
|
@ -18,174 +18,174 @@ package storage
|
|||
|
||||
import (
|
||||
"bytes"
|
||||
"fmt"
|
||||
"io/ioutil"
|
||||
"os"
|
||||
"testing"
|
||||
|
||||
"github.com/ethereum/go-ethereum/common"
|
||||
"github.com/ethereum/go-ethereum/log"
|
||||
)
|
||||
|
||||
func initDbStore(t *testing.T) *DbStore {
|
||||
type testDbStore struct {
|
||||
*DbStore
|
||||
dir string
|
||||
}
|
||||
|
||||
func newTestDbStore() (*testDbStore, error) {
|
||||
dir, err := ioutil.TempDir("", "bzz-storage-test")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
return nil, err
|
||||
}
|
||||
m, err := NewDbStore(dir, MakeHashFunc(SHA3Hash), defaultDbCapacity, defaultRadius)
|
||||
basekey := make([]byte, 32)
|
||||
db, err := NewDbStore(dir, MakeHashFunc(SHA3Hash), defaultDbCapacity, func(k Key) (ret uint8) { return uint8(Proximity(basekey[:], k[:])) })
|
||||
|
||||
return &testDbStore{db, dir}, err
|
||||
}
|
||||
|
||||
func (db *testDbStore) close() {
|
||||
db.Close()
|
||||
err := os.RemoveAll(db.dir)
|
||||
if err != nil {
|
||||
t.Fatal("can't create store:", err)
|
||||
panic(err)
|
||||
}
|
||||
return m
|
||||
}
|
||||
|
||||
func testDbStore(l int64, branches int64, t *testing.T) {
|
||||
m := initDbStore(t)
|
||||
defer m.Close()
|
||||
testStore(m, l, branches, t)
|
||||
func testDbStoreRandom(n int, processors int, chunksize int, t *testing.T) {
|
||||
db, err := newTestDbStore()
|
||||
if err != nil {
|
||||
t.Fatalf("init dbStore failed: %v", err)
|
||||
}
|
||||
defer db.close()
|
||||
db.trusted = true
|
||||
testStoreRandom(db, processors, n, chunksize, t)
|
||||
}
|
||||
|
||||
func TestDbStore128_0x1000000(t *testing.T) {
|
||||
testDbStore(0x1000000, 128, t)
|
||||
func testDbStoreCorrect(n int, processors int, chunksize int, t *testing.T) {
|
||||
db, err := newTestDbStore()
|
||||
if err != nil {
|
||||
t.Fatalf("init dbStore failed: %v", err)
|
||||
}
|
||||
defer db.close()
|
||||
testStoreCorrect(db, processors, n, chunksize, t)
|
||||
}
|
||||
|
||||
func TestDbStore128_10000_(t *testing.T) {
|
||||
testDbStore(10000, 128, t)
|
||||
func TestDbStoreRandom_1(t *testing.T) {
|
||||
testDbStoreRandom(1, 1, 0, t)
|
||||
}
|
||||
|
||||
func TestDbStore128_1000_(t *testing.T) {
|
||||
testDbStore(1000, 128, t)
|
||||
func TestDbStoreCorrect_1(t *testing.T) {
|
||||
testDbStoreCorrect(1, 1, 4096, t)
|
||||
}
|
||||
|
||||
func TestDbStore128_100_(t *testing.T) {
|
||||
testDbStore(100, 128, t)
|
||||
func TestDbStoreRandom_1_5k(t *testing.T) {
|
||||
testDbStoreRandom(8, 5000, 0, t)
|
||||
}
|
||||
|
||||
func TestDbStore2_100_(t *testing.T) {
|
||||
testDbStore(100, 2, t)
|
||||
func TestDbStoreRandom_8_5k(t *testing.T) {
|
||||
testDbStoreRandom(8, 5000, 0, t)
|
||||
}
|
||||
|
||||
func TestDbStoreCorrect_1_5k(t *testing.T) {
|
||||
testDbStoreCorrect(1, 5000, 4096, t)
|
||||
}
|
||||
|
||||
func TestDbStoreCorrect_8_5k(t *testing.T) {
|
||||
testDbStoreCorrect(8, 5000, 4096, t)
|
||||
}
|
||||
|
||||
func TestDbStoreNotFound(t *testing.T) {
|
||||
m := initDbStore(t)
|
||||
defer m.Close()
|
||||
_, err := m.Get(ZeroKey)
|
||||
db, err := newTestDbStore()
|
||||
if err != nil {
|
||||
t.Fatalf("init dbStore failed: %v", err)
|
||||
}
|
||||
defer db.close()
|
||||
|
||||
_, err = db.Get(ZeroKey)
|
||||
if err != notFound {
|
||||
t.Errorf("Expected notFound, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDbStoreSyncIterator(t *testing.T) {
|
||||
m := initDbStore(t)
|
||||
defer m.Close()
|
||||
keys := []Key{
|
||||
Key(common.Hex2Bytes("0000000000000000000000000000000000000000000000000000000000000000")),
|
||||
Key(common.Hex2Bytes("4000000000000000000000000000000000000000000000000000000000000000")),
|
||||
Key(common.Hex2Bytes("5000000000000000000000000000000000000000000000000000000000000000")),
|
||||
Key(common.Hex2Bytes("3000000000000000000000000000000000000000000000000000000000000000")),
|
||||
Key(common.Hex2Bytes("2000000000000000000000000000000000000000000000000000000000000000")),
|
||||
Key(common.Hex2Bytes("1000000000000000000000000000000000000000000000000000000000000000")),
|
||||
func TestIterator(t *testing.T) {
|
||||
var chunkcount int = 32
|
||||
var i int
|
||||
var poc uint
|
||||
chunkkeys := NewKeyCollection(chunkcount)
|
||||
chunkkeys_results := NewKeyCollection(chunkcount)
|
||||
chunks := make([]Chunk, chunkcount)
|
||||
|
||||
db, err := newTestDbStore()
|
||||
if err != nil {
|
||||
t.Fatalf("init dbStore failed: %v", err)
|
||||
}
|
||||
for _, key := range keys {
|
||||
m.Put(NewChunk(key, nil))
|
||||
defer db.close()
|
||||
|
||||
FakeChunk(getDefaultChunkSize(), chunkcount, chunks)
|
||||
|
||||
for i = 0; i < len(chunks); i++ {
|
||||
db.Put(&chunks[i])
|
||||
chunkkeys[i] = chunks[i].Key
|
||||
}
|
||||
it, err := m.NewSyncIterator(DbSyncState{
|
||||
Start: Key(common.Hex2Bytes("1000000000000000000000000000000000000000000000000000000000000000")),
|
||||
Stop: Key(common.Hex2Bytes("4000000000000000000000000000000000000000000000000000000000000000")),
|
||||
First: 2,
|
||||
Last: 4,
|
||||
|
||||
//testSplit(m, l, 128, chunkkeys, t)
|
||||
|
||||
for i = 0; i < len(chunkkeys); i++ {
|
||||
log.Trace(fmt.Sprintf("Chunk array pos %d/%d: '%v'", i, chunkcount, chunkkeys[i]))
|
||||
}
|
||||
|
||||
i = 0
|
||||
for poc = 0; poc <= 255; poc++ {
|
||||
err := db.SyncIterator(0, uint64(chunkkeys.Len()), uint8(poc), func(k Key, n uint64) bool {
|
||||
log.Trace(fmt.Sprintf("Got key %v number %d poc %d", k, n, uint8(poc)))
|
||||
chunkkeys_results[n] = k
|
||||
i++
|
||||
return true
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error creating NewSyncIterator")
|
||||
t.Fatalf("Iterator call failed: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
var chunk Key
|
||||
var res []Key
|
||||
for {
|
||||
chunk = it.Next()
|
||||
if chunk == nil {
|
||||
break
|
||||
for i = 0; i < chunkcount; i++ {
|
||||
if bytes.Compare(chunkkeys[i], chunkkeys_results[i]) != 0 {
|
||||
t.Fatalf("Chunk put #%d key '%v' does not match iterator's key '%v'", i, chunkkeys[i], chunkkeys_results[i])
|
||||
}
|
||||
res = append(res, chunk)
|
||||
}
|
||||
if len(res) != 1 {
|
||||
t.Fatalf("Expected 1 chunk, got %v: %v", len(res), res)
|
||||
}
|
||||
if !bytes.Equal(res[0][:], keys[3]) {
|
||||
t.Fatalf("Expected %v chunk, got %v", keys[3], res[0])
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
func benchmarkDbStorePut(n int, processors int, chunksize int, b *testing.B) {
|
||||
db, err := newTestDbStore()
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error creating NewSyncIterator")
|
||||
}
|
||||
|
||||
it, err = m.NewSyncIterator(DbSyncState{
|
||||
Start: Key(common.Hex2Bytes("1000000000000000000000000000000000000000000000000000000000000000")),
|
||||
Stop: Key(common.Hex2Bytes("5000000000000000000000000000000000000000000000000000000000000000")),
|
||||
First: 2,
|
||||
Last: 4,
|
||||
})
|
||||
|
||||
res = nil
|
||||
for {
|
||||
chunk = it.Next()
|
||||
if chunk == nil {
|
||||
break
|
||||
}
|
||||
res = append(res, chunk)
|
||||
}
|
||||
if len(res) != 2 {
|
||||
t.Fatalf("Expected 2 chunk, got %v: %v", len(res), res)
|
||||
}
|
||||
if !bytes.Equal(res[0][:], keys[3]) {
|
||||
t.Fatalf("Expected %v chunk, got %v", keys[3], res[0])
|
||||
}
|
||||
if !bytes.Equal(res[1][:], keys[2]) {
|
||||
t.Fatalf("Expected %v chunk, got %v", keys[2], res[1])
|
||||
b.Fatalf("init dbStore failed: %v", err)
|
||||
}
|
||||
defer db.close()
|
||||
db.trusted = true
|
||||
benchmarkStorePut(db, processors, n, chunksize, b)
|
||||
}
|
||||
|
||||
func benchmarkDbStoreGet(n int, processors int, chunksize int, b *testing.B) {
|
||||
db, err := newTestDbStore()
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error creating NewSyncIterator")
|
||||
b.Fatalf("init dbStore failed: %v", err)
|
||||
}
|
||||
defer db.close()
|
||||
db.trusted = true
|
||||
benchmarkStoreGet(db, processors, n, chunksize, b)
|
||||
}
|
||||
|
||||
it, _ = m.NewSyncIterator(DbSyncState{
|
||||
Start: Key(common.Hex2Bytes("1000000000000000000000000000000000000000000000000000000000000000")),
|
||||
Stop: Key(common.Hex2Bytes("4000000000000000000000000000000000000000000000000000000000000000")),
|
||||
First: 2,
|
||||
Last: 5,
|
||||
})
|
||||
res = nil
|
||||
for {
|
||||
chunk = it.Next()
|
||||
if chunk == nil {
|
||||
break
|
||||
}
|
||||
res = append(res, chunk)
|
||||
}
|
||||
if len(res) != 2 {
|
||||
t.Fatalf("Expected 2 chunk, got %v", len(res))
|
||||
}
|
||||
if !bytes.Equal(res[0][:], keys[4]) {
|
||||
t.Fatalf("Expected %v chunk, got %v", keys[4], res[0])
|
||||
}
|
||||
if !bytes.Equal(res[1][:], keys[3]) {
|
||||
t.Fatalf("Expected %v chunk, got %v", keys[3], res[1])
|
||||
func BenchmarkDbStorePut_1_5k(b *testing.B) {
|
||||
benchmarkDbStorePut(5000, 1, 4096, b)
|
||||
}
|
||||
|
||||
it, _ = m.NewSyncIterator(DbSyncState{
|
||||
Start: Key(common.Hex2Bytes("2000000000000000000000000000000000000000000000000000000000000000")),
|
||||
Stop: Key(common.Hex2Bytes("4000000000000000000000000000000000000000000000000000000000000000")),
|
||||
First: 2,
|
||||
Last: 5,
|
||||
})
|
||||
res = nil
|
||||
for {
|
||||
chunk = it.Next()
|
||||
if chunk == nil {
|
||||
break
|
||||
func BenchmarkDbStorePut_8_5k(b *testing.B) {
|
||||
benchmarkDbStorePut(5000, 8, 4096, b)
|
||||
}
|
||||
res = append(res, chunk)
|
||||
}
|
||||
if len(res) != 1 {
|
||||
t.Fatalf("Expected 1 chunk, got %v", len(res))
|
||||
}
|
||||
if !bytes.Equal(res[0][:], keys[3]) {
|
||||
t.Fatalf("Expected %v chunk, got %v", keys[3], res[0])
|
||||
|
||||
func BenchmarkDbStoreGet_1_5k(b *testing.B) {
|
||||
benchmarkDbStoreGet(5000, 1, 4096, b)
|
||||
}
|
||||
|
||||
func BenchmarkDbStoreGet_8_5k(b *testing.B) {
|
||||
benchmarkDbStoreGet(5000, 8, 4096, b)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -59,15 +59,16 @@ type DPA struct {
|
|||
|
||||
lock sync.Mutex
|
||||
running bool
|
||||
wg *sync.WaitGroup
|
||||
quitC chan bool
|
||||
}
|
||||
|
||||
// for testing locally
|
||||
func NewLocalDPA(datadir string) (*DPA, error) {
|
||||
func NewLocalDPA(datadir string, basekey []byte) (*DPA, error) {
|
||||
|
||||
hash := MakeHashFunc("SHA3")
|
||||
|
||||
dbStore, err := NewDbStore(datadir, hash, singletonSwarmDbCapacity, 0)
|
||||
dbStore, err := NewDbStore(datadir, hash, singletonSwarmDbCapacity, func(k Key) (ret uint8) { return uint8(Proximity(basekey[:], k[:])) })
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
@ -135,11 +136,8 @@ func (self *DPA) retrieveLoop() {
|
|||
|
||||
func (self *DPA) retrieveWorker() {
|
||||
for chunk := range self.retrieveC {
|
||||
log.Trace(fmt.Sprintf("dpa: retrieve loop : chunk %v", chunk.Key.Log()))
|
||||
storedChunk, err := self.Get(chunk.Key)
|
||||
if err == notFound {
|
||||
log.Trace(fmt.Sprintf("chunk %v not found", chunk.Key.Log()))
|
||||
} else if err != nil {
|
||||
if err != nil {
|
||||
log.Trace(fmt.Sprintf("error retrieving chunk %v: %v", chunk.Key.Log(), err))
|
||||
} else {
|
||||
chunk.SData = storedChunk.SData
|
||||
|
|
@ -169,9 +167,7 @@ func (self *DPA) storeWorker() {
|
|||
for chunk := range self.storeC {
|
||||
self.Put(chunk)
|
||||
if chunk.wg != nil {
|
||||
log.Trace(fmt.Sprintf("dpa: store processor %v", chunk.Key.Log()))
|
||||
chunk.wg.Done()
|
||||
|
||||
}
|
||||
select {
|
||||
case <-self.quitC:
|
||||
|
|
@ -200,7 +196,6 @@ func NewDpaChunkStore(localStore, netStore ChunkStore) *dpaChunkStore {
|
|||
// waits for response or times out
|
||||
func (self *dpaChunkStore) Get(key Key) (chunk *Chunk, err error) {
|
||||
chunk, err = self.netStore.Get(key)
|
||||
// timeout := time.Now().Add(searchTimeout)
|
||||
if chunk.SData != nil {
|
||||
log.Trace(fmt.Sprintf("DPA.Get: %v found locally, %d bytes", key.Log(), len(chunk.SData)))
|
||||
return
|
||||
|
|
@ -238,4 +233,5 @@ func (self *dpaChunkStore) Put(entry *Chunk) {
|
|||
}
|
||||
|
||||
// Close chunk store
|
||||
func (self *dpaChunkStore) Close() {}
|
||||
func (self *dpaChunkStore) Close() {
|
||||
}
|
||||
|
|
|
|||
|
|
@ -28,12 +28,17 @@ import (
|
|||
const testDataSize = 0x1000000
|
||||
|
||||
func TestDPArandom(t *testing.T) {
|
||||
dbStore := initDbStore(t)
|
||||
dbStore.setCapacity(50000)
|
||||
memStore := NewMemStore(dbStore, defaultCacheCapacity)
|
||||
tdb, err := newTestDbStore()
|
||||
if err != nil {
|
||||
t.Fatalf("init dbStore failed: %v", err)
|
||||
}
|
||||
defer tdb.close()
|
||||
db := tdb.DbStore
|
||||
db.setCapacity(50000)
|
||||
memStore := NewMemStore(db, defaultCacheCapacity)
|
||||
localStore := &LocalStore{
|
||||
memStore,
|
||||
dbStore,
|
||||
db,
|
||||
}
|
||||
chunker := NewTreeChunker(NewChunkerParams())
|
||||
dpa := &DPA{
|
||||
|
|
@ -65,7 +70,7 @@ func TestDPArandom(t *testing.T) {
|
|||
}
|
||||
ioutil.WriteFile("/tmp/slice.bzz.16M", slice, 0666)
|
||||
ioutil.WriteFile("/tmp/result.bzz.16M", resultSlice, 0666)
|
||||
localStore.memStore = NewMemStore(dbStore, defaultCacheCapacity)
|
||||
localStore.memStore = NewMemStore(db, defaultCacheCapacity)
|
||||
resultReader = dpa.Retrieve(key)
|
||||
for i := range resultSlice {
|
||||
resultSlice[i] = 0
|
||||
|
|
@ -83,13 +88,17 @@ func TestDPArandom(t *testing.T) {
|
|||
}
|
||||
|
||||
func TestDPA_capacity(t *testing.T) {
|
||||
dbStore := initDbStore(t)
|
||||
memStore := NewMemStore(dbStore, defaultCacheCapacity)
|
||||
tdb, err := newTestDbStore()
|
||||
if err != nil {
|
||||
t.Fatalf("init dbStore failed: %v", err)
|
||||
}
|
||||
defer tdb.close()
|
||||
db := tdb.DbStore
|
||||
memStore := NewMemStore(db, 0)
|
||||
localStore := &LocalStore{
|
||||
memStore,
|
||||
dbStore,
|
||||
db,
|
||||
}
|
||||
memStore.setCapacity(0)
|
||||
chunker := NewTreeChunker(NewChunkerParams())
|
||||
dpa := &DPA{
|
||||
Chunker: chunker,
|
||||
|
|
|
|||
|
|
@ -28,8 +28,8 @@ type LocalStore struct {
|
|||
}
|
||||
|
||||
// This constructor uses MemStore and DbStore as components
|
||||
func NewLocalStore(hash SwarmHasher, params *StoreParams) (*LocalStore, error) {
|
||||
dbStore, err := NewDbStore(params.ChunkDbPath, hash, params.DbCapacity, params.Radius)
|
||||
func NewLocalStore(hash SwarmHasher, params *StoreParams, basekey []byte) (*LocalStore, error) {
|
||||
dbStore, err := NewDbStore(params.ChunkDbPath, hash, params.DbCapacity, func(k Key) (ret uint8) { return uint8(Proximity(basekey[:], k[:])) })
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
|
|||
|
|
@ -16,35 +16,82 @@
|
|||
|
||||
package storage
|
||||
|
||||
import (
|
||||
"testing"
|
||||
)
|
||||
import "testing"
|
||||
|
||||
func testMemStore(l int64, branches int64, t *testing.T) {
|
||||
m := NewMemStore(nil, defaultCacheCapacity)
|
||||
testStore(m, l, branches, t)
|
||||
func newTestMemStore() *MemStore {
|
||||
return NewMemStore(nil, defaultCacheCapacity)
|
||||
}
|
||||
|
||||
func TestMemStore128_10000(t *testing.T) {
|
||||
testMemStore(10000, 128, t)
|
||||
func testMemStoreRandom(n int, processors int, chunksize int, t *testing.T) {
|
||||
m := newTestMemStore()
|
||||
defer m.Close()
|
||||
testStoreRandom(m, processors, n, chunksize, t)
|
||||
}
|
||||
|
||||
func TestMemStore128_1000(t *testing.T) {
|
||||
testMemStore(1000, 128, t)
|
||||
func testMemStoreCorrect(n int, processors int, chunksize int, t *testing.T) {
|
||||
m := newTestMemStore()
|
||||
defer m.Close()
|
||||
testStoreCorrect(m, processors, n, chunksize, t)
|
||||
}
|
||||
|
||||
func TestMemStore128_100(t *testing.T) {
|
||||
testMemStore(100, 128, t)
|
||||
func TestMemStoreRandom_1(t *testing.T) {
|
||||
testMemStoreRandom(1, 1, 0, t)
|
||||
}
|
||||
|
||||
func TestMemStore2_100(t *testing.T) {
|
||||
testMemStore(100, 2, t)
|
||||
func TestMemStoreCorrect_1(t *testing.T) {
|
||||
testMemStoreCorrect(1, 1, 4104, t)
|
||||
}
|
||||
|
||||
func TestMemStoreRandom_1_10k(t *testing.T) {
|
||||
testMemStoreRandom(1, 5000, 0, t)
|
||||
}
|
||||
|
||||
func TestMemStoreCorrect_1_10k(t *testing.T) {
|
||||
testMemStoreCorrect(1, 5000, 4096, t)
|
||||
}
|
||||
|
||||
func TestMemStoreRandom_8_10k(t *testing.T) {
|
||||
testMemStoreRandom(8, 5000, 0, t)
|
||||
}
|
||||
|
||||
func TestMemStoreCorrect_8_10k(t *testing.T) {
|
||||
testMemStoreCorrect(8, 5000, 4096, t)
|
||||
}
|
||||
|
||||
func TestMemStoreNotFound(t *testing.T) {
|
||||
m := NewMemStore(nil, defaultCacheCapacity)
|
||||
m := newTestMemStore()
|
||||
defer m.Close()
|
||||
|
||||
_, err := m.Get(ZeroKey)
|
||||
if err != notFound {
|
||||
t.Errorf("Expected notFound, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func benchmarkMemStorePut(n int, processors int, chunksize int, b *testing.B) {
|
||||
m := newTestMemStore()
|
||||
defer m.Close()
|
||||
benchmarkStorePut(m, processors, n, chunksize, b)
|
||||
}
|
||||
|
||||
func benchmarkMemStoreGet(n int, processors int, chunksize int, b *testing.B) {
|
||||
m := newTestMemStore()
|
||||
defer m.Close()
|
||||
benchmarkStoreGet(m, processors, n, chunksize, b)
|
||||
}
|
||||
|
||||
func BenchmarkMemStorePut_1_5k(b *testing.B) {
|
||||
benchmarkMemStorePut(5000, 1, 4096, b)
|
||||
}
|
||||
|
||||
func BenchmarkMemStorePut_8_5k(b *testing.B) {
|
||||
benchmarkMemStorePut(5000, 8, 4096, b)
|
||||
}
|
||||
|
||||
func BenchmarkMemStoreGet_1_5k(b *testing.B) {
|
||||
benchmarkMemStoreGet(5000, 1, 4096, b)
|
||||
}
|
||||
|
||||
func BenchmarkMemStoreGet_8_5k(b *testing.B) {
|
||||
benchmarkMemStoreGet(5000, 8, 4096, b)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -19,6 +19,8 @@ package storage
|
|||
import (
|
||||
"bytes"
|
||||
"crypto"
|
||||
"crypto/rand"
|
||||
"encoding/binary"
|
||||
"fmt"
|
||||
"hash"
|
||||
"io"
|
||||
|
|
@ -29,6 +31,8 @@ import (
|
|||
"github.com/ethereum/go-ethereum/crypto/sha3"
|
||||
)
|
||||
|
||||
const MaxPO = 7
|
||||
|
||||
type Hasher func() hash.Hash
|
||||
type SwarmHasher func() SwarmHash
|
||||
|
||||
|
|
@ -73,6 +77,26 @@ func (h Key) bits(i, j uint) uint {
|
|||
return res
|
||||
}
|
||||
|
||||
func Proximity(one, other []byte) (ret int) {
|
||||
b := (MaxPO-1)/8 + 1
|
||||
if b > len(one) {
|
||||
b = len(one)
|
||||
}
|
||||
m := 8
|
||||
for i := 0; i < b; i++ {
|
||||
oxo := one[i] ^ other[i]
|
||||
if i == b-1 {
|
||||
m = MaxPO % 8
|
||||
}
|
||||
for j := 0; j < m; j++ {
|
||||
if (uint8(oxo)>>uint8(7-j))&0x01 != 0 {
|
||||
return i*8 + j
|
||||
}
|
||||
}
|
||||
}
|
||||
return MaxPO
|
||||
}
|
||||
|
||||
func IsZeroKey(key Key) bool {
|
||||
return len(key) == 0 || bytes.Equal(key, ZeroKey)
|
||||
}
|
||||
|
|
@ -100,10 +124,10 @@ func (key Key) Hex() string {
|
|||
}
|
||||
|
||||
func (key Key) Log() string {
|
||||
if len(key[:]) < 4 {
|
||||
if len(key[:]) < 8 {
|
||||
return fmt.Sprintf("%x", []byte(key[:]))
|
||||
}
|
||||
return fmt.Sprintf("%08x", []byte(key[:4]))
|
||||
return fmt.Sprintf("%016x", []byte(key[:8]))
|
||||
}
|
||||
|
||||
func (key Key) String() string {
|
||||
|
|
@ -122,6 +146,27 @@ func (key *Key) UnmarshalJSON(value []byte) error {
|
|||
return nil
|
||||
}
|
||||
|
||||
type KeyCollection []Key
|
||||
|
||||
func NewKeyCollection(l int) KeyCollection {
|
||||
return make(KeyCollection, l)
|
||||
}
|
||||
|
||||
func (c KeyCollection) Len() int {
|
||||
return len(c)
|
||||
}
|
||||
|
||||
func (c KeyCollection) Less(i, j int) bool {
|
||||
if bytes.Compare(c[i], c[j]) == -1 {
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func (c KeyCollection) Swap(i, j int) {
|
||||
c[i], c[j] = c[j], c[i]
|
||||
}
|
||||
|
||||
// each chunk when first requested opens a record associated with the request
|
||||
// next time a request for the same chunk arrives, this record is updated
|
||||
// this request status keeps track of the request ID-s as well as the requesting
|
||||
|
|
@ -163,6 +208,32 @@ func NewChunk(key Key, rs *RequestStatus) *Chunk {
|
|||
return &Chunk{Key: key, Req: rs}
|
||||
}
|
||||
|
||||
func FakeChunk(size int64, count int, chunks []Chunk) int {
|
||||
var i int
|
||||
hasher := MakeHashFunc(SHA3Hash)()
|
||||
chunksize := getDefaultChunkSize()
|
||||
if size > chunksize {
|
||||
size = chunksize
|
||||
}
|
||||
|
||||
for i = 0; i < count; i++ {
|
||||
hasher.Reset()
|
||||
chunks[i].SData = make([]byte, size)
|
||||
rand.Read(chunks[i].SData)
|
||||
binary.LittleEndian.PutUint64(chunks[i].SData[:8], uint64(size))
|
||||
hasher.Write(chunks[i].SData)
|
||||
chunks[i].Key = make([]byte, 32)
|
||||
copy(chunks[i].Key, hasher.Sum(nil))
|
||||
}
|
||||
|
||||
return i
|
||||
}
|
||||
|
||||
func getDefaultChunkSize() int64 {
|
||||
return DefaultBranches * int64(MakeHashFunc(SHA3Hash)().Size())
|
||||
|
||||
}
|
||||
|
||||
/*
|
||||
The ChunkStore interface is implemented by :
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue