From 295512f5a83923a5e3d316cda898e158e1385a44 Mon Sep 17 00:00:00 2001 From: Balint Gabor Date: Mon, 22 Jan 2018 15:23:41 +0100 Subject: [PATCH] swarm/storage, swarm/network: Fix race condition There was a race condition in writing/reading chunk.SData between delivery.processReceivedChunks and dpa.retrieveWorker when the chunk was still fetching --- swarm/network/stream/delivery.go | 8 ++++---- swarm/storage/common_test.go | 12 ++++++++++++ swarm/storage/dbstore.go | 6 +++--- swarm/storage/dbstore_test.go | 4 ++-- swarm/storage/dpa.go | 4 ++-- swarm/storage/localstore.go | 19 +++++++++++++------ swarm/storage/memstore.go | 4 ++-- swarm/storage/memstore_test.go | 4 ++-- swarm/storage/netstore.go | 6 ++---- 9 files changed, 42 insertions(+), 25 deletions(-) diff --git a/swarm/network/stream/delivery.go b/swarm/network/stream/delivery.go index 67181c3f22..4355997fd0 100644 --- a/swarm/network/stream/delivery.go +++ b/swarm/network/stream/delivery.go @@ -18,6 +18,7 @@ package stream import ( "errors" + "fmt" "time" "github.com/ethereum/go-ethereum/log" @@ -167,12 +168,11 @@ R: for req := range d.receiveC { // this should be has locally chunk, err := d.db.Get(req.Key) - if err != nil { - log.Error("not in db? ", "key", req.Key, "chunk", chunk) + if err == nil { continue R } - if chunk.ReqC == nil { - continue R + if err != storage.ErrFetching { + panic(fmt.Sprintf("not in db? key %v chunk %v", req.Key, chunk)) } select { case <-chunk.ReqC: diff --git a/swarm/storage/common_test.go b/swarm/storage/common_test.go index 700ba8dac0..21f8e5bbc1 100644 --- a/swarm/storage/common_test.go +++ b/swarm/storage/common_test.go @@ -20,16 +20,28 @@ import ( "bytes" "crypto/rand" "encoding/binary" + "flag" "fmt" "hash" "io" + "os" "sync" "testing" "time" "github.com/ethereum/go-ethereum/crypto/sha3" + "github.com/ethereum/go-ethereum/log" ) +var ( + loglevel = flag.Int("loglevel", 2, "verbosity of logs") +) + +func init() { + flag.Parse() + log.Root().SetHandler(log.LvlFilterHandler(log.Lvl(*loglevel), log.StreamHandler(os.Stderr, log.TerminalFormat(false)))) +} + type brokenLimitedReader struct { lr io.Reader errAt int diff --git a/swarm/storage/dbstore.go b/swarm/storage/dbstore.go index 887c2e08de..0facd67c04 100644 --- a/swarm/storage/dbstore.go +++ b/swarm/storage/dbstore.go @@ -699,7 +699,7 @@ func (s *DbStore) get(key Key) (chunk *Chunk, err error) { decodeData(data, chunk) } else { - err = notFound + err = ErrNotFound } return @@ -712,8 +712,8 @@ func newMockGetDataFunc(mockStore *mock.NodeStore) func(key Key) (data []byte, e return func(key Key) (data []byte, err error) { data, err = mockStore.Get(key) if err == mock.ErrNotFound { - // preserve notFound error - err = notFound + // preserve ErrNotFound error + err = ErrNotFound } return data, err } diff --git a/swarm/storage/dbstore_test.go b/swarm/storage/dbstore_test.go index 7f751594e5..6b86ed518e 100644 --- a/swarm/storage/dbstore_test.go +++ b/swarm/storage/dbstore_test.go @@ -142,8 +142,8 @@ func testDbStoreNotFound(t *testing.T, mock bool) { defer db.close() _, err = db.Get(ZeroKey) - if err != notFound { - t.Errorf("Expected notFound, got %v", err) + if err != ErrNotFound { + t.Errorf("Expected ErrNotFound, got %v", err) } } diff --git a/swarm/storage/dpa.go b/swarm/storage/dpa.go index b0aafe0343..7633b61f0b 100644 --- a/swarm/storage/dpa.go +++ b/swarm/storage/dpa.go @@ -48,8 +48,8 @@ const ( ) var ( - notFound = errors.New("not found") - + ErrNotFound = errors.New("not found") + ErrFetching = errors.New("chunk still fetching") // timeout interval before retrieval is timed out searchTimeout = 3 * time.Second ) diff --git a/swarm/storage/localstore.go b/swarm/storage/localstore.go index 893670b232..ac6d642950 100644 --- a/swarm/storage/localstore.go +++ b/swarm/storage/localstore.go @@ -106,7 +106,15 @@ func (self *LocalStore) Put(chunk *Chunk) { // ChunkStores are remote and can have long latency func (self *LocalStore) Get(key Key) (chunk *Chunk, err error) { chunk, err = self.memStore.Get(key) + if err == nil { + if chunk.ReqC != nil { + select { + case <-chunk.ReqC: + default: + return chunk, ErrFetching + } + } return } chunk, err = self.DbStore.Get(key) @@ -123,12 +131,11 @@ func (self *LocalStore) GetOrCreateRequest(key Key) (chunk *Chunk, created bool) var err error chunk, err = self.Get(key) if err == nil { - if chunk.ReqC == nil { - log.Trace(fmt.Sprintf("LocalStore.GetOrRetrieve: %v found locally", key)) - } else { - log.Trace(fmt.Sprintf("LocalStore.GetOrRetrieve: %v hit on an existing request", key)) - // no need to launch again - } + log.Trace(fmt.Sprintf("LocalStore.GetOrRetrieve: %v found locally", key)) + return chunk, false + } + if err == ErrFetching { + log.Trace(fmt.Sprintf("LocalStore.GetOrRetrieve: %v hit on an existing request %v", key, chunk.ReqC)) return chunk, false } // no data and no request status diff --git a/swarm/storage/memstore.go b/swarm/storage/memstore.go index 7eb0aa06c2..65affc3ffe 100644 --- a/swarm/storage/memstore.go +++ b/swarm/storage/memstore.go @@ -214,7 +214,7 @@ func (s *MemStore) Get(hash Key) (chunk *Chunk, err error) { l := hash.bits(bitpos, node.bits) st := node.subtree[l] if st == nil { - return nil, notFound + return nil, ErrNotFound } bitpos += node.bits node = st @@ -232,7 +232,7 @@ func (s *MemStore) Get(hash Key) (chunk *Chunk, err error) { } } } else { - err = notFound + err = ErrNotFound } return diff --git a/swarm/storage/memstore_test.go b/swarm/storage/memstore_test.go index 6b4bc0da56..edf87917e8 100644 --- a/swarm/storage/memstore_test.go +++ b/swarm/storage/memstore_test.go @@ -63,8 +63,8 @@ func TestMemStoreNotFound(t *testing.T) { defer m.Close() _, err := m.Get(ZeroKey) - if err != notFound { - t.Errorf("Expected notFound, got %v", err) + if err != ErrNotFound { + t.Errorf("Expected ErrNotFound, got %v", err) } } diff --git a/swarm/storage/netstore.go b/swarm/storage/netstore.go index 334cf3635a..265baa5337 100644 --- a/swarm/storage/netstore.go +++ b/swarm/storage/netstore.go @@ -17,7 +17,6 @@ package storage import ( - "encoding/binary" "time" ) @@ -40,7 +39,7 @@ func (self *NetStore) Get(key Key) (chunk *Chunk, err error) { var created bool chunk, created = self.localStore.GetOrCreateRequest(key) if chunk.ReqC == nil { - return + return chunk, nil } if created { @@ -53,10 +52,9 @@ func (self *NetStore) Get(key Key) (chunk *Chunk, err error) { select { case <-t.C: - return nil, notFound + return nil, ErrNotFound case <-chunk.ReqC: } - chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8])) return chunk, nil }