swarm/storage: Rebase resource update

This commit is contained in:
Balint Gabor 2018-01-11 13:28:02 +01:00
parent e091310e95
commit 4d4a67a9cb
2 changed files with 22 additions and 79 deletions

View file

@ -4,7 +4,6 @@ import (
"crypto/ecdsa" "crypto/ecdsa"
"encoding/binary" "encoding/binary"
"fmt" "fmt"
"path/filepath"
"strconv" "strconv"
"sync" "sync"
"time" "time"
@ -109,19 +108,9 @@ type ResourceHandler struct {
} }
// Create or open resource update chunk store // Create or open resource update chunk store
func NewResourceHandler(privKey *ecdsa.PrivateKey, datadir string, cloudStore CloudStore, ethapi *rpc.Client) (*ResourceHandler, error) { func NewResourceHandler(privKey *ecdsa.PrivateKey, hasher SwarmHasher, chunkStore ChunkStore, ethapi *rpc.Client) (*ResourceHandler, error) {
path := filepath.Join(datadir, "resource")
dbStore, err := NewDbStore(datadir, nil, singletonSwarmDbCapacity, 0)
if err != nil {
return nil, err
}
localStore := &LocalStore{
memStore: NewMemStore(dbStore, singletonSwarmDbCapacity),
DbStore: dbStore,
}
hasher := MakeHashFunc("SHA3")
return &ResourceHandler{ return &ResourceHandler{
ChunkStore: newResourceChunkStore(path, hasher, localStore, cloudStore), ChunkStore: chunkStore,
ethapi: ethapi, ethapi: ethapi,
resources: make(map[string]*resource), resources: make(map[string]*resource),
hasher: hasher(), hasher: hasher(),
@ -554,48 +543,6 @@ func (self *ResourceHandler) verifyContent(chunkdata []byte) error {
return nil return nil
} }
type resourceChunkStore struct {
localStore ChunkStore
netStore ChunkStore
}
func newResourceChunkStore(path string, hasher SwarmHasher, localStore *LocalStore, cloudStore CloudStore) *resourceChunkStore {
return &resourceChunkStore{
localStore: localStore,
netStore: NewNetStore(hasher, localStore, cloudStore, NewDefaultStoreParams()),
}
}
func (r *resourceChunkStore) Get(key Key) (*Chunk, error) {
chunk, err := r.netStore.Get(key)
if err != nil {
return nil, err
}
// if the chunk has to be remotely retrieved, we define a timeout of how long to wait for it before failing.
// sadly due to the nature of swarm, the error will never be conclusive as to whether it was a network issue
// that caused the failure or that the chunk doesn't exist.
if chunk.Req == nil {
return chunk, nil
}
t := time.NewTimer(time.Second * 1)
select {
case <-t.C:
return nil, fmt.Errorf("timeout")
case <-chunk.C:
log.Trace("Received resource update chunk", "peer", chunk.Req.Source)
}
return chunk, nil
}
func (r *resourceChunkStore) Put(chunk *Chunk) {
r.netStore.Put(chunk)
}
func (r *resourceChunkStore) Close() {
r.netStore.Close()
r.localStore.Close()
}
func getNextBlock(start uint64, current uint64, frequency uint64) uint64 { func getNextBlock(start uint64, current uint64, frequency uint64) uint64 {
blockdiff := current - start blockdiff := current - start
periods := (blockdiff / frequency) + 1 periods := (blockdiff / frequency) + 1

View file

@ -17,7 +17,6 @@ import (
"github.com/ethereum/go-ethereum/contracts/ens" "github.com/ethereum/go-ethereum/contracts/ens"
"github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/rpc" "github.com/ethereum/go-ethereum/rpc"
) )
@ -26,10 +25,6 @@ var (
cleanF func() cleanF func()
) )
func init() {
log.Root().SetHandler(log.CallerFileHandler(log.LvlFilterHandler(log.LvlTrace, log.StreamHandler(os.Stderr, log.TerminalFormat(true)))))
}
type FakeRPC struct { type FakeRPC struct {
blockcount *uint64 blockcount *uint64
} }
@ -109,7 +104,7 @@ func TestResourceHandler(t *testing.T) {
// check that the new resource is stored correctly // check that the new resource is stored correctly
namehash := ens.EnsNode(resourcevalidname) namehash := ens.EnsNode(resourcevalidname)
chunk, err := rh.ChunkStore.(*resourceChunkStore).localStore.(*LocalStore).memStore.Get(Key(namehash[:])) chunk, err := rh.ChunkStore.Get(Key(namehash[:]))
if err != nil { if err != nil {
teardownTest(t, err) teardownTest(t, err)
} else if len(chunk.SData) < 16 { } else if len(chunk.SData) < 16 {
@ -159,7 +154,10 @@ func TestResourceHandler(t *testing.T) {
// it will match on second iteration startblocknumber + (resourcefrequency * 3) // it will match on second iteration startblocknumber + (resourcefrequency * 3)
blockCount = startblocknumber + (resourcefrequency * 4) blockCount = startblocknumber + (resourcefrequency * 4)
rh2, err := NewResourceHandler(privkey, datadir, &testCloudStore{}, rh.ethapi) rh2, err := newTestResourceHandler(datadir, privkey, rh.ethapi)
if err != nil {
teardownTest(t, err)
}
_, err = rh2.LookupLatest(resourcename, true) _, err = rh2.LookupLatest(resourcename, true)
if err != nil { if err != nil {
teardownTest(t, err) teardownTest(t, err)
@ -273,7 +271,8 @@ func setupTest() (rh *ResourceHandler, privkey *ecdsa.PrivateKey, datadir string
return return
} }
rh, err = NewResourceHandler(privkey, datadir, &testCloudStore{}, rpcclient) rh, err = newTestResourceHandler(datadir, privkey, rpcclient)
teardown = func(t *testing.T, err error) { teardown = func(t *testing.T, err error) {
cleanF() cleanF()
if err != nil { if err != nil {
@ -284,21 +283,18 @@ func setupTest() (rh *ResourceHandler, privkey *ecdsa.PrivateKey, datadir string
return return
} }
//func teardownTest(t *testing.T, errstr string) { func newTestResourceHandler(datadir string, privkey *ecdsa.PrivateKey, rpcclient *rpc.Client) (*ResourceHandler, error) {
// cleanF() path := filepath.Join(datadir, "resource")
// if errstr != "" { basekey := make([]byte, 32)
// t.Fatal(errstr) hasher := MakeHashFunc("SHA3")
// } dbStore, err := NewDbStore(path, hasher, singletonSwarmDbCapacity, func(k Key) (ret uint8) { return uint8(Proximity(basekey[:], k[:])) })
//} if err != nil {
return nil, err
type testCloudStore struct { }
localStore := &LocalStore{
memStore: NewMemStore(dbStore, singletonSwarmDbCapacity),
DbStore: dbStore,
} }
func (c *testCloudStore) Store(*Chunk) { return NewResourceHandler(privkey, hasher, localStore, rpcclient)
}
func (c *testCloudStore) Deliver(*Chunk) {
}
func (c *testCloudStore) Retrieve(*Chunk) {
} }