diff --git a/cmd/swarm/db.go b/cmd/swarm/db.go index dfd2d069b9..82db713f14 100644 --- a/cmd/swarm/db.go +++ b/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 { - utils.Fatalf("invalid arguments, please specify both (path to a local chunk database) and (path to write the tar archive to, - for stdout)") + if len(args) != 3 { + utils.Fatalf("invalid arguments, please specify both (path to a local chunk database), (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 { - utils.Fatalf("invalid arguments, please specify both (path to a local chunk database) and (path to read the tar archive from, - for stdin)") + if len(args) != 3 { + utils.Fatalf("invalid arguments, please specify both (path to a local chunk database), (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 { - utils.Fatalf("invalid arguments, please specify (path to a local chunk database)") + if len(args) != 2 { + utils.Fatalf("invalid arguments, please specify (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[:])) }) } diff --git a/cmd/swarm/hash.go b/cmd/swarm/hash.go index 792e8d0d7a..a6e6a6ba76 100644 --- a/cmd/swarm/hash.go +++ b/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 { diff --git a/swarm/api/api.go b/swarm/api/api.go index 8c4bca2ec0..e187b748d6 100644 --- a/swarm/api/api.go +++ b/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) { - return self.dpa.Store(data, size, wg, nil) +func (self *Api) Store(data io.Reader, size int64) (key storage.Key, wait func(), err error) { + 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, err := self.dpa.Store(r, int64(len(content)), wg, nil) + key, waitContent, err := self.dpa.Store(r, int64(len(content))) 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, nil + return key, func() { + waitContent() + waitManifest() + }, nil } // Get uses iterative manifest retrieval and prefix matching diff --git a/swarm/api/api_test.go b/swarm/api/api_test.go index e673f76c42..44bf8aadc2 100644 --- a/swarm/api/api_test.go +++ b/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) } diff --git a/swarm/api/filesystem.go b/swarm/api/filesystem.go index f5dc90e2e5..b6a2de8862 100644 --- a/swarm/api/filesystem.go +++ b/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{} - hash, err = self.api.dpa.Store(f, stat.Size(), wg, nil) + var wait func() + 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) diff --git a/swarm/api/http/server.go b/swarm/api/http/server.go index 74341899d2..7adddd9ff4 100644 --- a/swarm/api/http/server.go +++ b/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 diff --git a/swarm/api/http/server_test.go b/swarm/api/http/server_test.go index 305d5cf7db..6a35d2c785 100644 --- a/swarm/api/http/server_test.go +++ b/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") diff --git a/swarm/api/manifest.go b/swarm/api/manifest.go index 685a300fca..f6279a4ced 100644 --- a/swarm/api/manifest.go +++ b/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)), wg, nil) - wg.Wait() + key, _, err2 := self.dpa.Store(sr, int64(len(manifest))) self.hash = key return err2 } diff --git a/swarm/api/storage.go b/swarm/api/storage.go index 0e3abecfe4..ae94e15cb9 100644 --- a/swarm/api/storage.go +++ b/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 } diff --git a/swarm/network/requests.go b/swarm/network/requests.go index 311acfc927..27e0ea2853 100644 --- a/swarm/network/requests.go +++ b/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 { - 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 // +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 *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 { - req := msg.(*retrieveRequestMsgData) - req.from = p +func (self *StreamerPeer) handleRetrieveRequestMsg(msg interface{}) error { + req := msg.(*retrieveRequestMsg) + 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() - self.addRequester(rs, req) + rs = storage.NewRequestStatus(req.Key) + 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) - if err != nil { - return err - } + // TODO: chunk validation + chunk := storage.NewChunk(req.Key, nil) + chunk.SData = req.SData chunk.Source = p self.netStore.Put(chunk) log.Trace(fmt.Sprintf("delivery of %v from %v", chunk, p)) diff --git a/swarm/network/streamer.go b/swarm/network/streamer.go index 7941a261dd..0207f3a0fc 100644 --- a/swarm/network/streamer.go +++ b/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 ( @@ -43,9 +45,9 @@ type Stream string // Handover represents a statement that the upstream peer hands over the stream section type Handover struct { - Stream Stream // name of stream - Start, End uint64 // index of hashes - Root common.Hash // Root hash for indexed segment inclusion proofs + Stream Stream // name of stream + Start, End uint64 // index of hashes + 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() { - wg.Wait() - msg := s.TakeoverProof(req.Stream, req.From, req.Hashes, req.Root) - self.Send(msg, s.Priority()) - }() + // go func() { + // wg.Wait() + // msg := s.TakeoverProof(req.Stream, req.From, req.Hashes, req.Root) + // 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) { - hashes, from, to, proof := s.SetNextBatch(f, t) +func (self *StreamerPeer) SendUnsyncedKeys(s OutgoingStreamer, f, t uint64, priority int) error { + 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 -var StrSpec = &protocols.Spec{ +// StreamerSpec is the spec of the streamer protocol. +var StreamerSpec = &protocols.Spec{ Name: "stream", Version: 1, MaxMsgSize: 10 * 1024 * 1024, diff --git a/swarm/network/syncer.go b/swarm/network/syncer.go index 84f61af5d6..028945384e 100644 --- a/swarm/network/syncer.go +++ b/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 { - return self.db.CurrentStorageIndex(po) +func (self *DbAccess) currentBucketStorageIndex(po uint8) uint64 { + 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++ { - stream := fmt.Sprintf("SYNC-%02d-live", po) + for po := uint8(0); po < maxPO; 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) - streamer.RegisterOutgoingStreamer(stream, func(p *StreamerPeer) (OutgoingStreamer, error) { - return NewOutgoingProvableSwarmSyncer(po, db) - }) + // stream = Stream(fmt.Sprintf("SYNC-%02d-delete", po)) + // streamer.RegisterOutgoingStreamer(stream, func(p *StreamerPeer) (OutgoingStreamer, error) { + // 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 - storeC chan storage.Chunk + retrieveC 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 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 (s *IncomingSwarmSyncer) Priority() int { + return s.priority } +// // 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++ { - stream := fmt.Sprintf("SYNC-%02d-live", po) - streamer.RegisterIncomingStreamer(stream, func(p *StreamerPeer) (OutgoingStreamer, error) { - return NewIncomingSwarmSyncer(po, Mid, sessionAt, nil, p) + for po := uint8(0); po < maxPO; po++ { + stream := Stream(fmt.Sprintf("SYNC-%02d-live", po)) + streamer.RegisterIncomingStreamer(stream, func(p *StreamerPeer) (IncomingStreamer, error) { + return NewIncomingSwarmSyncer(po, High, nil, p, nil, nil) }) - stream = fmt.Sprintf("SYNC-%02d-history", po) - streamer.RegisterIncomingStreamer(stream, func(p *StreamerPeer) (OutgoingStreamer, error) { - intervals := loadIntervals(p, po, false) - 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) + stream = Stream(fmt.Sprintf("SYNC-%02d-history", po)) + streamer.RegisterIncomingStreamer(stream, func(p *StreamerPeer) (IncomingStreamer, error) { + //intervals := loadIntervals(p, po, false) + 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) + // }) } } // 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() { - storedC := <-chunk.storedC - <-storedC - } + return chunk.WaitToStore } // NextBatch adjusts the indexes by inspecting the intervals -func (self *IncomingSwarmSyncer) NextBatch(from, to uint64) (uint64, uint64) { - if from >= self.sessionAt { // live syncing - self.intervals[1] = to - } else if to >= self.sessionAt { // history sync complete +func (self *IncomingSwarmSyncer) NextBatch(from uint64) (nextFrom uint64, nextTo uint64) { + if self.intervals[0] >= self.sessionAt { // live syncing + nextFrom = from + self.intervals[1] = from + } else if from >= self.sessionAt { // history sync complete self.intervals = nil - from = 0 - } else if len(intervals) > 2 && to >= self.intervals[2] { // filled a gap in the intervals - self.intervals[1:] = self.intervals[3:] - from = self.intervals[1] - if len(intervals) > 2 { - to = self.intervals[2] + } else if len(self.intervals) > 2 && from >= self.intervals[2] { // filled a gap in the intervals + self.intervals = append(self.intervals[:1], self.intervals[3:]...) + nextFrom = self.intervals[1] + if len(self.intervals) > 2 { + nextTo = 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 } diff --git a/swarm/pss/pss.go b/swarm/pss/pss.go index 507b4ab655..4a7564d34c 100644 --- a/swarm/pss/pss.go +++ b/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 diff --git a/swarm/storage/chunker.go b/swarm/storage/chunker.go index 98cd6e75ea..f049d39f7e 100644 --- a/swarm/storage/chunker.go +++ b/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{ - Key: h, - SData: job.chunk, - Size: job.size, - wg: swg, - } + newChunk := NewChunk(h, nil) + newChunk.SData = job.chunk + newChunk.Size = job.size // 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{ - Key: key, - C: make(chan bool), // close channel to signal data delivery - } + chunk := NewChunk(key, nil) + chunk.C = make(chan bool) // 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) diff --git a/swarm/storage/common_test.go b/swarm/storage/common_test.go index 6fd66a03d9..bdc4814d51 100644 --- a/swarm/storage/common_test.go +++ b/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 { - store.Put(chunk) + wg.Add(1) + chunk := chunk + go func() { + defer wg.Done() + + store.Put(chunk) + <-chunk.dbStored + }() } }() } diff --git a/swarm/storage/dbstore.go b/swarm/storage/dbstore.go index c1bef29823..01a6b79f7f 100644 --- a/swarm/storage/dbstore.go +++ b/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) - if len(data) == 8 { - s.accessCnt = binary.LittleEndian.Uint64(data) - s.accessCnt++ - } + s.accessCnt = BytesToU64(data) + s.accessCnt++ data, _ = s.db.Get(keyDataIdx) - if len(data) == 8 { - s.dataIdx = BytesToU64(data) - s.dataIdx++ - } + 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() - defer s.lock.Unlock() + s.lock.RLock() + 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) { - 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) - } - - 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) + 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) + } + index.Access = s.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) - if chunk.dbStored != nil { + batchC := s.batchC + 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) - - batch := new(leveldb.Batch) - - accesscnt := make([]byte, 8) - binary.LittleEndian.PutUint64(accesscnt, s.accessCnt) - batch.Put(keyAccessCnt, accesscnt) - + s.batch.Put(keyAccessCnt, U64ToBytes(s.accessCnt)) s.accessCnt++ - s.updateIndexAccess(index) + index.Access = s.accessCnt idata = encodeIndex(index) - batch.Put(ikey, idata) - - s.db.Write(batch) - + s.batch.Put(ikey, idata) return true } @@ -537,9 +697,7 @@ func (s *DbStore) get(key Key) (chunk *Chunk, err error) { } } - chunk = &Chunk{ - Key: key, - } + chunk = NewChunk(key, nil) decodeData(data, chunk) } else { err = notFound diff --git a/swarm/storage/dbstore_test.go b/swarm/storage/dbstore_test.go index 27bab975a6..ea250bfb07 100644 --- a/swarm/storage/dbstore_test.go +++ b/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 } diff --git a/swarm/storage/dpa.go b/swarm/storage/dpa.go index b8f7f5fd8f..354ff32c15 100644 --- a/swarm/storage/dpa.go +++ b/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) { - return self.Chunker.Split(data, size, self.storeC, swg, wwg) +func (self *DPA) Store(data io.Reader, size int64) (key Key, wait func(), err error) { + 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 diff --git a/swarm/storage/dpa_test.go b/swarm/storage/dpa_test.go index 3bccd82d54..391418674b 100644 --- a/swarm/storage/dpa_test.go +++ b/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, err := dpa.Store(reader, testDataSize, wg, nil) + key, wait, err := dpa.Store(reader, testDataSize) 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, err := dpa.Store(reader, testDataSize, wg, nil) + key, wait, err := dpa.Store(reader, testDataSize) 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) diff --git a/swarm/storage/localstore.go b/swarm/storage/localstore.go index 2ed9fb305a..ddc10934d8 100644 --- a/swarm/storage/localstore.go +++ b/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) diff --git a/swarm/storage/memstore.go b/swarm/storage/memstore.go index 3cb25ac625..fed1ae0094 100644 --- a/swarm/storage/memstore.go +++ b/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())) - } + 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())) if node.entry.SData != nil { node.entry = nil diff --git a/swarm/storage/netstore.go b/swarm/storage/netstore.go index 5d4f17deb1..2afba75811 100644 --- a/swarm/storage/netstore.go +++ b/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 diff --git a/swarm/storage/pyramid.go b/swarm/storage/pyramid.go index 19d493405a..2ee3d5b58a 100644 --- a/swarm/storage/pyramid.go +++ b/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{ - Key: h, - SData: job.chunk, - Size: job.size, - wg: swg, - } + newChunk := NewChunk(h, nil) + newChunk.SData = job.chunk + newChunk.Size = job.size + newChunk.wg = swg // report hash of this chunk one level up (keys corresponds to the proper subslice of the parent chunk) copy(job.key, h) diff --git a/swarm/storage/types.go b/swarm/storage/types.go index e2c111f7b1..3cfe8a6f11 100644 --- a/swarm/storage/types.go +++ b/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 { diff --git a/swarm/swarm.go b/swarm/swarm.go index 3061d07a8a..b2349c0fc4 100644 --- a/swarm/swarm.go +++ b/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