From 73cc31563810de16e416cfdc4c22ad23e12395a9 Mon Sep 17 00:00:00 2001 From: zelig Date: Thu, 27 Apr 2017 02:12:03 +0200 Subject: [PATCH] swarm/storage, swarm/network: fix dbstore and syncer issues * increment dataindex read from db; fixes invalid hash error * increment last index; fixes repeated keys in history sync * only use MaxPO (=8) proxbins in dbstore; simplifies iterations * reduce logging output and make it useful * fix tests for fuse and http API * simplify chunkstore tests in common_test, streamline tests for dbstore, memstore and add benchmarks --- swarm/fuse/swarmfs_test.go | 20 ++- swarm/network/syncer.go | 35 ++-- swarm/storage/chunker_test.go | 32 ++-- swarm/storage/common_test.go | 211 +++++++++++++++--------- swarm/storage/database.go | 1 - swarm/storage/dbstore.go | 40 ++--- swarm/storage/dbstore_test.go | 293 ++++++++++++--------------------- swarm/storage/dpa_test.go | 27 ++- swarm/storage/memstore_test.go | 87 ++++++---- swarm/storage/types.go | 22 ++- swarm/storage/types_test.go | 17 -- swarm/testutil/http.go | 2 +- 12 files changed, 402 insertions(+), 385 deletions(-) delete mode 100644 swarm/storage/types_test.go diff --git a/swarm/fuse/swarmfs_test.go b/swarm/fuse/swarmfs_test.go index 69f3cc6157..7b51fb17a4 100644 --- a/swarm/fuse/swarmfs_test.go +++ b/swarm/fuse/swarmfs_test.go @@ -38,6 +38,24 @@ type fileInfo struct { contents []byte } +func testFuseFileSystem(t *testing.T, f func(*api.Api)) { + + datadir, err := ioutil.TempDir("", "fuse") + if err != nil { + t.Fatalf("unable to create temp dir: %v", err) + } + os.RemoveAll(datadir) + + dpa, err := storage.NewLocalDPA(datadir, storage.ZeroKey) + if err != nil { + return + } + api := api.NewApi(dpa, nil) + dpa.Start() + f(api) + dpa.Stop() +} + func createTestFilesAndUploadToSwarm(t *testing.T, api *api.Api, files map[string]fileInfo, uploadDir string) string { os.RemoveAll(uploadDir) @@ -810,7 +828,7 @@ func TestFUSE(t *testing.T) { } os.RemoveAll(datadir) - dpa, err := storage.NewLocalDPA(datadir) + dpa, err := storage.NewLocalDPA(datadir, storage.ZeroKey) if err != nil { t.Fatal(err) } diff --git a/swarm/network/syncer.go b/swarm/network/syncer.go index 0d0ec7306f..b48247241f 100644 --- a/swarm/network/syncer.go +++ b/swarm/network/syncer.go @@ -17,6 +17,7 @@ package network import ( + "bytes" "encoding/binary" "encoding/json" "fmt" @@ -285,8 +286,7 @@ func (self *syncer) newSyncRequest(req interface{}, p int) (*syncRequest, error) // * read is on demand, blocking unless history channel is read // * closes the channel once iteration finishes func (self *syncer) syncHistory(state *syncState) chan interface{} { - var roundCnt, stateCnt, totalCnt uint - var quit, wait bool + var roundCnt, roundsCnt, totalCnt uint history := make(chan interface{}, historyBufferSize) go func() { @@ -294,19 +294,27 @@ func (self *syncer) syncHistory(state *syncState) chan interface{} { defer close(history) last := state.Last since := state.Since + var prevKey storage.Key for { - log.Debug(fmt.Sprintf("syncer[%v]: syncing history since %v for chunks of proximity order %v", self.key.Log(), since, state.PO)) + var quit, wait bool + roundsCnt++ + log.Debug(fmt.Sprintf("syncer[%v]: syncing history range %v-%v PO%03d, round %v, total: %v", self.key.Log(), since, state.SessionAt, state.PO, roundsCnt, totalCnt)) err := self.dbAccess.iterator(since, state.SessionAt, state.PO, func(key storage.Key, idx uint64) bool { select { // if history channel cannot be written to, we fall through to default // and release the iterator // last is not set to idx so the lost key will be retrieved in the next - // batch given Since is set to last + // batch given Since is set to last + 1 case history <- key: roundCnt++ - stateCnt++ totalCnt++ last = idx + log.Trace(fmt.Sprintf("key: %v, round %v, cnt: %v, total: %v", key, roundsCnt, roundCnt, totalCnt)) + if bytes.Equal(key[:], prevKey[:]) { + log.Trace(fmt.Sprintf("repeating key: %v, round %v, cnt: %v, total: %v", key, roundsCnt, roundCnt, totalCnt)) + panic("repeating key") + } + prevKey = key return true case <-self.quit: quit = true @@ -315,19 +323,18 @@ func (self *syncer) syncHistory(state *syncState) chan interface{} { wait = true return false } - // return true //dummy return. }) if err != nil { - log.Debug(fmt.Sprintf("syncer[%v]: sync for %v failed: %v: ..abort syncing", self.key.Log(), state, err)) + log.Debug(fmt.Sprintf("syncer[%v]: sync for %v failed: %v: ..abort syncing, round %v, cnt: %v, total: %v", self.key.Log(), state, err, roundsCnt, roundCnt, totalCnt)) return } if quit { return } // advancing syncstate - since = last + since = last + 1 if !wait { - log.Debug(fmt.Sprintf("syncer[%v]: sync for %v failed: %v: ..abort syncing", self.key.Log(), state, err)) + log.Debug(fmt.Sprintf("syncer[%v]: nothing more to sync, round %v, cnt: %v, total: %v", self.key.Log(), roundsCnt, roundCnt, totalCnt)) break } // if history channel is no longer contented, continue outer loop @@ -339,8 +346,8 @@ func (self *syncer) syncHistory(state *syncState) chan interface{} { case <-self.quit: } } + roundCnt = 0 } - roundCnt = 0 }() return history } @@ -373,7 +380,7 @@ func (self *syncer) syncUnsyncedKeys() { state := self.state if state.IncludeCloser { - state.PO = 255 + state.PO = storage.MaxPO } history := self.syncHistory(self.state) @@ -396,10 +403,10 @@ LOOP: keys = self.keys[priority] break PRIORITIES } - log.Trace(fmt.Sprintf("syncer[%v/%v]: queue: [%v, %v, %v]", self.key.Log(), priority, len(self.keys[High]), len(self.keys[Medium]), len(self.keys[Low]))) + // log.Trace(fmt.Sprintf("syncer[%v/%v]: queue: [%v, %v, %v]", self.key.Log(), priority, len(self.keys[High]), len(self.keys[Medium]), len(self.keys[Low]))) // if the input queue is empty on this level, resort to history if there is any if uint(priority) == histPrior && history != nil { - log.Trace(fmt.Sprintf("syncer[%v]: reading history for %v", self.key.Log(), self.key)) + // log.Trace(fmt.Sprintf("syncer[%v]: reading history for %v", self.key.Log(), self.key)) keys = history break PRIORITIES } @@ -554,7 +561,7 @@ func (self *syncer) syncDeliveries() { log.Warn(fmt.Sprintf("syncer[%v]: failed to deliver %v: %v", self.key.Log(), req, err)) } else { success++ - log.Trace(fmt.Sprintf("syncer[%v]: %v successfully delivered", self.key.Log(), req)) + // log.Trace(fmt.Sprintf("syncer[%v]: %v successfully delivered", self.key.Log(), req)) } } if total%self.SyncBatchSize == 0 { diff --git a/swarm/storage/chunker_test.go b/swarm/storage/chunker_test.go index 40f870246f..e78e7298d2 100644 --- a/swarm/storage/chunker_test.go +++ b/swarm/storage/chunker_test.go @@ -22,7 +22,6 @@ import ( "encoding/binary" "fmt" "io" - "runtime" "sync" "testing" "time" @@ -232,28 +231,25 @@ func benchReadAll(reader LazySectionReader) { } func benchmarkJoin(n int, t *testing.B) { + chunker := NewTreeChunker(NewChunkerParams()) + tester := &chunkerTester{t: t} + data := testDataReader(n) + + chunkC := make(chan *Chunk, 1000) + swg := &sync.WaitGroup{} + + key := tester.Split(chunker, data, int64(n), chunkC, swg, nil) + t.ReportAllocs() + t.ResetTimer() for i := 0; i < t.N; i++ { - chunker := NewTreeChunker(NewChunkerParams()) - tester := &chunkerTester{t: t} - data := testDataReader(n) - - chunkC := make(chan *Chunk, 1000) - swg := &sync.WaitGroup{} - - key := tester.Split(chunker, data, int64(n), chunkC, swg, nil) - // t.StartTimer() chunkC = make(chan *Chunk, 1000) quitC := make(chan bool) reader := tester.Join(chunker, key, i, chunkC, quitC) benchReadAll(reader) close(chunkC) <-quitC - // t.StopTimer() } - stats := new(runtime.MemStats) - runtime.ReadMemStats(stats) - fmt.Println(stats.Sys) } func benchmarkSplitTree(n int, t *testing.B) { @@ -264,9 +260,6 @@ func benchmarkSplitTree(n int, t *testing.B) { data := testDataReader(n) tester.Split(chunker, data, int64(n), nil, nil, nil) } - stats := new(runtime.MemStats) - runtime.ReadMemStats(stats) - fmt.Println(stats.Sys) } func benchmarkSplitPyramid(n int, t *testing.B) { @@ -277,9 +270,6 @@ func benchmarkSplitPyramid(n int, t *testing.B) { data := testDataReader(n) tester.Split(splitter, data, int64(n), nil, nil, nil) } - stats := new(runtime.MemStats) - runtime.ReadMemStats(stats) - fmt.Println(stats.Sys) } func BenchmarkJoin_2(t *testing.B) { benchmarkJoin(100, t) } @@ -311,5 +301,3 @@ func BenchmarkSplitPyramid_5(t *testing.B) { benchmarkSplitPyramid(100000, t) } func BenchmarkSplitPyramid_6(t *testing.B) { benchmarkSplitPyramid(1000000, t) } func BenchmarkSplitPyramid_7(t *testing.B) { benchmarkSplitPyramid(10000000, t) } func BenchmarkSplitPyramid_8(t *testing.B) { benchmarkSplitPyramid(100000000, t) } - -// godep go test -bench ./swarm/storage -cpuprofile cpu.out -memprofile mem.out diff --git a/swarm/storage/common_test.go b/swarm/storage/common_test.go index a2910d7d6e..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,79 +151,50 @@ func testDataReaderAndSlice(l int) (r io.Reader, slice []byte) { return } -func testStore(m ChunkStore, indata io.Reader, 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: defaultHash, - }) - swg := &sync.WaitGroup{} - key, _ := chunker.Split(indata, 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) } - close(chunkC) - <-quit } -// only put, but fills an array supplied by the caller with the keys -func testSplit(m ChunkStore, l int64, branches int64, chunkkeys []Key, t *testing.T) Key { - var i int - chunkC := make(chan *Chunk) - go func() { - for chunk := range chunkC { - chunkkeys[i] = chunk.Key - i++ - 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") } - }() - chunker := NewTreeChunker(&ChunkerParams{ - Branches: branches, - Hash: defaultHash, - }) - swg := &sync.WaitGroup{} - key, _ := chunker.Split(rand.Reader, l, chunkC, swg, nil) - swg.Wait() - close(chunkC) - return 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) + } + } } diff --git a/swarm/storage/database.go b/swarm/storage/database.go index 7d3b57265e..2532490cc9 100644 --- a/swarm/storage/database.go +++ b/swarm/storage/database.go @@ -29,7 +29,6 @@ import ( ) const openFileLimit = 128 -//const openFileLimit = -1 type LDBDatabase struct { db *leveldb.DB diff --git a/swarm/storage/dbstore.go b/swarm/storage/dbstore.go index eaef7aea16..ded6006b07 100644 --- a/swarm/storage/dbstore.go +++ b/swarm/storage/dbstore.go @@ -25,7 +25,6 @@ package storage import ( "bytes" "encoding/binary" - "encoding/hex" "fmt" "sync" @@ -77,6 +76,8 @@ type DbStore struct { hashfunc Hasher po func(Key) uint8 lock sync.Mutex + + trusted bool // if hash integity check is to be performed (for testing only) } // TODO: Instead of passing the distance function, just pass the address from which distances are calculated @@ -115,9 +116,14 @@ func NewDbStore(path string, hash Hasher, capacity uint64, po func(Key) uint8) ( //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 @@ -449,7 +455,6 @@ func (s *DbStore) Put(chunk *Chunk) { po := s.po(chunk.Key) t_datakey := getDataKey(s.dataIdx, po) batch.Put(t_datakey, data) - log.Trace(fmt.Sprintf("batch put: datai dx %v prox %v chunkkey %v datakey %v data %v", s.dataIdx, s.po(chunk.Key), hex.EncodeToString(chunk.Key), t_datakey, hex.EncodeToString(data[0:64]))) index.Idx = s.dataIdx s.updateIndexAccess(&index) @@ -518,23 +523,24 @@ func (s *DbStore) get(key Key) (chunk *Chunk, err error) { proximity := s.po(key) datakey := getDataKey(indx.Idx, proximity) data, err = s.db.Get(datakey) - log.Trace(fmt.Sprintf("DBStore: Chunk %v indexkey %x datakey %x proximity %d", key.Log(), indx.Idx, datakey, proximity)) + 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(indx.Idx, getIndexKey(key), s.po(key)) return } - // - data_mod := data[32:] + if !s.trusted { + data_mod := data[32:] + hasher := s.hashfunc() + hasher.Write(data_mod) + hash := hasher.Sum(nil) - hasher := s.hashfunc() - hasher.Write(data_mod) - hash := hasher.Sum(nil) - - if !bytes.Equal(hash, key) { - s.delete(index.Idx, getIndexKey(key)) - log.Warn("Invalid Chunk in Database. Please repair with command: 'swarm cleandb'") + if !bytes.Equal(hash, 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'") + } } chunk = &Chunk{ @@ -581,10 +587,6 @@ func (s *DbStore) setCapacity(c uint64) { } } -func (s *DbStore) getEntryCnt() uint64 { - return s.entryCnt -} - func (s *DbStore) Close() { s.db.Close() } @@ -593,11 +595,11 @@ func (s *DbStore) Close() { 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() - it.Seek(getDataKey(since, po)) + seek := getDataKey(since, po) + it.Seek(seek) defer it.Release() for it.Valid() { dbkey := it.Key() diff --git a/swarm/storage/dbstore_test.go b/swarm/storage/dbstore_test.go index 3678391ca8..641347d26d 100644 --- a/swarm/storage/dbstore_test.go +++ b/swarm/storage/dbstore_test.go @@ -18,216 +18,94 @@ package storage import ( "bytes" - "crypto/rand" "fmt" - "io" "io/ioutil" + "os" "testing" - "github.com/ethereum/go-ethereum/crypto/sha3" "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 } - basekey := sha3.NewKeccak256().Sum([]byte("random")) - m, err := NewDbStore(dir, MakeHashFunc(defaultHash), defaultDbCapacity, func(k Key) (ret uint8) { return uint8(proximity(basekey[:], k[:])) }) + basekey := make([]byte, 32) + db, err := NewDbStore(dir, MakeHashFunc(defaultHash), 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(indata io.Reader, l int64, branches int64, t *testing.T) { - t.Skip() - if indata == nil { - indata = rand.Reader +func testDbStoreRandom(n int, processors int, chunksize int, t *testing.T) { + db, err := newTestDbStore() + if err != nil { + t.Fatalf("init dbStore failed: %v", err) } - m := initDbStore(t) - defer m.Close() - testStore(m, indata, l, branches, t) + defer db.close() + db.trusted = true + testStoreRandom(db, processors, n, chunksize, t) } -func TestDbStore128_0x1000000(t *testing.T) { - testDbStore(nil, 0x1000000, 128, t) -} - -func TestDbStore128_10000_(t *testing.T) { - testDbStore(nil, 10000, 128, t) -} - -func TestDbStore128_1000_(t *testing.T) { - testDbStore(nil, 1000, 128, t) -} - -func TestDbStore128_100_(t *testing.T) { - testDbStore(nil, 100, 128, t) -} - -func TestDbStore2_100_(t *testing.T) { - testDbStore(nil, 100, 2, t) -} - -func TestDbStore128_1000000_fixed_(t *testing.T) { - b := []byte{} - br := getFixedData(b, 1000000, 254) - testDbStore(br, 1000000, 2, t) -} - -func TestDbStore2_100_fixed_(t *testing.T) { - b := []byte{} - br := getFixedData(b, 100, 0) - testDbStore(br, 100, 2, t) -} - -func getFixedData(b []byte, l uint32, p uint8) io.Reader { - var i byte // it will wrap and still fit byte but not be of much use >255 cos its will only generate more of the same chunks - var c uint32 - if p == 0 { - p = 255 +func testDbStoreCorrect(n int, processors int, chunksize int, t *testing.T) { + db, err := newTestDbStore() + if err != nil { + t.Fatalf("init dbStore failed: %v", err) } - for c = 0; c < l; c++ { - b = append(b, byte(i)) - if i == p { - i = 0 - } else { - i++ - } - } - return bytes.NewReader(b) + defer db.close() + testStoreCorrect(db, processors, n, chunksize, t) +} + +func TestDbStoreRandom_1(t *testing.T) { + testDbStoreRandom(1, 1, 0, t) +} + +func TestDbStoreCorrect_1(t *testing.T) { + testDbStoreCorrect(1, 1, 4096, t) +} + +func TestDbStoreRandom_1_5k(t *testing.T) { + testDbStoreRandom(8, 5000, 0, 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, -// }) -// if err != nil { -// t.Fatalf("unexpected error creating NewSyncIterator") -// } - -// var chunk Key -// var res []Key -// for { -// chunk = it.Next() -// if chunk == nil { -// break -// } -// 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 -// } -// 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 = brokenLimitReader(data, size, errAt) -// 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 TestIterator(t *testing.T) { var chunkcount int = 32 var i int @@ -236,13 +114,16 @@ func TestIterator(t *testing.T) { chunkkeys_results := NewKeyCollection(chunkcount) chunks := make([]Chunk, chunkcount) - m := initDbStore(t) - defer m.Close() + db, err := newTestDbStore() + if err != nil { + t.Fatalf("init dbStore failed: %v", err) + } + defer db.close() FakeChunk(getDefaultChunkSize(), chunkcount, chunks) for i = 0; i < len(chunks); i++ { - m.Put(&chunks[i]) + db.Put(&chunks[i]) chunkkeys[i] = chunks[i].Key } @@ -254,7 +135,7 @@ func TestIterator(t *testing.T) { i = 0 for poc = 0; poc <= 255; poc++ { - err := m.SyncIterator(0, uint64(chunkkeys.Len()), uint8(poc), func(k Key, n uint64) bool { + 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++ @@ -272,3 +153,39 @@ func TestIterator(t *testing.T) { } } + +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_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/memstore_test.go b/swarm/storage/memstore_test.go index b06e68cac1..6b4bc0da56 100644 --- a/swarm/storage/memstore_test.go +++ b/swarm/storage/memstore_test.go @@ -16,51 +16,82 @@ package storage -import ( - "bytes" - "crypto/rand" - "io" - "testing" -) +import "testing" -func testMemStore(indata io.Reader, l int64, branches int64, t *testing.T) { - if indata == nil { - indata = rand.Reader - } - m := NewMemStore(nil, defaultCacheCapacity) - testStore(m, indata, l, branches, t) +func newTestMemStore() *MemStore { + return NewMemStore(nil, defaultCacheCapacity) } -func TestMemStore128_10000(t *testing.T) { - testMemStore(nil, 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(nil, 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(nil, 100, 128, t) +func TestMemStoreRandom_1(t *testing.T) { + testMemStoreRandom(1, 1, 0, t) } -func TestMemStore2_100(t *testing.T) { - testMemStore(nil, 100, 2, t) +func TestMemStoreCorrect_1(t *testing.T) { + testMemStoreCorrect(1, 1, 4104, t) } -func TestMemStore2_100_fixed_(t *testing.T) { - b := []byte{} - for i := 0; i < 100; i++ { - b = append(b, byte(i)) - } +func TestMemStoreRandom_1_10k(t *testing.T) { + testMemStoreRandom(1, 5000, 0, t) +} - br := bytes.NewReader(b) - testMemStore(br, 100, 2, 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 f8993f4b83..18f8f595fa 100644 --- a/swarm/storage/types.go +++ b/swarm/storage/types.go @@ -30,6 +30,8 @@ import ( "github.com/ethereum/go-ethereum/crypto/sha3" ) +const MaxPO = 7 + type Hasher func() hash.Hash // Peer is the recorded as Source on the chunk @@ -73,22 +75,24 @@ func (h Key) bits(i, j uint) uint { return res } -/* -func proximity(one, other []byte) (ret int) { - retbig, _ := binary.Varint(other) - ret = int(int8(retbig)) - return -}*/ func Proximity(one, other []byte) (ret int) { - for i := 0; i < len(one); i++ { + 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] - for j := 0; j < 8; j++ { + 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 len(one) * 8 + return MaxPO } func IsZeroKey(key Key) bool { diff --git a/swarm/storage/types_test.go b/swarm/storage/types_test.go deleted file mode 100644 index ceb7a635c5..0000000000 --- a/swarm/storage/types_test.go +++ /dev/null @@ -1,17 +0,0 @@ -// Copyright 2016 The go-ethereum Authors -// This file is part of the go-ethereum library. -// -// The go-ethereum library is free software: you can redistribute it and/or modify -// it under the terms of the GNU Lesser General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// The go-ethereum library is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Lesser General Public License for more details. -// -// You should have received a copy of the GNU Lesser General Public License -// along with the go-ethereum library. If not, see . - -package storage diff --git a/swarm/testutil/http.go b/swarm/testutil/http.go index f2922fab00..fbac14f0d7 100644 --- a/swarm/testutil/http.go +++ b/swarm/testutil/http.go @@ -38,7 +38,7 @@ func NewTestSwarmServer(t *testing.T) *TestSwarmServer { CacheCapacity: 5000, Radius: 0, } - localStore, err := storage.NewLocalStore(storage.MakeHashFunc("SHA3"), storeparams) + localStore, err := storage.NewLocalStore(storage.MakeHashFunc("SHA3"), storeparams, storage.ZeroKey) if err != nil { os.RemoveAll(dir) t.Fatal(err)