diff --git a/swarm/network/stream/common_test.go b/swarm/network/stream/common_test.go index ec29e16e33..07a96d7f68 100644 --- a/swarm/network/stream/common_test.go +++ b/swarm/network/stream/common_test.go @@ -30,15 +30,18 @@ import ( "sync/atomic" "time" + "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/p2p/simulations/adapters" p2ptest "github.com/ethereum/go-ethereum/p2p/testing" + "github.com/ethereum/go-ethereum/swarm/chunk" "github.com/ethereum/go-ethereum/swarm/network" "github.com/ethereum/go-ethereum/swarm/network/simulation" "github.com/ethereum/go-ethereum/swarm/state" "github.com/ethereum/go-ethereum/swarm/storage" - mockmem "github.com/ethereum/go-ethereum/swarm/storage/mock/mem" + "github.com/ethereum/go-ethereum/swarm/storage/localstore" + "github.com/ethereum/go-ethereum/swarm/storage/mock" "github.com/ethereum/go-ethereum/swarm/testutil" colorable "github.com/mattn/go-colorable" ) @@ -50,7 +53,6 @@ var ( useMockStore = flag.Bool("mockstore", false, "disabled mock store (default: enabled)") longrunning = flag.Bool("longrunning", false, "do run long-running tests") - bucketKeyDB = simulation.BucketKey("db") bucketKeyStore = simulation.BucketKey("store") bucketKeyFileStore = simulation.BucketKey("filestore") bucketKeyNetStore = simulation.BucketKey("netstore") @@ -112,16 +114,15 @@ func newNetStoreAndDeliveryWithRequestFunc(ctx *adapters.ServiceContext, bucket func netStoreAndDeliveryWithAddr(ctx *adapters.ServiceContext, bucket *sync.Map, addr *network.BzzAddr) (*storage.NetStore, *Delivery, func(), error) { n := ctx.Config.Node() - store, datadir, err := createTestLocalStorageForID(n.ID(), addr) - if *useMockStore { - store, datadir, err = createMockStore(mockmem.NewGlobalStore(), n.ID(), addr) - } + localStore, localStoreCleanup, err := newTestLocalStore(n.ID(), addr, nil) if err != nil { return nil, nil, nil, err } - localStore := store.(*storage.LocalStore) + netStore, err := storage.NewNetStore(localStore, nil) if err != nil { + localStoreCleanup() + localStore.Close() return nil, nil, nil, err } @@ -130,8 +131,7 @@ func netStoreAndDeliveryWithAddr(ctx *adapters.ServiceContext, bucket *sync.Map, kad := network.NewKademlia(addr.Over(), network.NewKadParams()) delivery := NewDelivery(kad, netStore) - bucket.Store(bucketKeyStore, store) - bucket.Store(bucketKeyDB, netStore) + bucket.Store(bucketKeyStore, localStore) bucket.Store(bucketKeyDelivery, delivery) bucket.Store(bucketKeyFileStore, fileStore) // for the kademlia object, we use the global key from the simulation package, @@ -140,13 +140,13 @@ func netStoreAndDeliveryWithAddr(ctx *adapters.ServiceContext, bucket *sync.Map, cleanup := func() { netStore.Close() - os.RemoveAll(datadir) + localStoreCleanup() } return netStore, delivery, cleanup, nil } -func newStreamerTester(registryOptions *RegistryOptions) (*p2ptest.ProtocolTester, *Registry, *storage.LocalStore, func(), error) { +func newStreamerTester(registryOptions *RegistryOptions) (*p2ptest.ProtocolTester, *Registry, *localstore.DB, func(), error) { // setup addr := network.RandomAddr() // tested peers peer address to := network.NewKademlia(addr.OAddr, network.NewKadParams()) @@ -160,11 +160,7 @@ func newStreamerTester(registryOptions *RegistryOptions) (*p2ptest.ProtocolTeste os.RemoveAll(datadir) } - params := storage.NewDefaultLocalStoreParams() - params.Init(datadir) - params.BaseKey = addr.Over() - - localStore, err := storage.NewTestLocalStoreForAddr(params) + localStore, err := localstore.New(datadir, addr.Over(), nil) if err != nil { removeDataDir() return nil, nil, nil, nil, err @@ -221,24 +217,25 @@ func newRoundRobinStore(stores ...storage.ChunkStore) *roundRobinStore { } // not used in this context, only to fulfill ChunkStore interface -func (rrs *roundRobinStore) Has(ctx context.Context, addr storage.Address) bool { - panic("RoundRobinStor doesn't support HasChunk") +func (rrs *roundRobinStore) Has(_ context.Context, _ storage.Address) (bool, error) { + return false, errors.New("RoundRobinStore doesn't support HasChunk") } -func (rrs *roundRobinStore) Get(ctx context.Context, addr storage.Address) (storage.Chunk, error) { +func (rrs *roundRobinStore) Get(_ context.Context, _ chunk.ModeGet, _ storage.Address) (storage.Chunk, error) { return nil, errors.New("get not well defined on round robin store") } -func (rrs *roundRobinStore) Put(ctx context.Context, chunk storage.Chunk) error { +func (rrs *roundRobinStore) Put(ctx context.Context, mode chunk.ModePut, ch storage.Chunk) error { i := atomic.AddUint32(&rrs.index, 1) idx := int(i) % len(rrs.stores) - return rrs.stores[idx].Put(ctx, chunk) + return rrs.stores[idx].Put(ctx, mode, ch) } -func (rrs *roundRobinStore) Close() { +func (rrs *roundRobinStore) Close() error { for _, store := range rrs.stores { store.Close() } + return nil } func readAll(fileStore *storage.FileStore, hash []byte) (int64, error) { @@ -304,24 +301,28 @@ func generateRandomFile() (string, error) { return string(b), nil } -//create a local store for the given node -func createTestLocalStorageForID(id enode.ID, addr *network.BzzAddr) (storage.ChunkStore, string, error) { - var datadir string - var err error - datadir, err = ioutil.TempDir("", fmt.Sprintf("syncer-test-%s", id.TerminalString())) +func newTestLocalStore(id enode.ID, addr *network.BzzAddr, globalStore mock.GlobalStorer) (localStore *localstore.DB, cleanup func(), err error) { + dir, err := ioutil.TempDir("", "swarm-stream-") if err != nil { - return nil, "", err + return nil, nil, err } - var store storage.ChunkStore - params := storage.NewDefaultLocalStoreParams() - params.ChunkDbPath = datadir - params.BaseKey = addr.Over() - store, err = storage.NewTestLocalStoreForAddr(params) + cleanup = func() { + os.RemoveAll(dir) + } + + var mockStore *mock.NodeStore + if globalStore != nil { + mockStore = globalStore.NewNodeStore(common.BytesToAddress(id.Bytes())) + } + + localStore, err = localstore.New(dir, addr.Over(), &localstore.Options{ + MockStore: mockStore, + }) if err != nil { - os.RemoveAll(datadir) - return nil, "", err + cleanup() + return nil, nil, err } - return store, datadir, nil + return localStore, cleanup, nil } // watchDisconnections receives simulation peer events in a new goroutine and sets atomic value diff --git a/swarm/network/stream/delivery_test.go b/swarm/network/stream/delivery_test.go index 50b7881504..26987704c7 100644 --- a/swarm/network/stream/delivery_test.go +++ b/swarm/network/stream/delivery_test.go @@ -25,6 +25,8 @@ import ( "testing" "time" + "github.com/ethereum/go-ethereum/swarm/chunk" + "github.com/ethereum/go-ethereum/node" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p/enode" @@ -189,8 +191,8 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) { }) hash := storage.Address(hash0[:]) - chunk := storage.NewChunk(hash, hash) - err = localStore.Put(context.TODO(), chunk) + ch := storage.NewChunk(hash, hash) + err = localStore.Put(context.TODO(), chunk.ModePutUpload, ch) if err != nil { t.Fatalf("Expected no err got %v", err) } @@ -241,8 +243,8 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) { } hash = storage.Address(hash1[:]) - chunk = storage.NewChunk(hash, hash1[:]) - err = localStore.Put(context.TODO(), chunk) + ch = storage.NewChunk(hash, hash1[:]) + err = localStore.Put(context.TODO(), chunk.ModePutUpload, ch) if err != nil { t.Fatalf("Expected no err got %v", err) } @@ -420,14 +422,14 @@ func TestStreamerDownstreamChunkDeliveryMsgExchange(t *testing.T) { defer cancel() // wait for the chunk to get stored - storedChunk, err := localStore.Get(ctx, chunkKey) + storedChunk, err := localStore.Get(ctx, chunk.ModeGetRequest, chunkKey) for err != nil { select { case <-ctx.Done(): t.Fatalf("Chunk is not in localstore after timeout, err: %v", err) default: } - storedChunk, err = localStore.Get(ctx, chunkKey) + storedChunk, err = localStore.Get(ctx, chunk.ModeGetRequest, chunkKey) time.Sleep(50 * time.Millisecond) } @@ -700,7 +702,7 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, chunkCount int, skipCheck b errs := make(chan error) for _, hash := range hashes { go func(h storage.Address) { - _, err := netStore.Get(ctx, h) + _, err := netStore.Get(ctx, chunk.ModeGetRequest, h) log.Warn("test check netstore get", "hash", h, "err", err) errs <- err }(hash) diff --git a/swarm/network/stream/snapshot_retrieval_test.go b/swarm/network/stream/snapshot_retrieval_test.go index 2fdf8e9e37..dd0c437f59 100644 --- a/swarm/network/stream/snapshot_retrieval_test.go +++ b/swarm/network/stream/snapshot_retrieval_test.go @@ -22,6 +22,7 @@ import ( "testing" "time" + "github.com/ethereum/go-ethereum/swarm/storage/localstore" "github.com/ethereum/go-ethereum/swarm/testutil" "github.com/ethereum/go-ethereum/node" @@ -278,7 +279,7 @@ func runRetrievalTest(t *testing.T, chunkCount int, nodeCount int) error { if !ok { return fmt.Errorf("No localstore") } - lstore := item.(*storage.LocalStore) + lstore := item.(*localstore.DB) conf.hashes, err = uploadFileToSingleNodeStore(node.ID(), chunkCount, lstore) if err != nil { return err diff --git a/swarm/network/stream/snapshot_sync_test.go b/swarm/network/stream/snapshot_sync_test.go index 9737ec0a54..90056df0a5 100644 --- a/swarm/network/stream/snapshot_sync_test.go +++ b/swarm/network/stream/snapshot_sync_test.go @@ -31,11 +31,13 @@ import ( "github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/p2p/simulations" "github.com/ethereum/go-ethereum/p2p/simulations/adapters" + "github.com/ethereum/go-ethereum/swarm/chunk" "github.com/ethereum/go-ethereum/swarm/network" "github.com/ethereum/go-ethereum/swarm/network/simulation" "github.com/ethereum/go-ethereum/swarm/pot" "github.com/ethereum/go-ethereum/swarm/state" "github.com/ethereum/go-ethereum/swarm/storage" + "github.com/ethereum/go-ethereum/swarm/storage/localstore" "github.com/ethereum/go-ethereum/swarm/storage/mock" mockmem "github.com/ethereum/go-ethereum/swarm/storage/mock/mem" "github.com/ethereum/go-ethereum/swarm/testutil" @@ -196,7 +198,7 @@ func runSim(conf *synctestConfig, ctx context.Context, sim *simulation.Simulatio if !ok { return fmt.Errorf("No localstore") } - lstore := item.(*storage.LocalStore) + lstore := item.(*localstore.DB) hashes, err := uploadFileToSingleNodeStore(node.ID(), chunkCount, lstore) if err != nil { return err @@ -225,25 +227,25 @@ func runSim(conf *synctestConfig, ctx context.Context, sim *simulation.Simulatio localChunks := conf.idToChunksMap[id] for _, ch := range localChunks { //get the real chunk by the index in the index array - chunk := conf.hashes[ch] - log.Trace(fmt.Sprintf("node has chunk: %s:", chunk)) + ch := conf.hashes[ch] + log.Trace(fmt.Sprintf("node has chunk: %s:", ch)) //check if the expected chunk is indeed in the localstore var err error if *useMockStore { //use the globalStore if the mockStore should be used; in that case, //the complete localStore stack is bypassed for getting the chunk - _, err = globalStore.Get(common.BytesToAddress(id.Bytes()), chunk) + _, err = globalStore.Get(common.BytesToAddress(id.Bytes()), ch) } else { //use the actual localstore item, ok := sim.NodeItem(id, bucketKeyStore) if !ok { return fmt.Errorf("Error accessing localstore") } - lstore := item.(*storage.LocalStore) - _, err = lstore.Get(ctx, chunk) + lstore := item.(*localstore.DB) + _, err = lstore.Get(ctx, chunk.ModeGetRequest, ch) } if err != nil { - log.Debug(fmt.Sprintf("Chunk %s NOT found for id %s", chunk, id)) + log.Debug(fmt.Sprintf("Chunk %s NOT found for id %s", ch, id)) // Do not get crazy with logging the warn message time.Sleep(500 * time.Millisecond) continue REPEAT @@ -251,10 +253,10 @@ func runSim(conf *synctestConfig, ctx context.Context, sim *simulation.Simulatio evt := &simulations.Event{ Type: EventTypeChunkArrived, Node: sim.Net.GetNode(id), - Data: chunk.String(), + Data: ch.String(), } sim.Net.Events().Send(evt) - log.Debug(fmt.Sprintf("Chunk %s IS FOUND for id %s", chunk, id)) + log.Debug(fmt.Sprintf("Chunk %s IS FOUND for id %s", ch, id)) } } return nil @@ -300,7 +302,7 @@ func mapKeysToNodes(conf *synctestConfig) { } //upload a file(chunks) to a single local node store -func uploadFileToSingleNodeStore(id enode.ID, chunkCount int, lstore *storage.LocalStore) ([]storage.Address, error) { +func uploadFileToSingleNodeStore(id enode.ID, chunkCount int, lstore *localstore.DB) ([]storage.Address, error) { log.Debug(fmt.Sprintf("Uploading to node id: %s", id)) fileStore := storage.NewFileStore(lstore, storage.NewFileStoreParams()) size := chunkSize diff --git a/swarm/network/stream/syncer_test.go b/swarm/network/stream/syncer_test.go index 07586714e1..f99c86e23f 100644 --- a/swarm/network/stream/syncer_test.go +++ b/swarm/network/stream/syncer_test.go @@ -21,13 +21,13 @@ import ( "errors" "fmt" "io/ioutil" - "math" "os" "sync" "testing" "time" - "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/swarm/chunk" + "github.com/ethereum/go-ethereum/node" "github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/p2p/simulations/adapters" @@ -36,7 +36,7 @@ import ( "github.com/ethereum/go-ethereum/swarm/network/simulation" "github.com/ethereum/go-ethereum/swarm/state" "github.com/ethereum/go-ethereum/swarm/storage" - "github.com/ethereum/go-ethereum/swarm/storage/mock" + "github.com/ethereum/go-ethereum/swarm/storage/localstore" "github.com/ethereum/go-ethereum/swarm/testutil" ) @@ -55,24 +55,6 @@ func TestSyncerSimulation(t *testing.T) { } } -func createMockStore(globalStore mock.GlobalStorer, id enode.ID, addr *network.BzzAddr) (lstore storage.ChunkStore, datadir string, err error) { - address := common.BytesToAddress(id.Bytes()) - mockStore := globalStore.NewNodeStore(address) - params := storage.NewDefaultLocalStoreParams() - - datadir, err = ioutil.TempDir("", "localMockStore-"+id.TerminalString()) - if err != nil { - return nil, "", err - } - params.Init(datadir) - params.BaseKey = addr.Over() - lstore, err = storage.NewLocalStore(params, mockStore) - if err != nil { - return nil, "", err - } - return lstore, datadir, nil -} - func testSyncBetweenNodes(t *testing.T, nodes, chunkCount int, skipCheck bool, po uint8) { sim := simulation.New(map[string]simulation.ServiceFunc{ @@ -181,17 +163,29 @@ func testSyncBetweenNodes(t *testing.T, nodes, chunkCount int, skipCheck bool, p if i < nodes-1 { hashCounts[i] = hashCounts[i+1] } - item, ok := sim.NodeItem(nodeIDs[i], bucketKeyDB) + item, ok := sim.NodeItem(nodeIDs[i], bucketKeyStore) if !ok { return fmt.Errorf("No DB") } - netStore := item.(*storage.NetStore) - netStore.Iterator(0, math.MaxUint64, po, func(addr storage.Address, index uint64) bool { - hashes[i] = append(hashes[i], addr) - totalHashes++ - hashCounts[i]++ - return true - }) + localStore := item.(*localstore.DB) + until, err := localStore.LastPullSubscriptionChunk(po) + if err != nil { + return err + } + c, _ := localStore.SubscribePull(ctx, po, nil, until) + for iterate := true; iterate; { + select { + case cd, ok := <-c: + if !ok { + iterate = false + } + hashes[i] = append(hashes[i], cd.Address) + totalHashes++ + hashCounts[i]++ + case <-ctx.Done(): + return ctx.Err() + } + } } var total, found int for _, node := range nodeIDs { @@ -200,12 +194,12 @@ func testSyncBetweenNodes(t *testing.T, nodes, chunkCount int, skipCheck bool, p for j := i; j < nodes; j++ { total += len(hashes[j]) for _, key := range hashes[j] { - item, ok := sim.NodeItem(nodeIDs[j], bucketKeyDB) + item, ok := sim.NodeItem(nodeIDs[j], bucketKeyStore) if !ok { return fmt.Errorf("No DB") } db := item.(*storage.NetStore) - _, err := db.Get(ctx, key) + _, err := db.Get(ctx, chunk.ModeGetRequest, key) if err == nil { found++ }