diff --git a/p2p/protocols/protocol.go b/p2p/protocols/protocol.go index 48fc5e9fcc..7b04069edf 100644 --- a/p2p/protocols/protocol.go +++ b/p2p/protocols/protocol.go @@ -34,7 +34,6 @@ import ( "reflect" "sync" - "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p" ) @@ -211,7 +210,6 @@ func (p *Peer) Run(handler func(msg interface{}) error) error { // if they are useful for other protocols // overwrite Disconnect for testing, so that protocol readloop quits func (p *Peer) Drop(err error) { - log.Error("p2p protocol DROP", "err", err) p.Disconnect(p2p.DiscSubprotocolError) } diff --git a/swarm/network/stream/delivery.go b/swarm/network/stream/delivery.go index 2f7dde3721..30bdd80fd8 100644 --- a/swarm/network/stream/delivery.go +++ b/swarm/network/stream/delivery.go @@ -34,14 +34,11 @@ const ( ) type Delivery struct { - db *storage.DBAPI - overlay network.Overlay - receiveC chan *ChunkDeliveryMsg - getPeer func(discover.NodeID) *Peer - quit chan struct{} - counterIn int - counterDone int - counterHash int + db *storage.DBAPI + overlay network.Overlay + receiveC chan *ChunkDeliveryMsg + getPeer func(discover.NodeID) *Peer + quit chan struct{} } func NewDelivery(overlay network.Overlay, db *storage.DBAPI) *Delivery { @@ -160,7 +157,7 @@ func (d *Delivery) handleRetrieveRequestMsg(sp *Peer, req *RetrieveRequestMsg) e if req.SkipCheck { err := sp.Deliver(chunk, s.priority) if err != nil { - sp.Drop(fmt.Errorf("handleRetrieveRequestMsg: %v", err)) + sp.Drop(err) } } streamer.deliveryC <- chunk.Key[:] @@ -179,67 +176,39 @@ func (d *Delivery) handleRetrieveRequestMsg(sp *Peer, req *RetrieveRequestMsg) e type ChunkDeliveryMsg struct { Key storage.Key SData []byte // the stored chunk Data (incl size) - peer *Peer + peer *Peer // set in handleChunkDeliveryMsg } func (d *Delivery) handleChunkDeliveryMsg(sp *Peer, req *ChunkDeliveryMsg) error { - d.counterIn++ req.peer = sp - log.Error("push to receiveC", "hash", storage.Key(req.Key).Hex()) d.receiveC <- req return nil } func (d *Delivery) processReceivedChunks() { - done := make(chan struct{}) - timer := time.NewTimer(2 * time.Second) - defer timer.Stop() - // R: +R: for req := range d.receiveC { - log.Error("pop from receiveC", "peer", req.peer.ID(), "hash", storage.Key(req.Key).Hex()) - timer.Reset(1 * time.Second) - go func(req *ChunkDeliveryMsg) { - defer func() { done <- struct{}{} }() - // this should be has locally - log.Error("before db.Get", "peer", req.peer.ID(), "hash", storage.Key(req.Key).Hex()) - chunk, err := d.db.Get(req.Key) - if !bytes.Equal(chunk.Key, req.Key) { - panic(fmt.Errorf("processReceivedChunks: chunk key %s != req key %s (peer %s)", chunk.Key.Hex(), storage.Key(req.Key).Hex(), req.peer.ID())) - } - log.Error("after db.Get", "peer", req.peer.ID(), "chunk", chunk.Key.Hex(), "reqC", chunk.ReqC, "err", err) - if err == nil { - log.Error("found existing?", "peer", req.peer.ID(), "hash", chunk.Key.Hex()) - // continue R - return - } - if err != storage.ErrFetching { - panic(fmt.Sprintf("not in db? key %v chunk %v", req.Key, chunk)) - } - select { - case <-chunk.ReqC: - log.Error("someone else delivered?", "hash", chunk.Key.Hex()) - // continue R - return - default: - } - // go func() { - chunk.SData = req.SData - log.Error("received delivery", "peer", req.peer.ID(), "hash", chunk.Key.Hex()) - d.db.Put(chunk) - log.Error("put to db", "peer", req.peer.ID(), "hash", chunk.Key.Hex()) - chunk.WaitToStore() - close(chunk.ReqC) - //log.Warn("received delivery stored", "hash", chunk.Key) - log.Error("requesters notified", "peer", req.peer.ID(), "hash", chunk.Key.Hex()) - d.counterDone++ - // }() - }(req) - select { - case <-timer.C: - log.Error("!!!unable to process delivery", "peer", req.peer.ID(), "hash", req.Key.Hex()) - case <-done: - log.Error("done processing delivery", "peer", req.peer.ID(), "hash", req.Key.Hex()) + // this should be has locally + chunk, err := d.db.Get(req.Key) + if !bytes.Equal(chunk.Key, req.Key) { + panic(fmt.Errorf("processReceivedChunks: chunk key %s != req key %s (peer %s)", chunk.Key.Hex(), storage.Key(req.Key).Hex(), req.peer.ID())) } + if err == nil { + continue R + } + if err != storage.ErrFetching { + panic(fmt.Sprintf("not in db? key %v chunk %v", req.Key, chunk)) + } + select { + case <-chunk.ReqC: + log.Error("someone else delivered?", "hash", chunk.Key.Hex()) + continue R + default: + } + chunk.SData = req.SData + d.db.Put(chunk) + chunk.WaitToStore() + close(chunk.ReqC) } } @@ -247,18 +216,17 @@ func (d *Delivery) processReceivedChunks() { func (d *Delivery) RequestFromPeers(hash []byte, skipCheck bool, peersToSkip ...discover.NodeID) error { var success bool var err error - log.Warn("request", "hash", hash) d.overlay.EachConn(hash, 255, func(p network.OverlayConn, po int, nn bool) bool { spId := p.(*network.BzzPeer).ID() for _, p := range peersToSkip { if p == spId { - log.Warn("skip peer", "peer", spId) + log.Trace("Delivery.RequestFromPeers: skip peer", "peer", spId) return true } } sp := d.getPeer(spId) if sp == nil { - log.Warn("peer not found", "id", spId) + log.Warn("Delivery.RequestFromPeers: peer not found", "id", spId) return true } // TODO: skip light nodes that do not accept retrieve requests @@ -274,12 +242,3 @@ func (d *Delivery) RequestFromPeers(hash []byte, skipCheck bool, peersToSkip ... } return errors.New("no peer found") } - -func (d *Delivery) PrintCounters(id discover.NodeID) { - if d.counterHash != d.counterDone { - log.Error(fmt.Sprintf("delivery %s: HASH and DONE not the same", id)) - } - log.Error(fmt.Sprintf("delivery %s chunks hash: %d", id, d.counterHash)) - log.Error(fmt.Sprintf("delivery %s chunks in: %d", id, d.counterIn)) - log.Error(fmt.Sprintf("delivery %s chunks done: %d", id, d.counterDone)) -} diff --git a/swarm/network/stream/messages.go b/swarm/network/stream/messages.go index aa8f5f75c4..b4aadc75f1 100644 --- a/swarm/network/stream/messages.go +++ b/swarm/network/stream/messages.go @@ -17,7 +17,6 @@ package stream import ( - "errors" "fmt" "sync" "time" @@ -83,7 +82,7 @@ func (p *Peer) handleSubscribeMsg(req *SubscribeMsg) error { log.Debug("received subscription", "peer", p.ID(), "stream", req.Stream, "Key", req.Key, "from", req.From, "to", req.To) go func() { if err := p.SendOfferedHashes(os, req.From, req.To); err != nil { - p.Drop(fmt.Errorf("handleSubscribeMsg SendOfferedHashes: %v", err)) + p.Drop(err) } }() return nil @@ -122,8 +121,6 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error { for i := 0; i < len(hashes); i += HashSize { hash := hashes[i : i+HashSize] - p.streamer.delivery.counterHash++ - if wait := s.NeedData(hash); wait != nil { want.Set(i/HashSize, true) wg.Add(1) @@ -171,19 +168,19 @@ func (p *Peer) handleOfferedHashesMsg(req *OfferedHashesMsg) error { } go func() { select { - case <-time.After(1 * time.Second): - p.Drop(errors.New("timeout waiting for batch to be delivered")) + case <-time.After(30 * time.Second): + p.Drop(err) return case err := <-s.next: if err != nil { - p.Drop(fmt.Errorf("handleOfferedHashesMsg next: %v", err)) + p.Drop(err) return } } log.Trace("sending want batch", "peer", p.ID(), "stream", msg.Stream, "Key", msg.Key, "from", msg.From, "to", msg.To) err := p.SendPriority(msg, s.priority) if err != nil { - p.Drop(fmt.Errorf("handleOfferedHashesMsg set priority: %v", err)) + p.Drop(err) } }() return nil @@ -216,7 +213,7 @@ func (p *Peer) handleWantedHashesMsg(req *WantedHashesMsg) error { // launch in go routine since GetBatch blocks until new hashes arrive go func() { if err := p.SendOfferedHashes(s, req.From, req.To); err != nil { - p.Drop(fmt.Errorf("handleWantedHashesMsg SendOfferedHashes: %v", err)) + p.Drop(err) } }() // go p.SendOfferedHashes(s, req.From, req.To) diff --git a/swarm/network/stream/peer.go b/swarm/network/stream/peer.go index cd7cdc2c9a..75bec53818 100644 --- a/swarm/network/stream/peer.go +++ b/swarm/network/stream/peer.go @@ -28,7 +28,7 @@ import ( "github.com/ethereum/go-ethereum/swarm/storage" ) -var sendTimeout = 1 * time.Second +var sendTimeout = 5 * time.Second // Peer is the Peer extention for the streaming protocol type Peer struct { @@ -101,11 +101,7 @@ func (p *Peer) SendOfferedHashes(s *server, f, t uint64) error { Stream: s.stream, Key: s.key, } - log.Error("Swarm syncer offer batch", "peer", p.ID(), "stream", s.stream, "key", s.key, "len", len(hashes), "from", from, "to", to) - for i := 0; i < len(hashes); i += HashSize { - hash := hashes[i : i+HashSize] - log.Error("Swarm syncer offer hash", "peer", p.ID(), "stream", s.stream, "hash", storage.Key(hash).Hex(), "len", len(hashes), "from", from, "to", to) - } + log.Trace("Swarm syncer offer batch", "peer", p.ID(), "stream", s.stream, "key", s.key, "len", len(hashes), "from", from, "to", to) return p.SendPriority(msg, s.priority) } diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go index e29881df79..d6514808be 100644 --- a/swarm/network/stream/stream.go +++ b/swarm/network/stream/stream.go @@ -38,8 +38,8 @@ const ( Mid High Top - PriorityQueue // number of queues - PriorityQueueCap = 3 // queue capacity + PriorityQueue // number of queues + PriorityQueueCap = 32 // queue capacity HashSize = 32 ) @@ -268,8 +268,7 @@ type client struct { live bool stream string key []byte - // quit chan struct{} - next chan error + next chan error } // Client interface for incoming peer Streamer diff --git a/swarm/network/stream/syncer_test.go b/swarm/network/stream/syncer_test.go index e80cbb77d1..58d780c36f 100644 --- a/swarm/network/stream/syncer_test.go +++ b/swarm/network/stream/syncer_test.go @@ -34,7 +34,7 @@ import ( "github.com/ethereum/go-ethereum/swarm/storage" ) -const dataChunkCount = 1000 +const dataChunkCount = 500 func TestSyncerSimulation(t *testing.T) { testSyncBetweenNodes(t, 2, 1, dataChunkCount, true, 1) @@ -70,13 +70,6 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck t.Fatal(err.Error()) } - // DEBUG: - defer func() { - for _, id := range sim.IDs { - deliveries[id].PrintCounters(id) - } - }() - // HACK: these are global variables in the test so that they are available for // the service constructor function // TODO: will this work with exec/docker adapter? @@ -186,7 +179,6 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck default: } - log.Error("starting dbs check", "node", id) i := nodeIndex[id] var total, found int for j := i; j < nodes; j++ { @@ -196,7 +188,6 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck if err == storage.ErrFetching { <-chunk.ReqC } else if err != nil { - log.Error("not found", "index", i, "origin", j, "key", key.Hex(), "err", err) continue } // needed for leveldb not to be closed? @@ -204,7 +195,7 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck found++ } } - log.Error("sync check", "node", id, "index", i, "bin", po, "found", found, "total", total) + log.Debug("sync check", "node", id, "index", i, "bin", po, "found", found, "total", total) return total == found, nil } diff --git a/swarm/storage/dbstore.go b/swarm/storage/dbstore.go index fecb199e1b..17a7534646 100644 --- a/swarm/storage/dbstore.go +++ b/swarm/storage/dbstore.go @@ -27,14 +27,13 @@ import ( "bytes" "encoding/binary" "encoding/hex" - "encoding/json" "fmt" "io" "io/ioutil" "sync" - "time" "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/rlp" "github.com/ethereum/go-ethereum/swarm/storage/mock" "github.com/syndtr/goleveldb/leveldb" "github.com/syndtr/goleveldb/leveldb/opt" @@ -82,7 +81,6 @@ type DbStore struct { po func(Key) uint8 batchC chan bool - quit chan struct{} batchesC chan struct{} batch *leveldb.Batch lock sync.RWMutex @@ -106,7 +104,6 @@ func NewDbStore(path string, hash SwarmHasher, capacity uint64, po func(Key) uin 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) @@ -221,12 +218,7 @@ func getDataKey(idx uint64, po uint8) []byte { } func encodeIndex(index *dpaDBIndex) []byte { - //data, _ := rlp.EncodeToBytes(index) - - data, err := json.Marshal(index) - if err != nil { - panic(err) - } + data, _ := rlp.EncodeToBytes(index) return data } @@ -235,9 +227,8 @@ func encodeData(chunk *Chunk) []byte { } func decodeIndex(data []byte, index *dpaDBIndex) error { - // dec := rlp.NewStream(bytes.NewReader(data), 0) - // return dec.Decode(index) - return json.Unmarshal(data, index) + dec := rlp.NewStream(bytes.NewReader(data), 0) + return dec.Decode(index) } @@ -546,55 +537,25 @@ func (s *DbStore) CurrentStorageIndex() uint64 { } func (s *DbStore) Put(chunk *Chunk) { - log.Error("DbStore.Put", "hash", chunk.Key.Hex()) - done := make(chan struct{}) - defer close(done) - key := Key(append(make([]byte, 0), chunk.Key...)) - go func() { - log.Error("DbStore.Put WAITER STARTED", "hash", chunk.Key.Hex()) - select { - case <-time.After(1 * time.Second): - log.Error("DbStore.Put WAITING", "hash", chunk.Key.Hex()) - case <-done: - log.Error("DbStore.Put EXITED", "hash", chunk.Key.Hex()) - if !bytes.Equal(chunk.Key, key) { - panic(fmt.Errorf("DbStore.Get: chunk key %s != req key %s", chunk.Key.Hex(), key.Hex())) - } - } - }() - ikey := getIndexKey(chunk.Key) var index dpaDBIndex po := s.po(chunk.Key) - log.Error("DbStore.db.Get is being called...", "hash", chunk.Key.Hex()) - - log.Error("DbStore.LOCK acquiring", "hash", chunk.Key.Hex()) s.lock.Lock() - log.Error("DbStore.LOCK acquired", "hash", chunk.Key.Hex()) defer s.lock.Unlock() idata, err := s.db.Get(ikey) - log.Error("DbStore.db.Get done", "hash", chunk.Key.Hex(), "err", err) if err != nil { s.doPut(chunk, ikey, &index, po) batchC := s.batchC go func() { - defer func() { - if err := recover(); err != nil { - log.Error("DbStore.Put PANIC", "hash", chunk.Key.Hex(), "err", err) - } - }() - <-batchC close(chunk.dbStored) }() - log.Error("DbStore.Put doPut", "hash", chunk.Key.Hex(), "dataIdx", s.dataIdx) } else { log.Trace(fmt.Sprintf("DbStore: chunk already exists, only update access")) decodeIndex(idata, &index) close(chunk.dbStored) - log.Error("DbStore.Put already found", "hash", chunk.Key.Hex()) } index.Access = s.accessCnt s.accessCnt++ @@ -604,7 +565,6 @@ func (s *DbStore) Put(chunk *Chunk) { case s.batchesC <- struct{}{}: default: } - log.Error("DbStore.db.Put done", "hash", chunk.Key.Hex(), "err", err) } // force putting into db, does not check access index @@ -734,11 +694,6 @@ func (s *DbStore) get(key Key) (chunk *Chunk, err error) { chunk = NewChunk(key, nil) decodeData(data, chunk) - - if !bytes.Equal(chunk.Key, key) { - panic(fmt.Errorf("DbStore.Get: chunk key %s != req key %s", chunk.Key.Hex(), key.Hex())) - } - } else { err = ErrNotFound } @@ -802,9 +757,8 @@ func (s *DbStore) SyncIterator(since uint64, until uint64, po uint8, f func(Key, untilkey := getDataKey(until, po) it := s.db.NewIterator() defer it.Release() - it.Seek(sincekey) - for it.Next() { + for ok := it.Seek(sincekey); ok; ok = it.Next() { dbkey := it.Key() if dbkey[0] != keyData || dbkey[1] != byte(po) || bytes.Compare(untilkey, dbkey) < 0 { break diff --git a/swarm/storage/localstore.go b/swarm/storage/localstore.go index 40103503be..0f77488b64 100644 --- a/swarm/storage/localstore.go +++ b/swarm/storage/localstore.go @@ -17,7 +17,6 @@ package storage import ( - "bytes" "encoding/binary" "fmt" "path/filepath" @@ -103,12 +102,7 @@ func (self *LocalStore) Put(chunk *Chunk) { dbStored: chunk.dbStored, } self.memStore.Put(c) - log.Error("put to memstore", "hash", c.Key.Hex()) self.DbStore.Put(c) - log.Error("put to dbstore", "hash", c.Key.Hex()) - if !bytes.Equal(chunk.Key, c.Key) { - panic(fmt.Errorf("LocalStore.Put: chunk %s != c %s", chunk.Key.Hex(), c.Key.Hex())) - } } // Get(chunk *Chunk) looks up a chunk in the local stores @@ -132,7 +126,7 @@ func (self *LocalStore) Get(key Key) (chunk *Chunk, err error) { return } chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8])) - //self.memStore.Put(chunk) + self.memStore.Put(chunk) return } diff --git a/swarm/swarm.go b/swarm/swarm.go index b97390e369..d566918fe7 100644 --- a/swarm/swarm.go +++ b/swarm/swarm.go @@ -132,7 +132,7 @@ func NewSwarm(ctx *node.ServiceContext, backend chequebook.Backend, ensClient *e db := storage.NewDBAPI(self.lstore) delivery := stream.NewDelivery(to, db) - self.streamer = stream.NewRegistry(addr, delivery) + self.streamer = stream.NewRegistry(addr, delivery, self.lstore, false) stream.RegisterSwarmSyncerServer(self.streamer, db) stream.RegisterSwarmSyncerClient(self.streamer, db)