cmd/swarm, swarm: merge important parts from swarm-gateways-db-fixes

cmd/swarm/db.go

@@ -23,6 +23,7 @@ import (
 	"path/filepath"
 
 	"github.com/ethereum/go-ethereum/cmd/utils"
+	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/swarm/storage"
 	"gopkg.in/urfave/cli.v1"
@@ -30,11 +31,11 @@ import (
 
 func dbExport(ctx *cli.Context) {
 	args := ctx.Args()
-	if len(args) != 2 {
+	if len(args) != 3 {
-		utils.Fatalf("invalid arguments, please specify both <chunkdb> (path to a local chunk database) and <file> (path to write the tar archive to, - for stdout)")
+		utils.Fatalf("invalid arguments, please specify both <chunkdb> (path to a local chunk database), <file> (path to write the tar archive to, - for stdout) and the base key")
 	}
 
-	store, err := openDbStore(args[0])
+	store, err := openDbStore(args[0], common.Hex2Bytes(args[2]))
 	if err != nil {
 		utils.Fatalf("error opening local chunk database: %s", err)
 	}
@@ -62,11 +63,11 @@ func dbExport(ctx *cli.Context) {
 
 func dbImport(ctx *cli.Context) {
 	args := ctx.Args()
-	if len(args) != 2 {
+	if len(args) != 3 {
-		utils.Fatalf("invalid arguments, please specify both <chunkdb> (path to a local chunk database) and <file> (path to read the tar archive from, - for stdin)")
+		utils.Fatalf("invalid arguments, please specify both <chunkdb> (path to a local chunk database), <file> (path to read the tar archive from, - for stdin) and the base key")
 	}
 
-	store, err := openDbStore(args[0])
+	store, err := openDbStore(args[0], common.Hex2Bytes(args[2]))
 	if err != nil {
 		utils.Fatalf("error opening local chunk database: %s", err)
 	}
@@ -94,11 +95,11 @@ func dbImport(ctx *cli.Context) {
 
 func dbClean(ctx *cli.Context) {
 	args := ctx.Args()
-	if len(args) != 1 {
+	if len(args) != 2 {
-		utils.Fatalf("invalid arguments, please specify <chunkdb> (path to a local chunk database)")
+		utils.Fatalf("invalid arguments, please specify <chunkdb> (path to a local chunk database) and the base key")
 	}
 
-	store, err := openDbStore(args[0])
+	store, err := openDbStore(args[0], common.Hex2Bytes(args[1]))
 	if err != nil {
 		utils.Fatalf("error opening local chunk database: %s", err)
 	}
@@ -107,10 +108,10 @@ func dbClean(ctx *cli.Context) {
 	store.Cleanup()
 }
 
-func openDbStore(path string) (*storage.DbStore, error) {
+func openDbStore(path string, basekey []byte) (*storage.DbStore, error) {
 	if _, err := os.Stat(filepath.Join(path, "CURRENT")); err != nil {
 		return nil, fmt.Errorf("invalid chunkdb path: %s", err)
 	}
 	hash := storage.MakeHashFunc("SHA3")
-	return storage.NewDbStore(path, hash, 10000000, 0)
+	return storage.NewDbStore(path, hash, 10000000, func(k storage.Key) (ret uint8) { return uint8(storage.Proximity(basekey[:], k[:])) })
 }

cmd/swarm/hash.go

@@ -39,7 +39,7 @@ func hash(ctx *cli.Context) {
 
 	stat, _ := f.Stat()
 	chunker := storage.NewTreeChunker(storage.NewChunkerParams())
-	key, err := chunker.Split(f, stat.Size(), nil, nil, nil)
+	key, _, err := chunker.Split(f, stat.Size(), nil)
 	if err != nil {
 		utils.Fatalf("%v\n", err)
 	} else {

swarm/api/api.go

@@ -22,7 +22,6 @@ import (
 	"net/http"
 	"regexp"
 	"strings"
-	"sync"
 
 	"bytes"
 	"mime"
@@ -71,8 +70,8 @@ func (self *Api) Retrieve(key storage.Key) storage.LazySectionReader {
 	return self.dpa.Retrieve(key)
 }
 
-func (self *Api) Store(data io.Reader, size int64, wg *sync.WaitGroup) (key storage.Key, err error) {
+func (self *Api) Store(data io.Reader, size int64) (key storage.Key, wait func(), err error) {
-	return self.dpa.Store(data, size, wg, nil)
+	return self.dpa.Store(data, size)
 }
 
 type ErrResolve error
@@ -109,21 +108,22 @@ func (self *Api) Resolve(uri *URI) (storage.Key, error) {
 }
 
 // Put provides singleton manifest creation on top of dpa store
-func (self *Api) Put(content, contentType string) (storage.Key, error) {
+func (self *Api) Put(content, contentType string) (k storage.Key, wait func(), err error) {
 	r := strings.NewReader(content)
-	wg := &sync.WaitGroup{}
+	key, waitContent, err := self.dpa.Store(r, int64(len(content)))
-	key, err := self.dpa.Store(r, int64(len(content)), wg, nil)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 	manifest := fmt.Sprintf(`{"entries":[{"hash":"%v","contentType":"%s"}]}`, key, contentType)
 	r = strings.NewReader(manifest)
-	key, err = self.dpa.Store(r, int64(len(manifest)), wg, nil)
+	key, waitManifest, err := self.dpa.Store(r, int64(len(manifest)))
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
-	wg.Wait()
+	return key, func() {
-	return key, nil
+		waitContent()
+		waitManifest()
+	}, nil
 }
 
 // Get uses iterative manifest retrieval and prefix matching

swarm/api/api_test.go

@@ -110,7 +110,7 @@ func TestApiPut(t *testing.T) {
 		content := "hello"
 		exp := expResponse(content, "text/plain", 0)
 		// exp := expResponse([]byte(content), "text/plain", 0)
-		key, err := api.Put(content, exp.MimeType)
+		key, _, err := api.Put(content, exp.MimeType)
 		if err != nil {
 			t.Fatalf("unexpected error: %v", err)
 		}

swarm/api/filesystem.go

@@ -43,6 +43,7 @@ func NewFileSystem(api *Api) *FileSystem {
 
 // Upload replicates a local directory as a manifest file and uploads it
 // using dpa store
+// This function waits the chunks to be stored.
 // TODO: localpath should point to a manifest
 //
 // DEPRECATED: Use the HTTP API instead
@@ -112,12 +113,12 @@ func (self *FileSystem) Upload(lpath, index string) (string, error) {
 			if err == nil {
 				stat, _ := f.Stat()
 				var hash storage.Key
-				wg := &sync.WaitGroup{}
+				var wait func()
-				hash, err = self.api.dpa.Store(f, stat.Size(), wg, nil)
+				hash, wait, err = self.api.dpa.Store(f, stat.Size())
 				if hash != nil {
 					list[i].Hash = hash.String()
 				}
-				wg.Wait()
+				wait()
 				awg.Done()
 				if err == nil {
 					first512 := make([]byte, 512)

swarm/api/http/server.go

@@ -99,7 +99,7 @@ func (s *Server) HandlePostRaw(w http.ResponseWriter, r *Request) {
 		return
 	}
 
-	key, err := s.api.Store(r.Body, r.ContentLength, nil)
+	key, _, err := s.api.Store(r.Body, r.ContentLength)
 	if err != nil {
 		s.Error(w, r, err)
 		return

swarm/api/http/server_test.go

@@ -23,7 +23,6 @@ import (
 	"io/ioutil"
 	"net/http"
 	"strings"
-	"sync"
 	"testing"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -59,15 +58,14 @@ func TestBzzGetPath(t *testing.T) {
 	srv := testutil.NewTestSwarmServer(t)
 	defer srv.Close()
 
-	wg := &sync.WaitGroup{}
-
 	for i, mf := range testmanifest {
 		reader[i] = bytes.NewReader([]byte(mf))
-		key[i], err = srv.Dpa.Store(reader[i], int64(len(mf)), wg, nil)
+		var wait func()
+		key[i], wait, err = srv.Dpa.Store(reader[i], int64(len(mf)))
 		if err != nil {
 			t.Fatal(err)
 		}
-		wg.Wait()
+		wait()
 	}
 
 	_, err = http.Get(srv.URL + "/bzz-raw:/" + common.ToHex(key[0])[2:] + "/a")

swarm/api/manifest.go

@@ -24,7 +24,6 @@ import (
 	"io"
 	"net/http"
 	"strings"
-	"sync"
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -65,7 +64,8 @@ func (a *Api) NewManifest() (storage.Key, error) {
 	if err != nil {
 		return nil, err
 	}
-	return a.Store(bytes.NewReader(data), int64(len(data)), &sync.WaitGroup{})
+	key, _, err := a.Store(bytes.NewReader(data), int64(len(data)))
+	return key, err
 }
 
 // ManifestWriter is used to add and remove entries from an underlying manifest
@@ -85,7 +85,7 @@ func (a *Api) NewManifestWriter(key storage.Key, quitC chan bool) (*ManifestWrit
 
 // AddEntry stores the given data and adds the resulting key to the manifest
 func (m *ManifestWriter) AddEntry(data io.Reader, e *ManifestEntry) (storage.Key, error) {
-	key, err := m.api.Store(data, e.Size, nil)
+	key, _, err := m.api.Store(data, e.Size)
 	if err != nil {
 		return nil, err
 	}
@@ -351,9 +351,7 @@ func (self *manifestTrie) recalcAndStore() error {
 	}
 
 	sr := bytes.NewReader(manifest)
-	wg := &sync.WaitGroup{}
+	key, _, err2 := self.dpa.Store(sr, int64(len(manifest)))
-	key, err2 := self.dpa.Store(sr, int64(len(manifest)), wg, nil)
-	wg.Wait()
 	self.hash = key
 	return err2
 }

swarm/api/storage.go

@@ -42,7 +42,7 @@ func NewStorage(api *Api) *Storage {
 //
 // DEPRECATED: Use the HTTP API instead
 func (self *Storage) Put(content, contentType string) (string, error) {
-	key, err := self.api.Put(content, contentType)
+	key, _, err := self.api.Put(content, contentType)
 	if err != nil {
 		return "", err
 	}

swarm/network/requests.go

@@ -17,8 +17,6 @@
 package network
 
 import (
-	"bytes"
-	"encoding/binary"
 	"fmt"
 
 	"github.com/ethereum/go-ethereum/log"
@@ -54,19 +52,19 @@ testing chunk availability etc etc, we can indicate it by limiting the size here
 Request ID can be newly generated or kept from the request originator.
 
 */
-type retrieveRequestMsgData struct {
+type retrieveRequestMsg struct {
 	Key     storage.Key   // target Key address of chunk to be retrieved
 	Id      uint64        // request id, request is a lookup if missing or zero
 	MaxSize uint64        // maximum size of delivery accepted
-	from    Peer        //
+	from    *StreamerPeer //
 }
 
-func (self retrieveRequestMsgData) String() string {
+func (self retrieveRequestMsg) String() string {
 	var from string
 	if self.from == nil {
 		from = "ourselves"
 	} else {
-		from = fmt.Sprintf("%x", self.from.OverlayAddr())
+		from = fmt.Sprintf("%x", self.from.Over())
 	}
 	var target []byte
 	if len(self.Key) > 3 {
@@ -77,9 +75,9 @@ func (self retrieveRequestMsgData) String() string {
 
 // entrypoint for retrieve requests coming from the bzz wire protocol
 // checks swap balance - return if peer has no credit
-func (self *RequestHandler) handleRetrieveRequestMsg(msg interface{}, p Peer) error {
+func (self *StreamerPeer) handleRetrieveRequestMsg(msg interface{}) error {
-	req := msg.(*retrieveRequestMsgData)
+	req := msg.(*retrieveRequestMsg)
-	req.from = p
+	req.from = self
 	// TODO:
 	// swap - record credit for 1 request
 	// note that only charge actual reqsearches
@@ -90,15 +88,15 @@ func (self *RequestHandler) handleRetrieveRequestMsg(msg interface{}, p Peer) er
 	chunk, _ := self.netStore.Get(req.Key)
 	rs := chunk.Req
 	if rs != nil {
-		rs = storage.NewRequest()
+		rs = storage.NewRequestStatus(req.Key)
-		self.addRequester(rs, req)
+		addRequester(rs, req)
 		chunk.Req = rs
 	}
 
 	// check if we can immediately deliver
 	if chunk.SData != nil {
 		if req.MaxSize == 0 || int64(req.MaxSize) >= chunk.Size {
-			err := self.netStore.Deliver(chunk)
+			err := self.Deliver(chunk, Top)
 			if err != nil {
 				log.Trace(fmt.Sprintf("%v - content found, delivery error: %v", req.Key.Log(), err))
 				return nil
@@ -118,7 +116,7 @@ adds a new peer to an existing open request
 only add if less than requesterCount peers forwarded the same request id so far
 note this is done irrespective of status (searching or found)
 */
-func (self *RequestHandler) addRequester(rs *storage.RequestStatus, req *retrieveRequestMsgData) {
+func addRequester(rs *storage.RequestStatus, req *retrieveRequestMsg) {
 	log.Trace(fmt.Sprintf("Depo.addRequester: key %v - add peer to req.Id %v", req.Key.Log(), req.Id))
 	list := rs.Requesters[req.Id]
 	rs.Requesters[req.Id] = append(list, req)
@@ -128,19 +126,20 @@ func (self *RequestHandler) addRequester(rs *storage.RequestStatus, req *retriev
  store requests are put in netstore so they are stored and then
  forwarded to the peers in their kademlia proximity bin by the syncer
 */
-type storeRequestMsgData struct {
+type storeRequestMsg struct {
+	Key   storage.Key
 	SData []byte // the stored chunk Data (incl size)
 	// optional
 	Id   uint64 // request ID. if delivery, the ID is retrieve request ID
 	from Peer   // [not serialised] protocol registers the requester
 }
 
-func (self storeRequestMsgData) String() string {
+func (self storeRequestMsg) String() string {
 	var from string
 	if self.from == nil {
 		from = "self"
 	} else {
-		from = fmt.Sprintf("%x", self.from.OverlayAddr())
+		from = fmt.Sprintf("%x", self.from.Over())
 	}
 	end := len(self.SData)
 	if len(self.SData) > 10 {
@@ -153,12 +152,11 @@ func (self storeRequestMsgData) String() string {
 // if key found locally, return. otherwise
 // remote is untrusted, so hash is verified and chunk passed on to NetStore
 func (self *RequestHandler) handleStoreRequestMsg(msg interface{}, p Peer) error {
-	req := msg.(*storeRequestMsgData)
+	req := msg.(*storeRequestMsg)
 	req.from = p
-	chunk, err := storage.NewChunkFromData(req.SData)
+	// TODO: chunk validation
-	if err != nil {
+	chunk := storage.NewChunk(req.Key, nil)
-		return err
+	chunk.SData = req.SData
-	}
 	chunk.Source = p
 	self.netStore.Put(chunk)
 	log.Trace(fmt.Sprintf("delivery of %v from %v", chunk, p))

swarm/network/streamer.go

@@ -18,13 +18,15 @@ package network
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"sync"
 
-	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/p2p/discover"
 	"github.com/ethereum/go-ethereum/p2p/protocols"
 	bv "github.com/ethereum/go-ethereum/swarm/network/bitvector"
 	pq "github.com/ethereum/go-ethereum/swarm/network/priorityqueue"
+	"github.com/ethereum/go-ethereum/swarm/storage"
 )
 
 const (
@@ -45,7 +47,7 @@ type Stream string
 type Handover struct {
 	Stream     Stream // name of stream
 	Start, End uint64 // index of hashes
-	Root       common.Hash // Root hash for indexed segment inclusion proofs
+	Root       []byte // Root hash for indexed segment inclusion proofs
 }
 
 // HandoverProof represents a signed statement that the upstream peer handed over the stream section
@@ -70,7 +72,7 @@ type TakeoverProofMsg TakeoverProof
 
 // String pretty prints TakeoverProofMsg
 func (self TakeoverProofMsg) String() string {
-	return fmt.Sprintf("Stream: '%v' [%v-%v], Root: %x, Sig: %x", self.Stream, self.From, self.To, self.Root, self.Sig)
+	return fmt.Sprintf("Stream: '%v' [%v-%v], Root: %x, Sig: %x", self.Stream, self.Start, self.End, self.Root, self.Sig)
 }
 
 // SubcribeMsg is the protocol msg for requesting a stream(section)
@@ -138,37 +140,46 @@ func (self *Streamer) RegisterOutgoingStreamer(stream Stream, f func(*StreamerPe
 }
 
 // GetIncomingStreamer accessor for incoming streamer constructors
-func (self *Streamer) GetIncomingStreamer(stream Stream) func(*StreamerPeer) (IncomingStreamer, error) {
+func (self *Streamer) GetIncomingStreamer(stream Stream) (func(*StreamerPeer) (IncomingStreamer, error), error) {
 	self.incomingLock.RLock()
 	defer self.incomingLock.RUnlock()
 	f := self.incoming[stream]
 	if f == nil {
-		return nil, fmt.Errorf("stream %v not registered", s)
+		return nil, fmt.Errorf("stream %v not registered", stream)
 	}
 	return f, nil
 }
 
 // GetOutgoingStreamer accessor for incoming streamer constructors
-func (self *Streamer) GetOutgoingStreamer(stream Stream) func(*StreamerPeer) (OutgoingStreamer, error) {
+func (self *Streamer) GetOutgoingStreamer(stream Stream) (func(*StreamerPeer) (OutgoingStreamer, error), error) {
 	self.outgoingLock.RLock()
 	defer self.outgoingLock.RUnlock()
 	f := self.outgoing[stream]
 	if f == nil {
-		return nil, fmt.Errorf("stream %v not registered", s)
+		return nil, fmt.Errorf("stream %v not registered", stream)
 	}
+	return f, nil
+}
+
+func (self *Streamer) NodeInfo() interface{} {
+	return nil
+}
+
+func (self *Streamer) PeerInfo(id discover.NodeID) interface{} {
+	return nil
 }
 
 // OutgoingStreamer interface for outgoing peer Streamer
 type OutgoingStreamer interface {
 	CurrentBatch() []byte
-	SetNextBatch(uint64, uint64) ([]byte, uint64, uint64, *HandoverProof)
+	SetNextBatch(uint64, uint64) ([]byte, uint64, uint64, *HandoverProof, error)
 	GetData([]byte) []byte
 	Priority() int
 }
 
 // IncomingStreamer interface for incoming peer Streamer
 type IncomingStreamer interface {
-	NextBatch(uint64, uint64) (uint64, uint64)
+	NextBatch(uint64) (uint64, uint64)
 	NeedData([]byte) func()
 	Priority() int
 }
@@ -178,6 +189,7 @@ type StreamerPeer struct {
 	Peer
 	streamer     *Streamer
 	pq           *pq.PriorityQueue
+	netStore     storage.ChunkStore
 	outgoingLock sync.RWMutex
 	incomingLock sync.RWMutex
 	outgoing     map[Stream]OutgoingStreamer
@@ -249,7 +261,10 @@ func (self *StreamerPeer) Subscribe(s Stream, from, to uint64) error {
 	if err != nil {
 		return err
 	}
-	is := f(self)
+	is, err := f(self)
+	if err != nil {
+		return err
+	}
 	self.setIncomingStreamer(s, is)
 	msg := &SubscribeMsg{
 		Stream:   s,
@@ -257,20 +272,24 @@ func (self *StreamerPeer) Subscribe(s Stream, from, to uint64) error {
 		To:       to,
 		Priority: uint8(is.Priority()),
 	}
-	self.Send(msg, is.Priority())
+	self.SendPriority(msg, is.Priority())
+	return nil
 }
 
 func (self *StreamerPeer) handleSubscribeMsg(msg interface{}) error {
 	req := msg.(*SubscribeMsg)
-	f, err := self.streamer.getOutgoingStreamer(req.Stream)
+	f, err := self.streamer.GetOutgoingStreamer(req.Stream)
+	if err != nil {
+		return err
+	}
+	s, err := f(self)
 	if err != nil {
 		return err
 	}
-	s := f(self)
 	if err := self.setOutgoingStreamer(req.Stream, s); err != nil {
 		return nil
 	}
-	self.UnsyncedKeys(s, req.From, req.To)
+	self.SendUnsyncedKeys(s, req.From, req.To, int(req.Priority))
 	return nil
 }
 
@@ -278,13 +297,15 @@ func (self *StreamerPeer) handleSubscribeMsg(msg interface{}) error {
 // Filter method
 func (self *StreamerPeer) handleUnsyncedKeysMsg(msg interface{}) error {
 	req := msg.(*UnsyncedKeysMsg)
-	req.C = make(chan struct{})
 	s, err := self.getIncomingStreamer(req.Stream)
 	if err != nil {
 		return err
 	}
 	hashes := req.Hashes
-	want := bv.New(len(hashes) / HashSize)
+	want, err := bv.New(len(hashes) / HashSize)
+	if err != nil {
+		return err
+	}
 	wg := sync.WaitGroup{}
 	for i := 0; i < len(hashes)/HashSize; i += HashSize {
 		hash := hashes[i : i+HashSize]
@@ -298,24 +319,24 @@ func (self *StreamerPeer) handleUnsyncedKeysMsg(msg interface{}) error {
 			}(wait)
 		}
 	}
-	go func() {
+	// go func() {
-		wg.Wait()
+	// 	wg.Wait()
-		msg := s.TakeoverProof(req.Stream, req.From, req.Hashes, req.Root)
+	// 	msg := s.TakeoverProof(req.Stream, req.From, req.Hashes, req.Root)
-		self.Send(msg, s.Priority())
+	// 	self.Send(msg, s.Priority())
-	}()
+	// }()
 	// only send wantedKeysMsg if all missing chunks of the previous batch arrived
 	// except
-	from, to = s.NextBatch(from, to)
+	from, to := s.NextBatch(req.To)
 	if from == to {
 		return nil
 	}
 	msg = &WantedKeysMsg{
 		Stream: req.Stream,
-		Want:   want,
+		Want:   want.Bytes(),
 		From:   from,
 		To:     to,
 	}
-	self.Send(msg, s.Priority())
+	self.SendPriority(msg, s.Priority())
 	return nil
 }
 
@@ -330,17 +351,22 @@ func (self *StreamerPeer) handleWantedKeysMsg(msg interface{}) error {
 	}
 	hashes := s.CurrentBatch()
 	// launch in go routine since GetBatch blocks until new hashes arrive
-	go self.UnsyncedKeys(s, req.From, req.To)
+	go self.SendUnsyncedKeys(s, req.From, req.To, s.Priority())
 	l := len(hashes) / HashSize
-	want := bv.NewFromBytes(req.Want, l)
+	want, err := bv.NewFromBytes(req.Want, l)
+	if err != nil {
+		return err
+	}
 	for i := 0; i < l; i++ {
 		if want.Get(i) {
 			hash := hashes[i*HashSize : (i+1)*HashSize]
 			data := s.GetData(hash)
 			if data == nil {
-				return errNotFound
+				return errors.New("not found")
 			}
-			if err := self.Deliver(data, s.Priority()); err != nil {
+			chunk := storage.NewChunk(hash, nil)
+			chunk.SData = data
+			if err := self.Deliver(chunk, s.Priority()); err != nil {
 				return err
 			}
 		}
@@ -350,7 +376,7 @@ func (self *StreamerPeer) handleWantedKeysMsg(msg interface{}) error {
 
 func (self *StreamerPeer) handleTakeoverProofMsg(msg interface{}) error {
 	req := msg.(*TakeoverProofMsg)
-	s, err := self.getOutgoingStreamer(req.Stream)
+	_, err := self.getOutgoingStreamer(req.Stream)
 	if err != nil {
 		return err
 	}
@@ -359,32 +385,36 @@ func (self *StreamerPeer) handleTakeoverProofMsg(msg interface{}) error {
 }
 
 // Deliver sends a storeRequestMsg protocol message to the peer
-func (self *StreamerPeer) Deliver(data []byte, priority int) error {
+func (self *StreamerPeer) Deliver(chunk *storage.Chunk, priority int) error {
 	msg := &storeRequestMsg{
-		SData: data,
+		Key:   chunk.Key,
+		SData: chunk.SData,
 	}
 	return self.pq.Push(nil, msg, priority)
 }
 
 // Deliver sends a storeRequestMsg protocol message to the peer
-func (self *StreamerPeer) Send(msg interface{}, priority int) error {
+func (self *StreamerPeer) SendPriority(msg interface{}, priority int) error {
 	return self.pq.Push(nil, msg, priority)
 }
 
 // UnsyncedKeys sends UnsyncedKeysMsg protocol msg
-func (self *StreamerPeer) SendUnsyncedKeys(s OutgoingStreamer, f, t uint64, priority int) {
+func (self *StreamerPeer) SendUnsyncedKeys(s OutgoingStreamer, f, t uint64, priority int) error {
-	hashes, from, to, proof := s.SetNextBatch(f, t)
+	hashes, from, to, proof, err := s.SetNextBatch(f, t)
+	if err != nil {
+		return err
+	}
 	msg := &UnsyncedKeysMsg{
 		HandoverProof: proof,
 		Hashes:        hashes,
 		From:          from,
 		To:            to,
 	}
-	self.Send(msg, s.Priority())
+	return self.SendPriority(msg, s.Priority())
 }
 
-// BzzSpec is the spec of the generic swarm handshake
+// StreamerSpec is the spec of the streamer protocol.
-var StrSpec = &protocols.Spec{
+var StreamerSpec = &protocols.Spec{
 	Name:       "stream",
 	Version:    1,
 	MaxMsgSize: 10 * 1024 * 1024,

swarm/network/syncer.go

@@ -18,6 +18,7 @@ package network
 
 import (
 	"bytes"
+	"errors"
 	"fmt"
 	"io"
 
@@ -45,8 +46,8 @@ func (self *DbAccess) get(key storage.Key) (*storage.Chunk, error) {
 }
 
 // current storage counter of chunk db
-func (self *DbAccess) currentStorageIndex(po int) uint64 {
+func (self *DbAccess) currentBucketStorageIndex(po uint8) uint64 {
-	return self.db.CurrentStorageIndex(po)
+	return self.db.CurrentBucketStorageIndex(po)
 }
 
 // iteration storage counter and proximity order
@@ -59,7 +60,7 @@ func (self *DbAccess) iterator(from uint64, to uint64, po uint8, f func(storage.
 // * live request delivery with or without checkback
 // * (live/non-live historical) chunk syncing per proximity bin
 type OutgoingSwarmSyncer struct {
-	po           int
+	po           uint8
 	db           *DbAccess
 	sessionAt    uint64
 	currentBatch []byte
@@ -67,35 +68,37 @@ type OutgoingSwarmSyncer struct {
 }
 
 // NewOutgoingSwarmSyncer is contructor for OutgoingSwarmSyncer
-func NewOutgoingSwarmSyncer(po int, db *DbAccess) *OutgoingSwarmSyncer {
+func NewOutgoingSwarmSyncer(po uint8, db *DbAccess) (*OutgoingSwarmSyncer, error) {
 	self := &OutgoingSwarmSyncer{
 		po:        po,
 		db:        db,
-		sessionAt: db.currentStorageIndex(po),
+		sessionAt: db.currentBucketStorageIndex(po),
 	}
-	return self
+	return self, nil
 }
 
+const maxPO = 32
+
 func RegisterOutgoingSyncers(streamer *Streamer, db *DbAccess) {
-	for po := 0; po < maxPO; po++ {
+	for po := uint8(0); po < maxPO; po++ {
-		stream := fmt.Sprintf("SYNC-%02d-live", po)
+		stream := Stream(fmt.Sprintf("SYNC-%02d-live", po))
 		streamer.RegisterOutgoingStreamer(stream, func(p *StreamerPeer) (OutgoingStreamer, error) {
 			return NewOutgoingSwarmSyncer(po, db)
 		})
-		stream = fmt.Sprintf("SYNC-%02d-history", po)
+		stream = Stream(fmt.Sprintf("SYNC-%02d-history", po))
 		streamer.RegisterOutgoingStreamer(stream, func(p *StreamerPeer) (OutgoingStreamer, error) {
 			return NewOutgoingSwarmSyncer(po, db)
 		})
-		stream = fmt.Sprintf("SYNC-%02d-delete", po)
+		// stream = Stream(fmt.Sprintf("SYNC-%02d-delete", po))
-		streamer.RegisterOutgoingStreamer(stream, func(p *StreamerPeer) (OutgoingStreamer, error) {
+		// streamer.RegisterOutgoingStreamer(stream, func(p *StreamerPeer) (OutgoingStreamer, error) {
-			return NewOutgoingProvableSwarmSyncer(po, db)
+		// 	return NewOutgoingProvableSwarmSyncer(po, db)
-		})
+		// })
 	}
 }
 
 // GetSection retrieves the actual chunk from localstore
 func (self *OutgoingSwarmSyncer) GetData(key []byte) []byte {
-	chunk, err := self.db.get(Key(key))
+	chunk, err := self.db.get(storage.Key(key))
 	if err != nil {
 		return nil
 	}
@@ -111,89 +114,100 @@ func (self *OutgoingSwarmSyncer) Priority() int {
 }
 
 // GetBatch retrieves the next batch of hashes from the dbstore
-func (self *OutgoingSwarmSyncer) SetNextBatch(from, to uint64) []byte {
+func (self *OutgoingSwarmSyncer) SetNextBatch(from, to uint64) ([]byte, uint64, uint64, *HandoverProof, error) {
 	var batch []byte
 	i := 0
 	err := self.db.iterator(from, to, self.po, func(key storage.Key, idx uint64) bool {
-		batch = append(batch, key[:])
+		batch = append(batch, key[:]...)
 		i++
 		to = idx
 		return i < batchSize
 	})
+	if err != nil {
+		return nil, 0, 0, nil, err
+	}
 	self.currentBatch = batch
 	log.Debug("Swarm batch", "po", self.po, "len", i, "from", from, "to", to)
-	return batch, from, to, proof
+	return batch, from, to, nil, nil
 }
 
 // IncomingSwarmSyncer
 type IncomingSwarmSyncer struct {
-	po            int
+	po            uint8
 	priority      int
 	sessionAt     uint64
 	nextC         chan struct{}
 	intervals     []uint64
 	sessionRoot   storage.Key
 	sessionReader storage.LazySectionReader
-	retrieveC     chan storage.Chunk
+	retrieveC     chan *storage.Chunk
-	storeC        chan storage.Chunk
+	storeC        chan *storage.Chunk
+	store         storage.ChunkStore
+	chunker       storage.Chunker
+	currentRoot   storage.Key
+	end, start    uint64
 }
 
 // NewIncomingSwarmSyncer is a contructor for provable data exchange syncer
-func NewIncomingSwarmSyncer(po int, priority int, sessionAt uint64, intervals []uint64, p Peer) *IncomingSwarmSyncer {
+func NewIncomingSwarmSyncer(po uint8, priority int, intervals []uint64, p Peer, store storage.ChunkStore, chunker storage.Chunker) (*IncomingSwarmSyncer, error) {
 	self := &IncomingSwarmSyncer{
 		po:        po,
 		priority:  priority,
-		sessionAt: sessionAt,
-		nextC:     make(chan struct{}, 1),
 		intervals: intervals,
+		store:     store,
+		chunker:   chunker,
 	}
-	return self
+	return self, nil
 }
 
-// NewIncomingProvableSwarmSyncer is a contructor for provable data exchange syncer
+func (s *IncomingSwarmSyncer) Priority() int {
-func NewIncomingProvableSwarmSyncer(po int, priority int, index uint64, sessionAt uint64, intervals []uint64, sessionRoot storage.Key, chunker *storage.PyramidChunker, store storage.ChunkStore, p Peer) *IncomingSwarmSyncer {
+	return s.priority
-	retrieveC := make(storage.Chunk, chunksCap)
-	RunChunkRequestor(p, retrieveC)
-	storeC := make(storage.Chunk, chunksCap)
-	RunChunkStorer(store, storeC)
-	self := &IncomingSwarmSyncer{
-		po:            po,
-		priority:      priority,
-		sessionAt:     sessionAt,
-		start:         index,
-		end:           index,
-		nextC:         make(chan struct{}, 1),
-		intervals:     intervals,
-		sessionRoot:   sessionRoot,
-		sessionReader: chunker.Join(sessionRoot, retrieveC),
-		retrieveC:     retrieveC,
-		storeC:        storeC,
-	}
-	return self
 }
 
+// // NewIncomingProvableSwarmSyncer is a contructor for provable data exchange syncer
+// func NewIncomingProvableSwarmSyncer(po int, priority int, index uint64, sessionAt uint64, intervals []uint64, sessionRoot storage.Key, chunker *storage.PyramidChunker, store storage.ChunkStore, p Peer) *IncomingSwarmSyncer {
+// 	retrieveC := make(storage.Chunk, chunksCap)
+// 	RunChunkRequestor(p, retrieveC)
+// 	storeC := make(storage.Chunk, chunksCap)
+// 	RunChunkStorer(store, storeC)
+// 	self := &IncomingSwarmSyncer{
+// 		po:            po,
+// 		priority:      priority,
+// 		sessionAt:     sessionAt,
+// 		start:         index,
+// 		end:           index,
+// 		nextC:         make(chan struct{}, 1),
+// 		intervals:     intervals,
+// 		sessionRoot:   sessionRoot,
+// 		sessionReader: chunker.Join(sessionRoot, retrieveC),
+// 		retrieveC:     retrieveC,
+// 		storeC:        storeC,
+// 	}
+// 	return self
+// }
+
 func RegisterIncomingSyncers(streamer *Streamer, db *DbAccess) {
-	for po := 0; po < maxPO; po++ {
+	for po := uint8(0); po < maxPO; po++ {
-		stream := fmt.Sprintf("SYNC-%02d-live", po)
+		stream := Stream(fmt.Sprintf("SYNC-%02d-live", po))
-		streamer.RegisterIncomingStreamer(stream, func(p *StreamerPeer) (OutgoingStreamer, error) {
+		streamer.RegisterIncomingStreamer(stream, func(p *StreamerPeer) (IncomingStreamer, error) {
-			return NewIncomingSwarmSyncer(po, Mid, sessionAt, nil, p)
+			return NewIncomingSwarmSyncer(po, High, nil, p, nil, nil)
 		})
-		stream = fmt.Sprintf("SYNC-%02d-history", po)
+		stream = Stream(fmt.Sprintf("SYNC-%02d-history", po))
-		streamer.RegisterIncomingStreamer(stream, func(p *StreamerPeer) (OutgoingStreamer, error) {
+		streamer.RegisterIncomingStreamer(stream, func(p *StreamerPeer) (IncomingStreamer, error) {
-			intervals := loadIntervals(p, po, false)
+			//intervals := loadIntervals(p, po, false)
-			return NewIncomingSwarmSyncer(po, Mid, sessionAt, intervals, p)
+			return NewIncomingSwarmSyncer(po, Mid, nil, p, nil, nil)
-		})
-		stream = fmt.Sprintf("SYNC-%02d-delete", po)
-		streamer.RegisterIncomingStreamer(stream, func(p *StreamerPeer) (OutgoingStreamer, error) {
-			intervals := loadIntervals(p, po, true)
-			return NewIncomingSwarmSyncer(po, Mid, sessionAt, intervals, p)
 		})
+		// stream = fmt.Sprintf("SYNC-%02d-delete", po)
+		// streamer.RegisterIncomingStreamer(stream, func(p *StreamerPeer) (OutgoingStreamer, error) {
+		// 	intervals := loadIntervals(p, po, true)
+		// 	return NewIncomingSwarmSyncer(po, Mid, sessionAt, intervals, p)
+		// })
 	}
 }
 
 // NeedData
 func (self *IncomingSwarmSyncer) NeedData(key []byte) func() {
-	chunk, err := self.store.Get(hash)
+	chunk, err := self.store.Get(key)
 	if err == nil {
 		if chunk.SData == nil {
 			// send a request instead
@@ -201,57 +215,59 @@ func (self *IncomingSwarmSyncer) NeedData(key []byte) func() {
 		}
 	}
 	// create request and wait until the chunk data arrives and is stored
-	return func() {
+	return chunk.WaitToStore
-		storedC := <-chunk.storedC
-		<-storedC
-	}
 }
 
 // NextBatch adjusts the indexes by inspecting the intervals
-func (self *IncomingSwarmSyncer) NextBatch(from, to uint64) (uint64, uint64) {
+func (self *IncomingSwarmSyncer) NextBatch(from uint64) (nextFrom uint64, nextTo uint64) {
-	if from >= self.sessionAt { // live syncing
+	if self.intervals[0] >= self.sessionAt { // live syncing
-		self.intervals[1] = to
+		nextFrom = from
-	} else if to >= self.sessionAt { // history sync complete
+		self.intervals[1] = from
+	} else if from >= self.sessionAt { // history sync complete
 		self.intervals = nil
-		from = 0
+	} else if len(self.intervals) > 2 && from >= self.intervals[2] { // filled a gap in the intervals
-	} else if len(intervals) > 2 && to >= self.intervals[2] { // filled a gap in the intervals
+		self.intervals = append(self.intervals[:1], self.intervals[3:]...)
-		self.intervals[1:] = self.intervals[3:]
+		nextFrom = self.intervals[1]
-		from = self.intervals[1]
+		if len(self.intervals) > 2 {
-		if len(intervals) > 2 {
+			nextTo = self.intervals[2]
-			to = self.intervals[2]
+		} else {
+			nextTo = self.sessionAt
 		}
 	} else {
-		self.intervals[1] = to
+		nextFrom = from
+		self.intervals[1] = from
+		nextTo = self.sessionAt
 	}
-	return from, to
+	return nextFrom, nextTo
 }
 
 //
-func (self *IncomingSwarmSyncer) TakeoverProof(s Stream, from uint64, hashes []byte, root storage.Key) error {
+func (self *IncomingSwarmSyncer) TakeoverProof(s Stream, from uint64, hashes []byte, root storage.Key) (*TakeoverProof, error) {
 	// for provable syncer currentRoot is non-zero length
 	if self.chunker != nil {
 		if from > self.sessionAt { // for live syncing currentRoot is always updated
-			expRoot, err := self.chunker.Append(self.currentRoot, bytes.NewReader(hashes), self.retrieveC, self.storeC)
+			//expRoot, err := self.chunker.Append(self.currentRoot, bytes.NewReader(hashes), self.retrieveC, self.storeC)
+			expRoot, err := self.chunker.Append(self.currentRoot, bytes.NewReader(hashes), self.retrieveC)
 			if err != nil {
-				return err
+				return nil, err
 			}
 			if !bytes.Equal(root, expRoot) {
-				return fmt.Errorf("HandoverProof mismatch")
+				return nil, fmt.Errorf("HandoverProof mismatch")
 			}
-			self.currentRoot = currentRoot
+			self.currentRoot = root
 		} else {
 			expHashes := make([]byte, len(hashes))
-			n, err := self.sessionReader.ReadAt(expHashes, int64(self.end*HashSize))
+			_, err := self.sessionReader.ReadAt(expHashes, int64(self.end*HashSize))
 			if err != nil && err != io.EOF {
-				return err
+				return nil, err
 			}
 			if !bytes.Equal(expHashes, hashes) {
-				return errInvalidProof
+				return nil, errors.New("invalid proof")
 			}
 		}
-		return nil
+		return nil, nil
 	}
-	self.end += len(hashes) / HashSize
+	self.end += uint64(len(hashes)) / HashSize
 	takeover := &Takeover{
 		Stream: s,
 		Start:  self.start,
@@ -262,5 +278,5 @@ func (self *IncomingSwarmSyncer) TakeoverProof(s Stream, from uint64, hashes []b
 	return &TakeoverProof{
 		Takeover: takeover,
 		Sig:      nil,
-	}
+	}, nil
 }

swarm/pss/pss.go

@@ -777,10 +777,8 @@ func (self *Pss) checkFwdCache(addr []byte, digest pssDigest) bool {
 
 // DPA storage handler for message cache
 func (self *Pss) storeMsg(msg *PssMsg) (pssDigest, error) {
-	swg := &sync.WaitGroup{}
-	wwg := &sync.WaitGroup{}
 	buf := bytes.NewReader(msg.serialize())
-	key, err := self.dpa.Store(buf, int64(buf.Len()), swg, wwg)
+	key, _, err := self.dpa.Store(buf, int64(buf.Len()))
 	if err != nil {
 		log.Warn("Could not store in swarm", "err", err)
 		return pssDigest{}, err

swarm/storage/chunker.go

@@ -118,23 +118,19 @@ func (self *TreeChunker) decrementWorkerCount() {
 	self.workerCount -= 1
 }
 
-func (self *TreeChunker) Split(data io.Reader, size int64, chunkC chan *Chunk, swg, wwg *sync.WaitGroup) (Key, error) {
+func (self *TreeChunker) Split(data io.Reader, size int64, chunkC chan *Chunk) (k Key, wait func(), err error) {
 	if self.chunkSize <= 0 {
 		panic("chunker must be initialised")
 	}
 
 	jobC := make(chan *hashJob, 2*ChunkProcessors)
 	wg := &sync.WaitGroup{}
+	storeWg := &sync.WaitGroup{}
 	errC := make(chan error)
 	quitC := make(chan bool)
 
-	// wwg = workers waitgroup keeps track of hashworkers spawned by this split call
-	if wwg != nil {
-		wwg.Add(1)
-	}
-
 	self.incrementWorkerCount()
-	go self.hashWorker(jobC, chunkC, errC, quitC, swg, wwg)
+	go self.hashWorker(jobC, chunkC, errC, quitC, storeWg)
 
 	depth := 0
 	treeSize := self.chunkSize
@@ -149,16 +145,12 @@ func (self *TreeChunker) Split(data io.Reader, size int64, chunkC chan *Chunk, s
 	// this waitgroup member is released after the root hash is calculated
 	wg.Add(1)
 	//launch actual recursive function passing the waitgroups
-	go self.split(depth, treeSize/self.branches, key, data, size, jobC, chunkC, errC, quitC, wg, swg, wwg)
+	go self.split(depth, treeSize/self.branches, key, data, size, jobC, chunkC, errC, quitC, wg, storeWg)
 
 	// closes internal error channel if all subprocesses in the workgroup finished
 	go func() {
 		// waiting for all threads to finish
 		wg.Wait()
-		// if storage waitgroup is non-nil, we wait for storage to finish too
-		if swg != nil {
-			swg.Wait()
-		}
 		close(errC)
 	}()
 
@@ -166,16 +158,16 @@ func (self *TreeChunker) Split(data io.Reader, size int64, chunkC chan *Chunk, s
 	select {
 	case err := <-errC:
 		if err != nil {
-			return nil, err
+			return nil, nil, err
 		}
 	case <-time.NewTimer(splitTimeout).C:
-		return nil, errOperationTimedOut
+		return nil, nil, errOperationTimedOut
 	}
 
-	return key, nil
+	return key, storeWg.Wait, nil
 }
 
-func (self *TreeChunker) split(depth int, treeSize int64, key Key, data io.Reader, size int64, jobC chan *hashJob, chunkC chan *Chunk, errC chan error, quitC chan bool, parentWg, swg, wwg *sync.WaitGroup) {
+func (self *TreeChunker) split(depth int, treeSize int64, key Key, data io.Reader, size int64, jobC chan *hashJob, chunkC chan *Chunk, errC chan error, quitC chan bool, parentWg, storeWg *sync.WaitGroup) {
 
 	//
 
@@ -225,7 +217,7 @@ func (self *TreeChunker) split(depth int, treeSize int64, key Key, data io.Reade
 		subTreeKey := chunk[8+i*self.hashSize : 8+(i+1)*self.hashSize]
 
 		childrenWg.Add(1)
-		self.split(depth-1, treeSize/self.branches, subTreeKey, data, secSize, jobC, chunkC, errC, quitC, childrenWg, swg, wwg)
+		self.split(depth-1, treeSize/self.branches, subTreeKey, data, secSize, jobC, chunkC, errC, quitC, childrenWg, storeWg)
 
 		i++
 		pos += treeSize
@@ -237,11 +229,8 @@ func (self *TreeChunker) split(depth int, treeSize int64, key Key, data io.Reade
 
 	worker := self.getWorkerCount()
 	if int64(len(jobC)) > worker && worker < ChunkProcessors {
-		if wwg != nil {
-			wwg.Add(1)
-		}
 		self.incrementWorkerCount()
-		go self.hashWorker(jobC, chunkC, errC, quitC, swg, wwg)
+		go self.hashWorker(jobC, chunkC, errC, quitC, storeWg)
 
 	}
 	select {
@@ -250,13 +239,10 @@ func (self *TreeChunker) split(depth int, treeSize int64, key Key, data io.Reade
 	}
 }
 
-func (self *TreeChunker) hashWorker(jobC chan *hashJob, chunkC chan *Chunk, errC chan error, quitC chan bool, swg, wwg *sync.WaitGroup) {
+func (self *TreeChunker) hashWorker(jobC chan *hashJob, chunkC chan *Chunk, errC chan error, quitC chan bool, storeWg *sync.WaitGroup) {
 	defer self.decrementWorkerCount()
 
 	hasher := self.hashFunc()
-	if wwg != nil {
-		defer wwg.Done()
-	}
 	for {
 		select {
 
@@ -265,7 +251,7 @@ func (self *TreeChunker) hashWorker(jobC chan *hashJob, chunkC chan *Chunk, errC
 				return
 			}
 			// now we got the hashes in the chunk, then hash the chunks
-			self.hashChunk(hasher, job, chunkC, swg)
+			self.hashChunk(hasher, job, chunkC, storeWg)
 		case <-quitC:
 			return
 		}
@@ -275,34 +261,31 @@ func (self *TreeChunker) hashWorker(jobC chan *hashJob, chunkC chan *Chunk, errC
 // The treeChunkers own Hash hashes together
 // - the size (of the subtree encoded in the Chunk)
 // - the Chunk, ie. the contents read from the input reader
-func (self *TreeChunker) hashChunk(hasher SwarmHash, job *hashJob, chunkC chan *Chunk, swg *sync.WaitGroup) {
+func (self *TreeChunker) hashChunk(hasher SwarmHash, job *hashJob, chunkC chan *Chunk, storeWg *sync.WaitGroup) {
 	hasher.ResetWithLength(job.chunk[:8]) // 8 bytes of length
 	hasher.Write(job.chunk[8:])           // minus 8 []byte length
 	h := hasher.Sum(nil)
 
-	newChunk := &Chunk{
+	newChunk := NewChunk(h, nil)
-		Key:   h,
+	newChunk.SData = job.chunk
-		SData: job.chunk,
+	newChunk.Size = job.size
-		Size:  job.size,
-		wg:    swg,
-	}
 
 	// report hash of this chunk one level up (keys corresponds to the proper subslice of the parent chunk)
 	copy(job.key, h)
 	// send off new chunk to storage
-	if chunkC != nil {
-		if swg != nil {
-			swg.Add(1)
-		}
-	}
 	job.parentWg.Done()
 
 	if chunkC != nil {
 		chunkC <- newChunk
+		storeWg.Add(1)
+		go func() {
+			defer storeWg.Done()
+			<-newChunk.dbStored
+		}()
 	}
 }
 
-func (self *TreeChunker) Append(key Key, data io.Reader, chunkC chan *Chunk, swg, wwg *sync.WaitGroup) (Key, error) {
+func (self *TreeChunker) Append(key Key, data io.Reader, chunkC chan *Chunk) (Key, error) {
 	return nil, errAppendOppNotSuported
 }
 
@@ -456,10 +439,8 @@ func (self *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, tr
 // block until they time out or arrive
 // abort if quitC is readable
 func retrieve(key Key, chunkC chan *Chunk, quitC chan bool) *Chunk {
-	chunk := &Chunk{
+	chunk := NewChunk(key, nil)
-		Key: key,
+	chunk.C = make(chan bool)
-		C:   make(chan bool), // close channel to signal data delivery
-	}
 	// submit chunk for retrieval
 	select {
 	case chunkC <- chunk: // submit retrieval request, someone should be listening on the other side (or we will time out globally)

swarm/storage/common_test.go

@@ -81,7 +81,14 @@ func mput(store ChunkStore, processors int, n int, f func(i int) *Chunk) (hs []K
 		go func() {
 			defer wg.Done()
 			for chunk := range c {
+				wg.Add(1)
+				chunk := chunk
+				go func() {
+					defer wg.Done()
+
 					store.Put(chunk)
+					<-chunk.dbStored
+				}()
 			}
 		}()
 	}

swarm/storage/dbstore.go

@@ -23,9 +23,13 @@
 package storage
 
 import (
+	"archive/tar"
 	"bytes"
 	"encoding/binary"
+	"encoding/hex"
 	"fmt"
+	"io"
+	"io/ioutil"
 	"sync"
 
 	"github.com/ethereum/go-ethereum/log"
@@ -74,7 +78,12 @@ type DbStore struct {
 
 	hashfunc SwarmHasher
 	po       func(Key) uint8
-	lock     sync.Mutex
+
+	batchC   chan bool
+	quit     chan struct{}
+	batchesC chan struct{}
+	batch    *leveldb.Batch
+	lock     sync.RWMutex
 	trusted  bool // if hash integity check is to be performed (for testing only)
 }
 
@@ -84,6 +93,13 @@ type DbStore struct {
 func NewDbStore(path string, hash SwarmHasher, capacity uint64, po func(Key) uint8) (s *DbStore, err error) {
 	s = new(DbStore)
 	s.hashfunc = hash
+
+	s.batchC = make(chan bool)
+	s.quit = make(chan struct{})
+	s.batchesC = make(chan struct{}, 1)
+	go s.writeBatches()
+	s.batch = new(leveldb.Batch)
+
 	s.db, err = NewLDBDatabase(path)
 	if err != nil {
 		return nil, err
@@ -96,8 +112,6 @@ func NewDbStore(path string, hash SwarmHasher, capacity uint64, po func(Key) uin
 	s.gcStartPos[0] = kpIndex
 	s.gcArray = make([]*gcItem, gcArraySize)
 
-	data, _ := s.db.Get(keyEntryCnt)
-	s.entryCnt = BytesToU64(data)
 	s.bucketCnt = make([]uint64, 0x100)
 	for i := 0; i < 0x100; i++ {
 		k := make([]byte, 2)
@@ -105,18 +119,17 @@ func NewDbStore(path string, hash SwarmHasher, capacity uint64, po func(Key) uin
 		k[1] = byte(uint8(i))
 		cnt, _ := s.db.Get(k)
 		s.bucketCnt[i] = BytesToU64(cnt)
+		s.bucketCnt[i]++
 	}
+	data, _ := s.db.Get(keyEntryCnt)
+	s.entryCnt = BytesToU64(data)
+	s.entryCnt++
 	data, _ = s.db.Get(keyAccessCnt)
-	//s.accessCnt = BytesToU64(data)
+	s.accessCnt = BytesToU64(data)
-	if len(data) == 8 {
-		s.accessCnt = binary.LittleEndian.Uint64(data)
 	s.accessCnt++
-	}
 	data, _ = s.db.Get(keyDataIdx)
-	if len(data) == 8 {
 	s.dataIdx = BytesToU64(data)
 	s.dataIdx++
-	}
 
 	s.gcPos, _ = s.db.Get(keyGCPos)
 	if s.gcPos == nil {
@@ -295,6 +308,92 @@ func (s *DbStore) collectGarbage(ratio float32) {
 	s.db.Put(keyGCPos, s.gcPos)
 }
 
+// Export writes all chunks from the store to a tar archive, returning the
+// number of chunks written.
+func (s *DbStore) Export(out io.Writer) (int64, error) {
+	tw := tar.NewWriter(out)
+	defer tw.Close()
+
+	it := s.db.NewIterator()
+	defer it.Release()
+	var count int64
+	for ok := it.Seek([]byte{kpIndex}); ok; ok = it.Next() {
+		key := it.Key()
+		if (key == nil) || (key[0] != kpIndex) {
+			break
+		}
+
+		var index dpaDBIndex
+		decodeIndex(it.Value(), &index)
+
+		hash := key[1:]
+
+		data, err := s.db.Get(getDataKey(index.Idx, s.po(hash)))
+		if err != nil {
+			log.Warn(fmt.Sprintf("Chunk %x found but could not be accessed: %v", key[:], err))
+			continue
+		}
+
+		hdr := &tar.Header{
+			Name: hex.EncodeToString(hash),
+			Mode: 0644,
+			Size: int64(len(data)),
+		}
+		if err := tw.WriteHeader(hdr); err != nil {
+			return count, err
+		}
+		if _, err := tw.Write(data); err != nil {
+			return count, err
+		}
+		count++
+	}
+
+	return count, nil
+}
+
+// of chunks read.
+func (s *DbStore) Import(in io.Reader) (int64, error) {
+	tr := tar.NewReader(in)
+
+	var count int64
+	var wg sync.WaitGroup
+	for {
+		hdr, err := tr.Next()
+		if err == io.EOF {
+			break
+		} else if err != nil {
+			return count, err
+		}
+
+		if len(hdr.Name) != 64 {
+			log.Warn("ignoring non-chunk file", "name", hdr.Name)
+			continue
+		}
+
+		key, err := hex.DecodeString(hdr.Name)
+		if err != nil {
+			log.Warn("ignoring invalid chunk file", "name", hdr.Name, "err", err)
+			continue
+		}
+
+		data, err := ioutil.ReadAll(tr)
+		if err != nil {
+			return count, err
+		}
+		chunk := NewChunk(key, nil)
+		chunk.SData = data
+		s.Put(chunk)
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			<-chunk.dbStored
+		}()
+		count++
+	}
+	wg.Wait()
+	return count, nil
+}
+
 func (s *DbStore) Cleanup() {
 	//Iterates over the database and checks that there are no faulty chunks
 	it := s.db.NewIterator()
@@ -334,26 +433,6 @@ func (s *DbStore) Cleanup() {
 	log.Warn(fmt.Sprintf("Found %v errors out of %v entries", errorsFound, total))
 }
 
-func (s *DbStore) Dump() {
-	//Iterates over the database and checks that there are no faulty chunks
-	it := s.db.NewIterator()
-	startPosition := []byte{kpIndex}
-	it.Seek(startPosition)
-	var key []byte
-	var total int
-	for it.Valid() {
-		key = it.Key()
-		if (key == nil) || (key[0] != kpIndex) {
-			break
-		}
-		total++
-		fmt.Printf("%x\n", key[1:])
-		it.Next()
-	}
-	it.Release()
-	log.Warn(fmt.Sprintf("logged %v chunks", total))
-}
-
 func (s *DbStore) ReIndex() {
 	//Iterates over the database and checks that there are no faulty chunks
 	it := s.db.NewIterator()
@@ -411,6 +490,13 @@ func (s *DbStore) delete(idx uint64, idxKey []byte, po uint8) {
 	s.db.Write(batch)
 }
 
+func (s *DbStore) CurrentBucketStorageIndex(po uint8) uint64 {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+
+	return s.bucketCnt[po]
+}
+
 func (s *DbStore) Size() uint64 {
 	s.lock.Lock()
 	defer s.lock.Unlock()
@@ -418,11 +504,64 @@ func (s *DbStore) Size() uint64 {
 }
 
 func (s *DbStore) CurrentStorageIndex() uint64 {
-	s.lock.Lock()
+	s.lock.RLock()
-	defer s.lock.Unlock()
+	defer s.lock.RUnlock()
 	return s.dataIdx
 }
 
+// TODO: remove the old code for Put
+// func (s *DbStore) Put(chunk *Chunk) {
+// 	s.lock.Lock()
+// 	defer s.lock.Unlock()
+
+// 	ikey := getIndexKey(chunk.Key)
+// 	var index dpaDBIndex
+
+// 	if s.tryAccessIdx(ikey, &index) {
+// 		if chunk.dbStored != nil {
+// 			close(chunk.dbStored)
+// 		}
+// 		log.Trace(fmt.Sprintf("Storing to DB: chunk already exists, only update access"))
+// 		return // already exists, only update access
+// 	}
+
+// 	data := encodeData(chunk)
+
+// 	if s.entryCnt >= s.capacity {
+// 		s.collectGarbage(gcArrayFreeRatio)
+// 	}
+
+// 	po := s.po(chunk.Key)
+// 	t_datakey := getDataKey(s.dataIdx, po)
+// 	s.batch.Put(t_datakey, data)
+
+// 	index.Idx = s.dataIdx
+// 	s.updateIndexAccess(&index)
+
+// 	idata := encodeIndex(&index)
+// 	s.batch.Put(ikey, idata)
+
+// 	s.batch.Put(keyEntryCnt, U64ToBytes(s.entryCnt))
+// 	s.entryCnt++
+// 	s.batch.Put(keyDataIdx, U64ToBytes(s.dataIdx))
+// 	s.dataIdx++
+// 	accesscnt := make([]byte, 8)
+// 	binary.LittleEndian.PutUint64(accesscnt, s.accessCnt)
+// 	s.batch.Put(keyAccessCnt, accesscnt)
+// 	s.accessCnt++
+
+// 	s.bucketCnt[po]++
+// 	cntKey := make([]byte, 2)
+// 	cntKey[0] = keyDistanceCnt
+// 	cntKey[1] = po
+// 	s.batch.Put(cntKey, U64ToBytes(s.bucketCnt[po]))
+
+// 	if chunk.dbStored != nil {
+// 		close(chunk.dbStored)
+// 	}
+// 	log.Trace(fmt.Sprintf("DbStore.Put: %v. db storage counter: %v ", chunk.Key.Log(), s.dataIdx))
+// }
+
 func (s *DbStore) Put(chunk *Chunk) {
 	s.lock.Lock()
 	defer s.lock.Unlock()
@@ -430,54 +569,84 @@ func (s *DbStore) Put(chunk *Chunk) {
 	ikey := getIndexKey(chunk.Key)
 	var index dpaDBIndex
 
-	if s.tryAccessIdx(ikey, &index) {
+	po := s.po(chunk.Key)
-		if chunk.dbStored != nil {
+
+	idata, err := s.db.Get(ikey)
+	if err != nil {
+		s.doPut(chunk, ikey, &index, po)
+	} else {
+		log.Trace(fmt.Sprintf("DbStore: chunk already exists, only update access"))
+		decodeIndex(idata, &index)
 		close(chunk.dbStored)
 	}
-		log.Trace(fmt.Sprintf("Storing to DB: chunk already exists, only update access"))
+	index.Access = s.accessCnt
-		return // already exists, only update access
-	}
-
-	data := encodeData(chunk)
-
-	if s.entryCnt >= s.capacity {
-		s.collectGarbage(gcArrayFreeRatio)
-	}
-
-	batch := new(leveldb.Batch)
-
-	po := s.po(chunk.Key)
-	t_datakey := getDataKey(s.dataIdx, po)
-	batch.Put(t_datakey, data)
-
-	index.Idx = s.dataIdx
-	s.updateIndexAccess(&index)
-
-	idata := encodeIndex(&index)
-	batch.Put(ikey, idata)
-
-	batch.Put(keyEntryCnt, U64ToBytes(s.entryCnt))
-	s.entryCnt++
-	batch.Put(keyDataIdx, U64ToBytes(s.dataIdx))
-	s.dataIdx++
-	accesscnt := make([]byte, 8)
-	binary.LittleEndian.PutUint64(accesscnt, s.accessCnt)
-	batch.Put(keyAccessCnt, accesscnt)
 	s.accessCnt++
+	idata = encodeIndex(&index)
+	s.batch.Put(ikey, idata)
+	select {
+	case <-s.quit:
+	case s.batchesC <- struct{}{}:
+	default:
+	}
+}
+
+// force putting into db, does not check access index
+func (s *DbStore) doPut(chunk *Chunk, ikey []byte, index *dpaDBIndex, po uint8) {
+	data := encodeData(chunk)
+	s.batch.Put(getDataKey(s.dataIdx, po), data)
+	index.Idx = s.dataIdx
+	s.entryCnt++
+	s.dataIdx++
 
 	s.bucketCnt[po]++
 	cntKey := make([]byte, 2)
 	cntKey[0] = keyDistanceCnt
 	cntKey[1] = po
-	batch.Put(cntKey, U64ToBytes(s.bucketCnt[po]))
+	s.batch.Put(cntKey, U64ToBytes(s.bucketCnt[po]))
 
-	s.db.Write(batch)
+	batchC := s.batchC
-	if chunk.dbStored != nil {
+	go func() {
+		<-batchC
 		close(chunk.dbStored)
-	}
+	}()
+
 	log.Trace(fmt.Sprintf("DbStore.Put: %v. db storage counter: %v ", chunk.Key.Log(), s.dataIdx))
 }
 
+func (s *DbStore) writeBatches() {
+	for range s.batchesC {
+		s.lock.Lock()
+		b := s.batch
+		e := s.entryCnt
+		d := s.dataIdx
+		a := s.accessCnt
+		c := s.batchC
+		s.batchC = make(chan bool)
+		s.batch = new(leveldb.Batch)
+		s.lock.Unlock()
+		log.Trace(fmt.Sprintf("DbStore: spawn batch write (%d chunks) ", b.Len()))
+		s.writeBatch(b, e, d, a)
+		close(c)
+		if e >= s.capacity {
+			log.Trace(fmt.Sprintf("DbStore: collecting garbage...(%d chunks)", e))
+			s.collectGarbage(gcArrayFreeRatio)
+		}
+	}
+	log.Trace(fmt.Sprintf("DbStore: quit batch write loop"))
+}
+
+// must be called non concurrently
+func (s *DbStore) writeBatch(b *leveldb.Batch, entryCnt, dataIdx, accessCnt uint64) {
+	b.Put(keyEntryCnt, U64ToBytes(entryCnt))
+	b.Put(keyDataIdx, U64ToBytes(dataIdx))
+	b.Put(keyAccessCnt, U64ToBytes(accessCnt))
+	l := s.batch.Len()
+	if err := s.db.Write(b); err != nil {
+		log.Error(fmt.Sprintf("unable to write batch: %v", err))
+	}
+	log.Trace(fmt.Sprintf("DbStore: batch write (%d chunks) complete", l))
+}
+
 // try to find index; if found, update access cnt and return true
 func (s *DbStore) tryAccessIdx(ikey []byte, index *dpaDBIndex) bool {
 	idata, err := s.db.Get(ikey)
@@ -485,20 +654,11 @@ func (s *DbStore) tryAccessIdx(ikey []byte, index *dpaDBIndex) bool {
 		return false
 	}
 	decodeIndex(idata, index)
-
+	s.batch.Put(keyAccessCnt, U64ToBytes(s.accessCnt))
-	batch := new(leveldb.Batch)
-
-	accesscnt := make([]byte, 8)
-	binary.LittleEndian.PutUint64(accesscnt, s.accessCnt)
-	batch.Put(keyAccessCnt, accesscnt)
-
 	s.accessCnt++
-	s.updateIndexAccess(index)
+	index.Access = s.accessCnt
 	idata = encodeIndex(index)
-	batch.Put(ikey, idata)
+	s.batch.Put(ikey, idata)
-
-	s.db.Write(batch)
-
 	return true
 }
 
@@ -537,9 +697,7 @@ func (s *DbStore) get(key Key) (chunk *Chunk, err error) {
 			}
 		}
 
-		chunk = &Chunk{
+		chunk = NewChunk(key, nil)
-			Key: key,
-		}
 		decodeData(data, chunk)
 	} else {
 		err = notFound

swarm/storage/dbstore_test.go

@@ -112,7 +112,11 @@ func TestIterator(t *testing.T) {
 	var poc uint
 	chunkkeys := NewKeyCollection(chunkcount)
 	chunkkeys_results := NewKeyCollection(chunkcount)
-	chunks := make([]Chunk, chunkcount)
+	var chunks []*Chunk
+
+	for i := 0; i < chunkcount; i++ {
+		chunks = append(chunks, NewChunk(nil, nil))
+	}
 
 	db, err := newTestDbStore()
 	if err != nil {
@@ -123,7 +127,7 @@ func TestIterator(t *testing.T) {
 	FakeChunk(getDefaultChunkSize(), chunkcount, chunks)
 
 	for i = 0; i < len(chunks); i++ {
-		db.Put(&chunks[i])
+		db.Put(chunks[i])
 		chunkkeys[i] = chunks[i].Key
 	}
 

swarm/storage/dpa.go

@@ -97,8 +97,8 @@ func (self *DPA) Retrieve(key Key) LazySectionReader {
 
 // Public API. Main entry point for document storage directly. Used by the
 // FS-aware API and httpaccess
-func (self *DPA) Store(data io.Reader, size int64, swg *sync.WaitGroup, wwg *sync.WaitGroup) (key Key, err error) {
+func (self *DPA) Store(data io.Reader, size int64) (key Key, wait func(), err error) {
-	return self.Chunker.Split(data, size, self.storeC, swg, wwg)
+	return self.Chunker.Split(data, size, self.storeC)
 }
 
 func (self *DPA) Start() {
@@ -163,12 +163,8 @@ func (self *DPA) storeLoop() {
 }
 
 func (self *DPA) storeWorker() {
-
 	for chunk := range self.storeC {
 		self.Put(chunk)
-		if chunk.wg != nil {
-			chunk.wg.Done()
-		}
 		select {
 		case <-self.quitC:
 			return

swarm/storage/dpa_test.go

@@ -21,7 +21,6 @@ import (
 	"io"
 	"io/ioutil"
 	"os"
-	"sync"
 	"testing"
 )
 
@@ -50,12 +49,11 @@ func TestDPArandom(t *testing.T) {
 	defer os.RemoveAll("/tmp/bzz")
 
 	reader, slice := testDataReaderAndSlice(testDataSize)
-	wg := &sync.WaitGroup{}
+	key, wait, err := dpa.Store(reader, testDataSize)
-	key, err := dpa.Store(reader, testDataSize, wg, nil)
 	if err != nil {
 		t.Errorf("Store error: %v", err)
 	}
-	wg.Wait()
+	wait()
 	resultReader := dpa.Retrieve(key)
 	resultSlice := make([]byte, len(slice))
 	n, err := resultReader.ReadAt(resultSlice, 0)
@@ -106,12 +104,11 @@ func TestDPA_capacity(t *testing.T) {
 	}
 	dpa.Start()
 	reader, slice := testDataReaderAndSlice(testDataSize)
-	wg := &sync.WaitGroup{}
+	key, wait, err := dpa.Store(reader, testDataSize)
-	key, err := dpa.Store(reader, testDataSize, wg, nil)
 	if err != nil {
 		t.Errorf("Store error: %v", err)
 	}
-	wg.Wait()
+	wait()
 	resultReader := dpa.Retrieve(key)
 	resultSlice := make([]byte, len(slice))
 	n, err := resultReader.ReadAt(resultSlice, 0)

swarm/storage/localstore.go

@@ -42,7 +42,6 @@ func NewLocalStore(hash SwarmHasher, params *StoreParams, basekey []byte) (*Loca
 // LocalStore is itself a chunk store
 // unsafe, in that the data is not integrity checked
 func (self *LocalStore) Put(chunk *Chunk) {
-	chunk.dbStored = make(chan bool)
 	self.memStore.Put(chunk)
 	if chunk.wg != nil {
 		chunk.wg.Add(1)

swarm/storage/memstore.go

@@ -280,13 +280,9 @@ func (s *MemStore) removeOldest() {
 
 	}
 
-	if node.entry.dbStored != nil {
 	log.Trace(fmt.Sprintf("Memstore Clean: Waiting for chunk %v to be saved", node.entry.Key.Log()))
 	<-node.entry.dbStored
 	log.Trace(fmt.Sprintf("Memstore Clean: Chunk %v saved to DBStore. Ready to clear from mem.", node.entry.Key.Log()))
-	} else {
-		log.Trace(fmt.Sprintf("Memstore Clean: Chunk %v already in DB. Ready to delete.", node.entry.Key.Log()))
-	}
 
 	if node.entry.SData != nil {
 		node.entry = nil

swarm/storage/netstore.go

@@ -131,7 +131,7 @@ func (self *NetStore) Get(key Key) (*Chunk, error) {
 	}
 	// no data and no request status
 	log.Trace(fmt.Sprintf("NetStore.Get: %v not found locally. open new request", key))
-	chunk = NewChunk(key, newRequestStatus(key))
+	chunk = NewChunk(key, NewRequestStatus(key))
 	self.localStore.memStore.Put(chunk)
 	go self.cloud.Retrieve(chunk)
 	return chunk, nil

swarm/storage/pyramid.go

@@ -271,12 +271,10 @@ func (self *PyramidChunker) processChunk(id int64, hasher SwarmHash, job *chunkJ
 	hasher.Write(job.chunk[8:])           // minus 8 []byte length
 	h := hasher.Sum(nil)
 
-	newChunk := &Chunk{
+	newChunk := NewChunk(h, nil)
-		Key:   h,
+	newChunk.SData = job.chunk
-		SData: job.chunk,
+	newChunk.Size = job.size
-		Size:  job.size,
+	newChunk.wg = swg
-		wg:    swg,
-	}
 
 	// report hash of this chunk one level up (keys corresponds to the proper subslice of the parent chunk)
 	copy(job.key, h)

swarm/storage/types.go

@@ -180,7 +180,7 @@ type RequestStatus struct {
 	Requesters map[uint64][]interface{}
 }
 
-func newRequestStatus(key Key) *RequestStatus {
+func NewRequestStatus(key Key) *RequestStatus {
 	return &RequestStatus{
 		Key:        key,
 		Requesters: make(map[uint64][]interface{}),
@@ -205,10 +205,14 @@ type Chunk struct {
 }
 
 func NewChunk(key Key, rs *RequestStatus) *Chunk {
-	return &Chunk{Key: key, Req: rs}
+	return &Chunk{Key: key, Req: rs, dbStored: make(chan bool)}
 }
 
-func FakeChunk(size int64, count int, chunks []Chunk) int {
+func (c *Chunk) WaitToStore() {
+	<-c.dbStored
+}
+
+func FakeChunk(size int64, count int, chunks []*Chunk) int {
 	var i int
 	hasher := MakeHashFunc(SHA3Hash)()
 	chunksize := getDefaultChunkSize()
@@ -269,14 +273,14 @@ type Splitter interface {
 	   The caller gets returned an error channel, if an error is encountered during splitting, it is fed to errC error channel.
 	   A closed error signals process completion at which point the key can be considered final if there were no errors.
 	*/
-	Split(io.Reader, int64, chan *Chunk, *sync.WaitGroup, *sync.WaitGroup) (Key, error)
+	Split(io.Reader, int64, chan *Chunk) (Key, func(), error)
 
 	/* This is the first step in making files mutable (not chunks)..
 	   Append allows adding more data chunks to the end of the already existsing file.
 	   The key for the root chunk is supplied to load the respective tree.
 	   Rest of the parameters behave like Split.
 	*/
-	Append(Key, io.Reader, chan *Chunk, *sync.WaitGroup, *sync.WaitGroup) (Key, error)
+	Append(Key, io.Reader, chan *Chunk) (Key, error)
 }
 
 type Joiner interface {

swarm/swarm.go

@@ -97,7 +97,7 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, ensClient *e
 	log.Debug(fmt.Sprintf("Setting up Swarm service components"))
 
 	hash := storage.MakeHashFunc(config.ChunkerParams.Hash)
-	self.lstore, err = storage.NewLocalStore(hash, config.StoreParams)
+	self.lstore, err = storage.NewLocalStore(hash, config.StoreParams, common.Hex2Bytes(config.BzzKey))
 	if err != nil {
 		return
 	}
@@ -346,32 +346,6 @@ func (self *Swarm) SetChequebook(ctx context.Context) error {
 	return nil
 }
 
-// Local swarm without netStore
-func NewLocalSwarm(datadir, port string) (self *Swarm, err error) {
-
-	prvKey, err := crypto.GenerateKey()
-	if err != nil {
-		return
-	}
-
-	config := api.NewConfig()
-	config.Path = datadir
-	config.Port = port
-	config.Init(prvKey)
-
-	dpa, err := storage.NewLocalDPA(datadir)
-	if err != nil {
-		return
-	}
-
-	self = &Swarm{
-		api:    api.NewApi(dpa, nil),
-		config: config,
-	}
-
-	return
-}
-
 // serialisable info about swarm
 type Info struct {
 	*api.Config