gaspersic/go-ethereum
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

swarm/chunk: move chunk related declarations to chunk package (#19170)

Browse Source
This commit is contained in:
Janoš Guljaš
2019-02-26 16:09:32 +01:00
committed by Anton Evangelatov
parent b7e0dec6bd
commit f0233948d2
28 changed files with 325 additions and 286 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
swarm/storage/localstore/gc_test.go
View File

@@ -23,7 +23,7 @@ import (
"testing"
"time"
"github.com/ethereum/go-ethereum/swarm/storage"
"github.com/ethereum/go-ethereum/swarm/chunk"
)
// TestDB_collectGarbageWorker tests garbage collection runs
@@ -64,11 +64,11 @@ func testDB_collectGarbageWorker(t *testing.T) {
uploader := db.NewPutter(ModePutUpload)
syncer := db.NewSetter(ModeSetSync)
addrs := make([]storage.Address, 0)
addrs := make([]chunk.Address, 0)
// upload random chunks
for i := 0; i < chunkCount; i++ {
chunk := generateRandomChunk()
chunk := generateTestRandomChunk()
err := uploader.Put(chunk)
if err != nil {
@@ -106,8 +106,8 @@ func testDB_collectGarbageWorker(t *testing.T) {
// the first synced chunk should be removed
t.Run("get the first synced chunk", func(t *testing.T) {
_, err := db.NewGetter(ModeGetRequest).Get(addrs[0])
if err != storage.ErrChunkNotFound {
t.Errorf("got error %v, want %v", err, storage.ErrChunkNotFound)
if err != chunk.ErrChunkNotFound {
t.Errorf("got error %v, want %v", err, chunk.ErrChunkNotFound)
}
})
@@ -137,11 +137,11 @@ func TestDB_collectGarbageWorker_withRequests(t *testing.T) {
testHookCollectGarbageChan <- collectedCount
})()
addrs := make([]storage.Address, 0)
addrs := make([]chunk.Address, 0)
// upload random chunks just up to the capacity
for i := 0; i < int(db.capacity)-1; i++ {
chunk := generateRandomChunk()
chunk := generateTestRandomChunk()
err := uploader.Put(chunk)
if err != nil {
@@ -156,6 +156,14 @@ func TestDB_collectGarbageWorker_withRequests(t *testing.T) {
addrs = append(addrs, chunk.Address())
}
// set update gc test hook to signal when
// update gc goroutine is done by closing
// testHookUpdateGCChan channel
testHookUpdateGCChan := make(chan struct{})
resetTestHookUpdateGC := setTestHookUpdateGC(func() {
close(testHookUpdateGCChan)
})
// request the latest synced chunk
// to prioritize it in the gc index
// not to be collected
@@ -164,18 +172,29 @@ func TestDB_collectGarbageWorker_withRequests(t *testing.T) {
t.Fatal(err)
}
// wait for update gc goroutine to finish for garbage
// collector to be correctly triggered after the last upload
select {
case <-testHookUpdateGCChan:
case <-time.After(10 * time.Second):
t.Fatal("updateGC was not called after getting chunk with ModeGetRequest")
}
// no need to wait for update gc hook anymore
resetTestHookUpdateGC()
// upload and sync another chunk to trigger
// garbage collection
chunk := generateRandomChunk()
err = uploader.Put(chunk)
ch := generateTestRandomChunk()
err = uploader.Put(ch)
if err != nil {
t.Fatal(err)
}
err = syncer.Set(chunk.Address())
err = syncer.Set(ch.Address())
if err != nil {
t.Fatal(err)
}
addrs = append(addrs, chunk.Address())
addrs = append(addrs, ch.Address())
// wait for garbage collection
@@ -217,8 +236,8 @@ func TestDB_collectGarbageWorker_withRequests(t *testing.T) {
// the second synced chunk should be removed
t.Run("get gc-ed chunk", func(t *testing.T) {
_, err := db.NewGetter(ModeGetRequest).Get(addrs[1])
if err != storage.ErrChunkNotFound {
t.Errorf("got error %v, want %v", err, storage.ErrChunkNotFound)
if err != chunk.ErrChunkNotFound {
t.Errorf("got error %v, want %v", err, chunk.ErrChunkNotFound)
}
})
@@ -254,7 +273,7 @@ func TestDB_gcSize(t *testing.T) {
count := 100
for i := 0; i < count; i++ {
chunk := generateRandomChunk()
chunk := generateTestRandomChunk()
err := uploader.Put(chunk)
if err != nil {

10
swarm/storage/localstore/index_test.go
View File

@@ -21,7 +21,7 @@ import (
"math/rand"
"testing"
"github.com/ethereum/go-ethereum/swarm/storage"
"github.com/ethereum/go-ethereum/swarm/chunk"
)
// TestDB_pullIndex validates the ordering of keys in pull index.
@@ -43,7 +43,7 @@ func TestDB_pullIndex(t *testing.T) {
// upload random chunks
for i := 0; i < chunkCount; i++ {
chunk := generateRandomChunk()
chunk := generateTestRandomChunk()
err := uploader.Put(chunk)
if err != nil {
@@ -62,8 +62,8 @@ func TestDB_pullIndex(t *testing.T) {
}
testItemsOrder(t, db.pullIndex, chunks, func(i, j int) (less bool) {
poi := storage.Proximity(db.baseKey, chunks[i].Address())
poj := storage.Proximity(db.baseKey, chunks[j].Address())
poi := chunk.Proximity(db.baseKey, chunks[i].Address())
poj := chunk.Proximity(db.baseKey, chunks[j].Address())
if poi < poj {
return true
}
@@ -95,7 +95,7 @@ func TestDB_gcIndex(t *testing.T) {
// upload random chunks
for i := 0; i < chunkCount; i++ {
chunk := generateRandomChunk()
chunk := generateTestRandomChunk()
err := uploader.Put(chunk)
if err != nil {

12
swarm/storage/localstore/localstore.go
View File

@@ -24,8 +24,8 @@ import (
"time"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/swarm/chunk"
"github.com/ethereum/go-ethereum/swarm/shed"
"github.com/ethereum/go-ethereum/swarm/storage"
"github.com/ethereum/go-ethereum/swarm/storage/mock"
)
@@ -392,8 +392,8 @@ func (db *DB) Close() (err error) {
// po computes the proximity order between the address
// and database base key.
func (db *DB) po(addr storage.Address) (bin uint8) {
return uint8(storage.Proximity(db.baseKey, addr))
func (db *DB) po(addr chunk.Address) (bin uint8) {
return uint8(chunk.Proximity(db.baseKey, addr))
}
var (
@@ -409,7 +409,7 @@ var (
// If the address is locked this function will check it
// in a for loop for addressLockTimeout time, after which
// it will return ErrAddressLockTimeout error.
func (db *DB) lockAddr(addr storage.Address) (unlock func(), err error) {
func (db *DB) lockAddr(addr chunk.Address) (unlock func(), err error) {
start := time.Now()
lockKey := hex.EncodeToString(addr)
for {
@@ -426,7 +426,7 @@ func (db *DB) lockAddr(addr storage.Address) (unlock func(), err error) {
}
// chunkToItem creates new Item with data provided by the Chunk.
func chunkToItem(ch storage.Chunk) shed.Item {
func chunkToItem(ch chunk.Chunk) shed.Item {
return shed.Item{
Address: ch.Address(),
Data: ch.Data(),
@@ -434,7 +434,7 @@ func chunkToItem(ch storage.Chunk) shed.Item {
}
// addressToItem creates new Item with a provided address.
func addressToItem(addr storage.Address) shed.Item {
func addressToItem(addr chunk.Address) shed.Item {
return shed.Item{
Address: addr,
}

57
swarm/storage/localstore/localstore_test.go
View File

@@ -29,9 +29,8 @@ import (
"testing"
"time"
ch "github.com/ethereum/go-ethereum/swarm/chunk"
"github.com/ethereum/go-ethereum/swarm/chunk"
"github.com/ethereum/go-ethereum/swarm/shed"
"github.com/ethereum/go-ethereum/swarm/storage"
"github.com/syndtr/goleveldb/leveldb"
)
@@ -61,7 +60,7 @@ func TestDB(t *testing.T) {
db, cleanupFunc := newTestDB(t, nil)
defer cleanupFunc()
chunk := generateRandomChunk()
chunk := generateTestRandomChunk()
err := db.NewPutter(ModePutUpload).Put(chunk)
if err != nil {
@@ -115,7 +114,7 @@ func TestDB_updateGCSem(t *testing.T) {
db, cleanupFunc := newTestDB(t, nil)
defer cleanupFunc()
chunk := generateRandomChunk()
chunk := generateTestRandomChunk()
err := db.NewPutter(ModePutUpload).Put(chunk)
if err != nil {
@@ -188,7 +187,7 @@ func BenchmarkNew(b *testing.B) {
uploader := db.NewPutter(ModePutUpload)
syncer := db.NewSetter(ModeSetSync)
for i := 0; i < count; i++ {
chunk := generateFakeRandomChunk()
chunk := generateTestRandomChunk()
err := uploader.Put(chunk)
if err != nil {
b.Fatal(err)
@@ -251,53 +250,47 @@ func newTestDB(t testing.TB, o *Options) (db *DB, cleanupFunc func()) {
return db, cleanupFunc
}
// generateRandomChunk generates a valid Chunk with
// data size of default chunk size.
func generateRandomChunk() storage.Chunk {
return storage.GenerateRandomChunk(ch.DefaultSize)
}
func init() {
// needed for generateFakeRandomChunk
// needed for generateTestRandomChunk
rand.Seed(time.Now().UnixNano())
}
// generateFakeRandomChunk generates a Chunk that is not
// generateTestRandomChunk generates a Chunk that is not
// valid, but it contains a random key and a random value.
// This function is faster then storage.GenerateRandomChunk
// This function is faster then storage.generateTestRandomChunk
// which generates a valid chunk.
// Some tests in this package do not need valid chunks, just
// random data, and their execution time can be decreased
// using this function.
func generateFakeRandomChunk() storage.Chunk {
data := make([]byte, ch.DefaultSize)
func generateTestRandomChunk() chunk.Chunk {
data := make([]byte, chunk.DefaultSize)
rand.Read(data)
key := make([]byte, 32)
rand.Read(key)
return storage.NewChunk(key, data)
return chunk.NewChunk(key, data)
}
// TestGenerateFakeRandomChunk validates that
// generateFakeRandomChunk returns random data by comparing
// TestGenerateTestRandomChunk validates that
// generateTestRandomChunk returns random data by comparing
// two generated chunks.
func TestGenerateFakeRandomChunk(t *testing.T) {
c1 := generateFakeRandomChunk()
c2 := generateFakeRandomChunk()
func TestGenerateTestRandomChunk(t *testing.T) {
c1 := generateTestRandomChunk()
c2 := generateTestRandomChunk()
addrLen := len(c1.Address())
if addrLen != 32 {
t.Errorf("first chunk address length %v, want %v", addrLen, 32)
}
dataLen := len(c1.Data())
if dataLen != ch.DefaultSize {
t.Errorf("first chunk data length %v, want %v", dataLen, ch.DefaultSize)
if dataLen != chunk.DefaultSize {
t.Errorf("first chunk data length %v, want %v", dataLen, chunk.DefaultSize)
}
addrLen = len(c2.Address())
if addrLen != 32 {
t.Errorf("second chunk address length %v, want %v", addrLen, 32)
}
dataLen = len(c2.Data())
if dataLen != ch.DefaultSize {
t.Errorf("second chunk data length %v, want %v", dataLen, ch.DefaultSize)
if dataLen != chunk.DefaultSize {
t.Errorf("second chunk data length %v, want %v", dataLen, chunk.DefaultSize)
}
if bytes.Equal(c1.Address(), c2.Address()) {
t.Error("fake chunks addresses do not differ")
@@ -309,7 +302,7 @@ func TestGenerateFakeRandomChunk(t *testing.T) {
// newRetrieveIndexesTest returns a test function that validates if the right
// chunk values are in the retrieval indexes.
func newRetrieveIndexesTest(db *DB, chunk storage.Chunk, storeTimestamp, accessTimestamp int64) func(t *testing.T) {
func newRetrieveIndexesTest(db *DB, chunk chunk.Chunk, storeTimestamp, accessTimestamp int64) func(t *testing.T) {
return func(t *testing.T) {
item, err := db.retrievalDataIndex.Get(addressToItem(chunk.Address()))
if err != nil {
@@ -328,7 +321,7 @@ func newRetrieveIndexesTest(db *DB, chunk storage.Chunk, storeTimestamp, accessT
// newRetrieveIndexesTestWithAccess returns a test function that validates if the right
// chunk values are in the retrieval indexes when access time must be stored.
func newRetrieveIndexesTestWithAccess(db *DB, chunk storage.Chunk, storeTimestamp, accessTimestamp int64) func(t *testing.T) {
func newRetrieveIndexesTestWithAccess(db *DB, chunk chunk.Chunk, storeTimestamp, accessTimestamp int64) func(t *testing.T) {
return func(t *testing.T) {
item, err := db.retrievalDataIndex.Get(addressToItem(chunk.Address()))
if err != nil {
@@ -348,7 +341,7 @@ func newRetrieveIndexesTestWithAccess(db *DB, chunk storage.Chunk, storeTimestam
// newPullIndexTest returns a test function that validates if the right
// chunk values are in the pull index.
func newPullIndexTest(db *DB, chunk storage.Chunk, storeTimestamp int64, wantError error) func(t *testing.T) {
func newPullIndexTest(db *DB, chunk chunk.Chunk, storeTimestamp int64, wantError error) func(t *testing.T) {
return func(t *testing.T) {
item, err := db.pullIndex.Get(shed.Item{
Address: chunk.Address(),
@@ -365,7 +358,7 @@ func newPullIndexTest(db *DB, chunk storage.Chunk, storeTimestamp int64, wantErr
// newPushIndexTest returns a test function that validates if the right
// chunk values are in the push index.
func newPushIndexTest(db *DB, chunk storage.Chunk, storeTimestamp int64, wantError error) func(t *testing.T) {
func newPushIndexTest(db *DB, chunk chunk.Chunk, storeTimestamp int64, wantError error) func(t *testing.T) {
return func(t *testing.T) {
item, err := db.pushIndex.Get(shed.Item{
Address: chunk.Address(),
@@ -382,7 +375,7 @@ func newPushIndexTest(db *DB, chunk storage.Chunk, storeTimestamp int64, wantErr
// newGCIndexTest returns a test function that validates if the right
// chunk values are in the push index.
func newGCIndexTest(db *DB, chunk storage.Chunk, storeTimestamp, accessTimestamp int64) func(t *testing.T) {
func newGCIndexTest(db *DB, chunk chunk.Chunk, storeTimestamp, accessTimestamp int64) func(t *testing.T) {
return func(t *testing.T) {
item, err := db.gcIndex.Get(shed.Item{
Address: chunk.Address(),
@@ -436,7 +429,7 @@ func newIndexGCSizeTest(db *DB) func(t *testing.T) {
// testIndexChunk embeds storageChunk with additional data that is stored
// in database. It is used for index values validations.
type testIndexChunk struct {
storage.Chunk
chunk.Chunk
storeTimestamp int64
}

12
swarm/storage/localstore/mode_get.go
View File

@@ -18,8 +18,8 @@ package localstore
import (
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/swarm/chunk"
"github.com/ethereum/go-ethereum/swarm/shed"
"github.com/ethereum/go-ethereum/swarm/storage"
"github.com/syndtr/goleveldb/leveldb"
)
@@ -51,23 +51,23 @@ func (db *DB) NewGetter(mode ModeGet) *Getter {
}
// Get returns a chunk from the database. If the chunk is
// not found storage.ErrChunkNotFound will be returned.
// not found chunk.ErrChunkNotFound will be returned.
// All required indexes will be updated required by the
// Getter Mode.
func (g *Getter) Get(addr storage.Address) (chunk storage.Chunk, err error) {
func (g *Getter) Get(addr chunk.Address) (ch chunk.Chunk, err error) {
out, err := g.db.get(g.mode, addr)
if err != nil {
if err == leveldb.ErrNotFound {
return nil, storage.ErrChunkNotFound
return nil, chunk.ErrChunkNotFound
}
return nil, err
}
return storage.NewChunk(out.Address, out.Data), nil
return chunk.NewChunk(out.Address, out.Data), nil
}
// get returns Item from the retrieval index
// and updates other indexes.
func (db *DB) get(mode ModeGet, addr storage.Address) (out shed.Item, err error) {
func (db *DB) get(mode ModeGet, addr chunk.Address) (out shed.Item, err error) {
item := addressToItem(addr)
out, err = db.retrievalDataIndex.Get(item)

4
swarm/storage/localstore/mode_get_test.go
View File

@@ -32,7 +32,7 @@ func TestModeGetRequest(t *testing.T) {
return uploadTimestamp
})()
chunk := generateRandomChunk()
chunk := generateTestRandomChunk()
err := db.NewPutter(ModePutUpload).Put(chunk)
if err != nil {
@@ -146,7 +146,7 @@ func TestModeGetSync(t *testing.T) {
return uploadTimestamp
})()
chunk := generateRandomChunk()
chunk := generateTestRandomChunk()
err := db.NewPutter(ModePutUpload).Put(chunk)
if err != nil {

4
swarm/storage/localstore/mode_put.go
View File

@@ -17,8 +17,8 @@
package localstore
import (
"github.com/ethereum/go-ethereum/swarm/chunk"
"github.com/ethereum/go-ethereum/swarm/shed"
"github.com/ethereum/go-ethereum/swarm/storage"
"github.com/syndtr/goleveldb/leveldb"
)
@@ -53,7 +53,7 @@ func (db *DB) NewPutter(mode ModePut) *Putter {
// Put stores the Chunk to database and depending
// on the Putter mode, it updates required indexes.
func (p *Putter) Put(ch storage.Chunk) (err error) {
func (p *Putter) Put(ch chunk.Chunk) (err error) {
return p.db.put(p.mode, chunkToItem(ch))
}

18
swarm/storage/localstore/mode_put_test.go
View File

@@ -23,7 +23,7 @@ import (
"testing"
"time"
"github.com/ethereum/go-ethereum/swarm/storage"
"github.com/ethereum/go-ethereum/swarm/chunk"
)
// TestModePutRequest validates ModePutRequest index values on the provided DB.
@@ -33,7 +33,7 @@ func TestModePutRequest(t *testing.T) {
putter := db.NewPutter(ModePutRequest)
chunk := generateRandomChunk()
chunk := generateTestRandomChunk()
// keep the record when the chunk is stored
var storeTimestamp int64
@@ -87,7 +87,7 @@ func TestModePutSync(t *testing.T) {
return wantTimestamp
})()
chunk := generateRandomChunk()
chunk := generateTestRandomChunk()
err := db.NewPutter(ModePutSync).Put(chunk)
if err != nil {
@@ -109,7 +109,7 @@ func TestModePutUpload(t *testing.T) {
return wantTimestamp
})()
chunk := generateRandomChunk()
chunk := generateTestRandomChunk()
err := db.NewPutter(ModePutUpload).Put(chunk)
if err != nil {
@@ -132,7 +132,7 @@ func TestModePutUpload_parallel(t *testing.T) {
chunkCount := 1000
workerCount := 100
chunkChan := make(chan storage.Chunk)
chunkChan := make(chan chunk.Chunk)
errChan := make(chan error)
doneChan := make(chan struct{})
defer close(doneChan)
@@ -159,13 +159,13 @@ func TestModePutUpload_parallel(t *testing.T) {
}(i)
}
chunks := make([]storage.Chunk, 0)
chunks := make([]chunk.Chunk, 0)
var chunksMu sync.Mutex
// send chunks to workers
go func() {
for i := 0; i < chunkCount; i++ {
chunk := generateRandomChunk()
chunk := generateTestRandomChunk()
select {
case chunkChan <- chunk:
case <-doneChan:
@@ -271,9 +271,9 @@ func benchmarkPutUpload(b *testing.B, o *Options, count, maxParallelUploads int)
defer cleanupFunc()
uploader := db.NewPutter(ModePutUpload)
chunks := make([]storage.Chunk, count)
chunks := make([]chunk.Chunk, count)
for i := 0; i < count; i++ {
chunks[i] = generateFakeRandomChunk()
chunks[i] = generateTestRandomChunk()
}
errs := make(chan error)
b.StartTimer()

6
swarm/storage/localstore/mode_set.go
View File

@@ -17,7 +17,7 @@
package localstore
import (
"github.com/ethereum/go-ethereum/swarm/storage"
"github.com/ethereum/go-ethereum/swarm/chunk"
"github.com/syndtr/goleveldb/leveldb"
)
@@ -53,7 +53,7 @@ func (db *DB) NewSetter(mode ModeSet) *Setter {
// Set updates database indexes for a specific
// chunk represented by the address.
func (s *Setter) Set(addr storage.Address) (err error) {
func (s *Setter) Set(addr chunk.Address) (err error) {
return s.db.set(s.mode, addr)
}
@@ -61,7 +61,7 @@ func (s *Setter) Set(addr storage.Address) (err error) {
// chunk represented by the address.
// It acquires lockAddr to protect two calls
// of this function for the same address in parallel.
func (db *DB) set(mode ModeSet, addr storage.Address) (err error) {
func (db *DB) set(mode ModeSet, addr chunk.Address) (err error) {
// protect parallel updates
unlock, err := db.lockAddr(addr)
if err != nil {

6
swarm/storage/localstore/mode_set_test.go
View File

@@ -28,7 +28,7 @@ func TestModeSetAccess(t *testing.T) {
db, cleanupFunc := newTestDB(t, nil)
defer cleanupFunc()
chunk := generateRandomChunk()
chunk := generateTestRandomChunk()
wantTimestamp := time.Now().UTC().UnixNano()
defer setNow(func() (t int64) {
@@ -56,7 +56,7 @@ func TestModeSetSync(t *testing.T) {
db, cleanupFunc := newTestDB(t, nil)
defer cleanupFunc()
chunk := generateRandomChunk()
chunk := generateTestRandomChunk()
wantTimestamp := time.Now().UTC().UnixNano()
defer setNow(func() (t int64) {
@@ -89,7 +89,7 @@ func TestModeSetRemove(t *testing.T) {
db, cleanupFunc := newTestDB(t, nil)
defer cleanupFunc()
chunk := generateRandomChunk()
chunk := generateTestRandomChunk()
err := db.NewPutter(ModePutUpload).Put(chunk)
if err != nil {

10
swarm/storage/localstore/retrieval_index_test.go
View File

@@ -20,7 +20,7 @@ import (
"strconv"
"testing"
"github.com/ethereum/go-ethereum/swarm/storage"
"github.com/ethereum/go-ethereum/swarm/chunk"
)
// BenchmarkRetrievalIndexes uploads a number of chunks in order to measure
@@ -64,9 +64,9 @@ func benchmarkRetrievalIndexes(b *testing.B, o *Options, count int) {
uploader := db.NewPutter(ModePutUpload)
syncer := db.NewSetter(ModeSetSync)
requester := db.NewGetter(ModeGetRequest)
addrs := make([]storage.Address, count)
addrs := make([]chunk.Address, count)
for i := 0; i < count; i++ {
chunk := generateFakeRandomChunk()
chunk := generateTestRandomChunk()
err := uploader.Put(chunk)
if err != nil {
b.Fatal(err)
@@ -134,9 +134,9 @@ func benchmarkUpload(b *testing.B, o *Options, count int) {
db, cleanupFunc := newTestDB(b, o)
defer cleanupFunc()
uploader := db.NewPutter(ModePutUpload)
chunks := make([]storage.Chunk, count)
chunks := make([]chunk.Chunk, count)
for i := 0; i < count; i++ {
chunk := generateFakeRandomChunk()
chunk := generateTestRandomChunk()
chunks[i] = chunk
}
b.StartTimer()

4
swarm/storage/localstore/subscription_pull.go
View File

@@ -24,8 +24,8 @@ import (
"sync"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/swarm/chunk"
"github.com/ethereum/go-ethereum/swarm/shed"
"github.com/ethereum/go-ethereum/swarm/storage"
)
// SubscribePull returns a channel that provides chunk addresses and stored times from pull syncing index.
@@ -161,7 +161,7 @@ func (db *DB) SubscribePull(ctx context.Context, bin uint8, since, until *ChunkD
// ChunkDescriptor holds information required for Pull syncing. This struct
// is provided by subscribing to pull index.
type ChunkDescriptor struct {
Address storage.Address
Address chunk.Address
StoreTimestamp int64
}

72
swarm/storage/localstore/subscription_pull_test.go
View File

@@ -24,7 +24,7 @@ import (
"testing"
"time"
"github.com/ethereum/go-ethereum/swarm/storage"
"github.com/ethereum/go-ethereum/swarm/chunk"
)
// TestDB_SubscribePull uploads some chunks before and after
@@ -37,7 +37,7 @@ func TestDB_SubscribePull(t *testing.T) {
uploader := db.NewPutter(ModePutUpload)
addrs := make(map[uint8][]storage.Address)
addrs := make(map[uint8][]chunk.Address)
var addrsMu sync.Mutex
var wantedChunksCount int
@@ -53,7 +53,7 @@ func TestDB_SubscribePull(t *testing.T) {
// to validate the number of addresses received by the subscription
errChan := make(chan error)
for bin := uint8(0); bin <= uint8(storage.MaxPO); bin++ {
for bin := uint8(0); bin <= uint8(chunk.MaxPO); bin++ {
ch, stop := db.SubscribePull(ctx, bin, nil, nil)
defer stop()
@@ -84,7 +84,7 @@ func TestDB_SubscribePull_multiple(t *testing.T) {
uploader := db.NewPutter(ModePutUpload)
addrs := make(map[uint8][]storage.Address)
addrs := make(map[uint8][]chunk.Address)
var addrsMu sync.Mutex
var wantedChunksCount int
@@ -105,7 +105,7 @@ func TestDB_SubscribePull_multiple(t *testing.T) {
// start a number of subscriptions
// that all of them will write every address error to errChan
for j := 0; j < subsCount; j++ {
for bin := uint8(0); bin <= uint8(storage.MaxPO); bin++ {
for bin := uint8(0); bin <= uint8(chunk.MaxPO); bin++ {
ch, stop := db.SubscribePull(ctx, bin, nil, nil)
defer stop()
@@ -137,7 +137,7 @@ func TestDB_SubscribePull_since(t *testing.T) {
uploader := db.NewPutter(ModePutUpload)
addrs := make(map[uint8][]storage.Address)
addrs := make(map[uint8][]chunk.Address)
var addrsMu sync.Mutex
var wantedChunksCount int
@@ -156,20 +156,20 @@ func TestDB_SubscribePull_since(t *testing.T) {
last = make(map[uint8]ChunkDescriptor)
for i := 0; i < count; i++ {
chunk := generateRandomChunk()
ch := generateTestRandomChunk()
err := uploader.Put(chunk)
err := uploader.Put(ch)
if err != nil {
t.Fatal(err)
}
bin := db.po(chunk.Address())
bin := db.po(ch.Address())
if _, ok := addrs[bin]; !ok {
addrs[bin] = make([]storage.Address, 0)
addrs[bin] = make([]chunk.Address, 0)
}
if wanted {
addrs[bin] = append(addrs[bin], chunk.Address())
addrs[bin] = append(addrs[bin], ch.Address())
wantedChunksCount++
}
@@ -178,7 +178,7 @@ func TestDB_SubscribePull_since(t *testing.T) {
lastTimestampMu.RUnlock()
last[bin] = ChunkDescriptor{
Address: chunk.Address(),
Address: ch.Address(),
StoreTimestamp: storeTimestamp,
}
}
@@ -199,7 +199,7 @@ func TestDB_SubscribePull_since(t *testing.T) {
// to validate the number of addresses received by the subscription
errChan := make(chan error)
for bin := uint8(0); bin <= uint8(storage.MaxPO); bin++ {
for bin := uint8(0); bin <= uint8(chunk.MaxPO); bin++ {
var since *ChunkDescriptor
if c, ok := last[bin]; ok {
since = &c
@@ -228,7 +228,7 @@ func TestDB_SubscribePull_until(t *testing.T) {
uploader := db.NewPutter(ModePutUpload)
addrs := make(map[uint8][]storage.Address)
addrs := make(map[uint8][]chunk.Address)
var addrsMu sync.Mutex
var wantedChunksCount int
@@ -247,20 +247,20 @@ func TestDB_SubscribePull_until(t *testing.T) {
last = make(map[uint8]ChunkDescriptor)
for i := 0; i < count; i++ {
chunk := generateRandomChunk()
ch := generateTestRandomChunk()
err := uploader.Put(chunk)
err := uploader.Put(ch)
if err != nil {
t.Fatal(err)
}
bin := db.po(chunk.Address())
bin := db.po(ch.Address())
if _, ok := addrs[bin]; !ok {
addrs[bin] = make([]storage.Address, 0)
addrs[bin] = make([]chunk.Address, 0)
}
if wanted {
addrs[bin] = append(addrs[bin], chunk.Address())
addrs[bin] = append(addrs[bin], ch.Address())
wantedChunksCount++
}
@@ -269,7 +269,7 @@ func TestDB_SubscribePull_until(t *testing.T) {
lastTimestampMu.RUnlock()
last[bin] = ChunkDescriptor{
Address: chunk.Address(),
Address: ch.Address(),
StoreTimestamp: storeTimestamp,
}
}
@@ -290,7 +290,7 @@ func TestDB_SubscribePull_until(t *testing.T) {
// to validate the number of addresses received by the subscription
errChan := make(chan error)
for bin := uint8(0); bin <= uint8(storage.MaxPO); bin++ {
for bin := uint8(0); bin <= uint8(chunk.MaxPO); bin++ {
until, ok := last[bin]
if !ok {
continue
@@ -318,7 +318,7 @@ func TestDB_SubscribePull_sinceAndUntil(t *testing.T) {
uploader := db.NewPutter(ModePutUpload)
addrs := make(map[uint8][]storage.Address)
addrs := make(map[uint8][]chunk.Address)
var addrsMu sync.Mutex
var wantedChunksCount int
@@ -337,20 +337,20 @@ func TestDB_SubscribePull_sinceAndUntil(t *testing.T) {
last = make(map[uint8]ChunkDescriptor)
for i := 0; i < count; i++ {
chunk := generateRandomChunk()
ch := generateTestRandomChunk()
err := uploader.Put(chunk)
err := uploader.Put(ch)
if err != nil {
t.Fatal(err)
}
bin := db.po(chunk.Address())
bin := db.po(ch.Address())
if _, ok := addrs[bin]; !ok {
addrs[bin] = make([]storage.Address, 0)
addrs[bin] = make([]chunk.Address, 0)
}
if wanted {
addrs[bin] = append(addrs[bin], chunk.Address())
addrs[bin] = append(addrs[bin], ch.Address())
wantedChunksCount++
}
@@ -359,7 +359,7 @@ func TestDB_SubscribePull_sinceAndUntil(t *testing.T) {
lastTimestampMu.RUnlock()
last[bin] = ChunkDescriptor{
Address: chunk.Address(),
Address: ch.Address(),
StoreTimestamp: storeTimestamp,
}
}
@@ -386,7 +386,7 @@ func TestDB_SubscribePull_sinceAndUntil(t *testing.T) {
// to validate the number of addresses received by the subscription
errChan := make(chan error)
for bin := uint8(0); bin <= uint8(storage.MaxPO); bin++ {
for bin := uint8(0); bin <= uint8(chunk.MaxPO); bin++ {
var since *ChunkDescriptor
if c, ok := upload1[bin]; ok {
since = &c
@@ -412,23 +412,23 @@ func TestDB_SubscribePull_sinceAndUntil(t *testing.T) {
// uploadRandomChunksBin uploads random chunks to database and adds them to
// the map of addresses ber bin.
func uploadRandomChunksBin(t *testing.T, db *DB, uploader *Putter, addrs map[uint8][]storage.Address, addrsMu *sync.Mutex, wantedChunksCount *int, count int) {
func uploadRandomChunksBin(t *testing.T, db *DB, uploader *Putter, addrs map[uint8][]chunk.Address, addrsMu *sync.Mutex, wantedChunksCount *int, count int) {
addrsMu.Lock()
defer addrsMu.Unlock()
for i := 0; i < count; i++ {
chunk := generateRandomChunk()
ch := generateTestRandomChunk()
err := uploader.Put(chunk)
err := uploader.Put(ch)
if err != nil {
t.Fatal(err)
}
bin := db.po(chunk.Address())
bin := db.po(ch.Address())
if _, ok := addrs[bin]; !ok {
addrs[bin] = make([]storage.Address, 0)
addrs[bin] = make([]chunk.Address, 0)
}
addrs[bin] = append(addrs[bin], chunk.Address())
addrs[bin] = append(addrs[bin], ch.Address())
*wantedChunksCount++
}
@@ -437,7 +437,7 @@ func uploadRandomChunksBin(t *testing.T, db *DB, uploader *Putter, addrs map[uin
// readPullSubscriptionBin is a helper function that reads all ChunkDescriptors from a channel and
// sends error to errChan, even if it is nil, to count the number of ChunkDescriptors
// returned by the channel.
func readPullSubscriptionBin(ctx context.Context, bin uint8, ch <-chan ChunkDescriptor, addrs map[uint8][]storage.Address, addrsMu *sync.Mutex, errChan chan error) {
func readPullSubscriptionBin(ctx context.Context, bin uint8, ch <-chan ChunkDescriptor, addrs map[uint8][]chunk.Address, addrsMu *sync.Mutex, errChan chan error) {
var i int // address index
for {
select {

8
swarm/storage/localstore/subscription_push.go
View File

@@ -21,16 +21,16 @@ import (
"sync"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/swarm/chunk"
"github.com/ethereum/go-ethereum/swarm/shed"
"github.com/ethereum/go-ethereum/swarm/storage"
)
// SubscribePush returns a channel that provides storage chunks with ordering from push syncing index.
// Returned stop function will terminate current and further iterations, and also it will close
// the returned channel without any errors. Make sure that you check the second returned parameter
// from the channel to stop iteration when its value is false.
func (db *DB) SubscribePush(ctx context.Context) (c <-chan storage.Chunk, stop func()) {
chunks := make(chan storage.Chunk)
func (db *DB) SubscribePush(ctx context.Context) (c <-chan chunk.Chunk, stop func()) {
chunks := make(chan chunk.Chunk)
trigger := make(chan struct{}, 1)
db.pushTriggersMu.Lock()
@@ -65,7 +65,7 @@ func (db *DB) SubscribePush(ctx context.Context) (c <-chan storage.Chunk, stop f
}
select {
case chunks <- storage.NewChunk(dataItem.Address, dataItem.Data):
case chunks <- chunk.NewChunk(dataItem.Address, dataItem.Data):
// set next iteration start item
// when its chunk is successfully sent to channel
sinceItem = &item

10
swarm/storage/localstore/subscription_push_test.go
View File

@@ -24,7 +24,7 @@ import (
"testing"
"time"
"github.com/ethereum/go-ethereum/swarm/storage"
"github.com/ethereum/go-ethereum/swarm/chunk"
)
// TestDB_SubscribePush uploads some chunks before and after
@@ -36,7 +36,7 @@ func TestDB_SubscribePush(t *testing.T) {
uploader := db.NewPutter(ModePutUpload)
chunks := make([]storage.Chunk, 0)
chunks := make([]chunk.Chunk, 0)
var chunksMu sync.Mutex
uploadRandomChunks := func(count int) {
@@ -44,7 +44,7 @@ func TestDB_SubscribePush(t *testing.T) {
defer chunksMu.Unlock()
for i := 0; i < count; i++ {
chunk := generateRandomChunk()
chunk := generateTestRandomChunk()
err := uploader.Put(chunk)
if err != nil {
@@ -124,7 +124,7 @@ func TestDB_SubscribePush_multiple(t *testing.T) {
uploader := db.NewPutter(ModePutUpload)
addrs := make([]storage.Address, 0)
addrs := make([]chunk.Address, 0)
var addrsMu sync.Mutex
uploadRandomChunks := func(count int) {
@@ -132,7 +132,7 @@ func TestDB_SubscribePush_multiple(t *testing.T) {
defer addrsMu.Unlock()
for i := 0; i < count; i++ {
chunk := generateRandomChunk()
chunk := generateTestRandomChunk()
err := uploader.Put(chunk)
if err != nil {