p2p/protocols, swarm/network/stream, swarm/storage: clean debug changes

Branch swarm-network-rewrite-syncer-test contains a number of
changes related to swarm/network/stream package debugging.
This change removes this changes and sets changed variables to
the ones in swarm-network-rewrite-syncer branch.

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)
 }
 

swarm/network/stream/delivery.go

@@ -34,14 +34,11 @@ const (
 )
 
 type Delivery struct {
-	db          *storage.DBAPI
+	db       *storage.DBAPI
-	overlay     network.Overlay
+	overlay  network.Overlay
-	receiveC    chan *ChunkDeliveryMsg
+	receiveC chan *ChunkDeliveryMsg
-	getPeer     func(discover.NodeID) *Peer
+	getPeer  func(discover.NodeID) *Peer
-	quit        chan struct{}
+	quit     chan struct{}
-	counterIn   int
-	counterDone int
-	counterHash int
 }
 
 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{})
+R:
-	timer := time.NewTimer(2 * time.Second)
-	defer timer.Stop()
-	// R:
 	for req := range d.receiveC {
-		log.Error("pop from receiveC", "peer", req.peer.ID(), "hash", storage.Key(req.Key).Hex())
+		// this should be has locally
-		timer.Reset(1 * time.Second)
+		chunk, err := d.db.Get(req.Key)
-		go func(req *ChunkDeliveryMsg) {
+		if !bytes.Equal(chunk.Key, req.Key) {
-			defer func() { done <- struct{}{} }()
+			panic(fmt.Errorf("processReceivedChunks: chunk key %s != req key %s (peer %s)", chunk.Key.Hex(), storage.Key(req.Key).Hex(), req.peer.ID()))
-			// 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())
 		}
+		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))
-}

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):
+		case <-time.After(30 * time.Second):
-			p.Drop(errors.New("timeout waiting for batch to be delivered"))
+			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)

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)
+	log.Trace("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)
-	}
 	return p.SendPriority(msg, s.priority)
 }
 

swarm/network/stream/stream.go

@@ -38,8 +38,8 @@ const (
 	Mid
 	High
 	Top
-	PriorityQueue        // number of queues
+	PriorityQueue         // number of queues
-	PriorityQueueCap = 3 // queue capacity
+	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

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
 	}
 

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, _ := rlp.EncodeToBytes(index)
-
-	data, err := json.Marshal(index)
-	if err != nil {
-		panic(err)
-	}
 	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)
+	dec := rlp.NewStream(bytes.NewReader(data), 0)
-	// return dec.Decode(index)
+	return dec.Decode(index)
-	return json.Unmarshal(data, 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

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
 }
 

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)