From 9129b6bf44f21b2acbf8c6b7c3f2e1d33c7d7257 Mon Sep 17 00:00:00 2001 From: Janos Guljas Date: Tue, 2 Jan 2018 18:53:41 +0100 Subject: [PATCH] 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. --- swarm/storage/common_test.go | 192 +++++++++++----- swarm/storage/dbstore.go | 396 ++++++++++++++++++--------------- swarm/storage/dbstore_test.go | 262 +++++++++++----------- swarm/storage/dpa.go | 16 +- swarm/storage/dpa_test.go | 27 ++- swarm/storage/localstore.go | 4 +- swarm/storage/memstore_test.go | 77 +++++-- swarm/storage/types.go | 75 ++++++- 8 files changed, 643 insertions(+), 406 deletions(-) diff --git a/swarm/storage/common_test.go b/swarm/storage/common_test.go index cd4c2ef139..6fd66a03d9 100644 --- a/swarm/storage/common_test.go +++ b/swarm/storage/common_test.go @@ -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) { - - chunkC := make(chan *Chunk) - go func() { - for chunk := range chunkC { - m.Put(chunk) - if chunk.wg != nil { - chunk.wg.Done() - } - } - }() - 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) - - 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 - } - 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) +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) + } +} + +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) + } +} + +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) + } +} + +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) + } } - close(chunkC) - <-quit } diff --git a/swarm/storage/dbstore.go b/swarm/storage/dbstore.go index 454319f229..c1bef29823 100644 --- a/swarm/storage/dbstore.go +++ b/swarm/storage/dbstore.go @@ -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 ( - keyAccessCnt = []byte{2} - keyEntryCnt = []byte{3} - keyDataIdx = []byte{4} - keyGCPos = []byte{5} + 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 - - lock sync.Mutex + 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) - s.dataIdx = BytesToU64(data) + 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) -// 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 { + 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 } - key = Key(make([]byte, len(dbkey)-1)) - copy(key[:], dbkey[1:]) - if bytes.Compare(key[:], self.Start) <= 0 { - self.it.Next() - continue - } - if bytes.Compare(key[:], self.Stop) > 0 { + + key := make([]byte, 32) + copy(key, it.Value()[:32]) + if !f(Key(key), binary.BigEndian.Uint64(dbkey[2:])) { break } - var index dpaDBIndex - decodeIndex(self.it.Value(), &index) - self.it.Next() - if (index.Idx >= self.First) && (index.Idx < self.Last) { - return - } + it.Next() } - 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 +} diff --git a/swarm/storage/dbstore_test.go b/swarm/storage/dbstore_test.go index dd165b5768..27bab975a6 100644 --- a/swarm/storage/dbstore_test.go +++ b/swarm/storage/dbstore_test.go @@ -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")), - } - for _, key := range keys { - m.Put(NewChunk(key, nil)) - } - it, err := m.NewSyncIterator(DbSyncState{ - Start: Key(common.Hex2Bytes("1000000000000000000000000000000000000000000000000000000000000000")), - Stop: Key(common.Hex2Bytes("4000000000000000000000000000000000000000000000000000000000000000")), - First: 2, - Last: 4, - }) +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("unexpected error creating NewSyncIterator") + t.Fatalf("init dbStore failed: %v", err) + } + defer db.close() + + FakeChunk(getDefaultChunkSize(), chunkcount, chunks) + + for i = 0; i < len(chunks); i++ { + db.Put(&chunks[i]) + chunkkeys[i] = chunks[i].Key } - var chunk Key - var res []Key - for { - chunk = it.Next() - if chunk == nil { - break + //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("Iterator call failed: %v", err) } - 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]) } - 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 + 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) != 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]) } - if err != nil { - t.Fatalf("unexpected error creating NewSyncIterator") - } - - 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]) - } - - 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 - } - 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 benchmarkDbStorePut(n int, processors int, chunksize int, b *testing.B) { + db, err := newTestDbStore() + if err != nil { + 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 { + b.Fatalf("init dbStore failed: %v", err) + } + defer db.close() + db.trusted = true + benchmarkStoreGet(db, processors, n, chunksize, b) +} + +func BenchmarkDbStorePut_1_5k(b *testing.B) { + benchmarkDbStorePut(5000, 1, 4096, b) +} + +func BenchmarkDbStorePut_8_5k(b *testing.B) { + benchmarkDbStorePut(5000, 8, 4096, b) +} + +func BenchmarkDbStoreGet_1_5k(b *testing.B) { + benchmarkDbStoreGet(5000, 1, 4096, b) +} + +func BenchmarkDbStoreGet_8_5k(b *testing.B) { + benchmarkDbStoreGet(5000, 8, 4096, b) } diff --git a/swarm/storage/dpa.go b/swarm/storage/dpa.go index 49a362555e..b8f7f5fd8f 100644 --- a/swarm/storage/dpa.go +++ b/swarm/storage/dpa.go @@ -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() { +} diff --git a/swarm/storage/dpa_test.go b/swarm/storage/dpa_test.go index a23b9efebe..3bccd82d54 100644 --- a/swarm/storage/dpa_test.go +++ b/swarm/storage/dpa_test.go @@ -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, diff --git a/swarm/storage/localstore.go b/swarm/storage/localstore.go index bf9eeb2e77..2ed9fb305a 100644 --- a/swarm/storage/localstore.go +++ b/swarm/storage/localstore.go @@ -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 } diff --git a/swarm/storage/memstore_test.go b/swarm/storage/memstore_test.go index 2e0ab535af..6b4bc0da56 100644 --- a/swarm/storage/memstore_test.go +++ b/swarm/storage/memstore_test.go @@ -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) +} diff --git a/swarm/storage/types.go b/swarm/storage/types.go index d35f1f9294..e2c111f7b1 100644 --- a/swarm/storage/types.go +++ b/swarm/storage/types.go @@ -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 :