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

@@ -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)
 	}

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 roundCnt, roundsCnt, totalCnt uint
-	var quit, wait bool
 	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 {

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

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) {
+func (r *brokenLimitedReader) Read(buf []byte) (int, error) {
-	if self.off+len(buf) > self.errAt {
+	if r.off+len(buf) > r.errAt {
 		return 0, fmt.Errorf("Broken reader")
 	}
-	self.off += len(buf)
+	r.off += len(buf)
-	return self.lr.Read(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) {
+func testStoreRandom(m ChunkStore, processors int, n int, chunksize int, t *testing.T) {
-
+	hs := mputRandomKey(m, processors, n, chunksize)
-	chunkC := make(chan *Chunk)
+	err := mget(m, hs, nil)
-	go func() {
+	if err != nil {
-		for chunk := range chunkC {
+		t.Fatalf("testStore failed: %v", err)
-			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)
 	}
-	close(chunkC)
-	<-quit
 }
 
-// only put, but fills an array supplied by the caller with the keys
+func testStoreCorrect(m ChunkStore, processors int, n int, chunksize int, t *testing.T) {
-func testSplit(m ChunkStore, l int64, branches int64, chunkkeys []Key, t *testing.T) Key {
+	hs := mputChunks(m, processors, n, chunksize, sha3.NewKeccak256())
-	var i int
+	f := func(h Key, chunk *Chunk) error {
-	chunkC := make(chan *Chunk)
+		if !bytes.Equal(h, chunk.Key) {
-	go func() {
+			return fmt.Errorf("key does not match retrieved chunk Key")
-		for chunk := range chunkC {
-			chunkkeys[i] = chunk.Key
-			i++
-			m.Put(chunk)
-			if chunk.wg != nil {
-				chunk.wg.Done()
-			}
 		}
-	}()
+		hasher := sha3.NewKeccak256()
-	chunker := NewTreeChunker(&ChunkerParams{
+		hasher.Write(chunk.SData)
-		Branches: branches,
+		exp := hasher.Sum(nil)
-		Hash:     defaultHash,
+		if !bytes.Equal(h, exp) {
-	})
+			return fmt.Errorf("key is not hash of chunk data")
-	swg := &sync.WaitGroup{}
+		}
-	key, _ := chunker.Split(rand.Reader, l, chunkC, swg, nil)
+		return nil
-	swg.Wait()
+	}
-	close(chunkC)
+	err := mget(m, hs, f)
-	return key
+	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)
+		}
+	}
 }

swarm/storage/database.go

@@ -29,7 +29,6 @@ import (
 )
 
 const openFileLimit = 128
-//const openFileLimit = -1
 
 type LDBDatabase struct {
 	db   *leveldb.DB

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
 		}
 
-		//
+		if !s.trusted {
-		data_mod := data[32:]
+			data_mod := data[32:]
+			hasher := s.hashfunc()
+			hasher.Write(data_mod)
+			hash := hasher.Sum(nil)
 
-		hasher := s.hashfunc()
+			if !bytes.Equal(hash, key) {
-		hasher.Write(data_mod)
+				log.Trace(fmt.Sprintf("Apparent key/hash mismatch. Hash %x, key %v", hash, key[:]))
-		hash := hasher.Sum(nil)
+				s.delete(indx.Idx, getIndexKey(key), s.po(key))
-
+				log.Warn("Invalid Chunk in Database. Please repair with command: 'swarm cleandb'")
-		if !bytes.Equal(hash, key) {
+			}
-			s.delete(index.Idx, getIndexKey(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()

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"))
+	basekey := make([]byte, 32)
-	m, err := NewDbStore(dir, MakeHashFunc(defaultHash), defaultDbCapacity, func(k Key) (ret uint8) { return uint8(proximity(basekey[:], k[:])) })
+	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) {
+func testDbStoreRandom(n int, processors int, chunksize int, t *testing.T) {
-	t.Skip()
+	db, err := newTestDbStore()
-	if indata == nil {
+	if err != nil {
-		indata = rand.Reader
+		t.Fatalf("init dbStore failed: %v", err)
 	}
-	m := initDbStore(t)
+	defer db.close()
-	defer m.Close()
+	db.trusted = true
-	testStore(m, indata, l, branches, t)
+	testStoreRandom(db, processors, n, chunksize, t)
 }
 
-func TestDbStore128_0x1000000(t *testing.T) {
+func testDbStoreCorrect(n int, processors int, chunksize int, t *testing.T) {
-	testDbStore(nil, 0x1000000, 128, t)
+	db, err := newTestDbStore()
-}
+	if err != nil {
-
+		t.Fatalf("init dbStore failed: %v", err)
-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
 	}
-	for c = 0; c < l; c++ {
+	defer db.close()
-		b = append(b, byte(i))
+	testStoreCorrect(db, processors, n, chunksize, t)
-		if i == p {
+}
-			i = 0
+
-		} else {
+func TestDbStoreRandom_1(t *testing.T) {
-			i++
+	testDbStoreRandom(1, 1, 0, t)
-		}
+}
-	}
+
-	return bytes.NewReader(b)
+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)
+	db, err := newTestDbStore()
-	defer m.Close()
+	if err != nil {
-	_, err := m.Get(ZeroKey)
+		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)
+	db, err := newTestDbStore()
-	defer m.Close()
+	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)
+}

swarm/storage/dpa_test.go

@@ -28,12 +28,17 @@ import (
 const testDataSize = 0x1000000
 
 func TestDPArandom(t *testing.T) {
-	dbStore := initDbStore(t)
+	tdb, err := newTestDbStore()
-	dbStore.setCapacity(50000)
+	if err != nil {
-	memStore := NewMemStore(dbStore, defaultCacheCapacity)
+		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)
+	tdb, err := newTestDbStore()
-	memStore := NewMemStore(dbStore, defaultCacheCapacity)
+	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,

swarm/storage/memstore_test.go

@@ -16,51 +16,82 @@
 
 package storage
 
-import (
+import "testing"
-	"bytes"
-	"crypto/rand"
-	"io"
-	"testing"
-)
 
-func testMemStore(indata io.Reader, l int64, branches int64, t *testing.T) {
+func newTestMemStore() *MemStore {
-	if indata == nil {
+	return NewMemStore(nil, defaultCacheCapacity)
-		indata = rand.Reader
-	}
-	m := NewMemStore(nil, defaultCacheCapacity)
-	testStore(m, indata, l, branches, t)
 }
 
-func TestMemStore128_10000(t *testing.T) {
+func testMemStoreRandom(n int, processors int, chunksize int, t *testing.T) {
-	testMemStore(nil, 10000, 128, t)
+	m := newTestMemStore()
+	defer m.Close()
+	testStoreRandom(m, processors, n, chunksize, t)
 }
 
-func TestMemStore128_1000(t *testing.T) {
+func testMemStoreCorrect(n int, processors int, chunksize int, t *testing.T) {
-	testMemStore(nil, 1000, 128, t)
+	m := newTestMemStore()
+	defer m.Close()
+	testStoreCorrect(m, processors, n, chunksize, t)
 }
 
-func TestMemStore128_100(t *testing.T) {
+func TestMemStoreRandom_1(t *testing.T) {
-	testMemStore(nil, 100, 128, t)
+	testMemStoreRandom(1, 1, 0, t)
 }
 
-func TestMemStore2_100(t *testing.T) {
+func TestMemStoreCorrect_1(t *testing.T) {
-	testMemStore(nil, 100, 2, t)
+	testMemStoreCorrect(1, 1, 4104, t)
 }
 
-func TestMemStore2_100_fixed_(t *testing.T) {
+func TestMemStoreRandom_1_10k(t *testing.T) {
-	b := []byte{}
+	testMemStoreRandom(1, 5000, 0, t)
-	for i := 0; i < 100; i++ {
+}
-		b = append(b, byte(i))
-	}
 
-	br := bytes.NewReader(b)
+func TestMemStoreCorrect_1_10k(t *testing.T) {
-	testMemStore(br, 100, 2, 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)
+}

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 {

swarm/storage/types_test.go

@@ -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 <http://www.gnu.org/licenses/>.
-
-package storage

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)