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

@@ -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 {
+		if err == nil {
-			log.Error("not in db? ", "key", req.Key, "chunk", chunk)
 			continue R
 		}
-		if chunk.ReqC == nil {
+		if err != storage.ErrFetching {
-			continue R
+			panic(fmt.Sprintf("not in db? key %v chunk %v", req.Key, chunk))
 		}
 		select {
 		case <-chunk.ReqC:

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

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
+			// preserve ErrNotFound error
-			err = notFound
+			err = ErrNotFound
 		}
 		return data, err
 	}

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 {
+	if err != ErrNotFound {
-		t.Errorf("Expected notFound, got %v", err)
+		t.Errorf("Expected ErrNotFound, got %v", err)
 	}
 }
 

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
 )

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))
-			log.Trace(fmt.Sprintf("LocalStore.GetOrRetrieve: %v found locally", key))
+		return chunk, false
-		} else {
+	}
-			log.Trace(fmt.Sprintf("LocalStore.GetOrRetrieve: %v hit on an existing request", key))
+	if err == ErrFetching {
-			// no need to launch again
+		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

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

swarm/storage/memstore_test.go

@@ -63,8 +63,8 @@ func TestMemStoreNotFound(t *testing.T) {
 	defer m.Close()
 
 	_, err := m.Get(ZeroKey)
-	if err != notFound {
+	if err != ErrNotFound {
-		t.Errorf("Expected notFound, got %v", err)
+		t.Errorf("Expected ErrNotFound, got %v", err)
 	}
 }
 

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
 }