more debug

swarm/network/stream/delivery.go

@@ -161,7 +161,7 @@ type ChunkDeliveryMsg struct {
 	SData []byte // the stored chunk Data (incl size)
 }
 
-func (d *Delivery) handleChunkDeliveryMsg(req *ChunkDeliveryMsg) error {
+func (d *Delivery) handleChunkDeliveryMsg(sp *Peer, req *ChunkDeliveryMsg) error {
 	d.counterIn++
 	d.receiveC <- req
 	return nil
@@ -172,7 +172,9 @@ R:
 	for req := range d.receiveC {
 		// this should be has locally
 		chunk, err := d.db.Get(req.Key)
+		log.Error("pick from receiveC", "chunk", chunk.Key.Hex(), "reqC", chunk.ReqC, "err", err)
 		if err == nil {
+			log.Error("found existing?", "hash", chunk.Key.Hex())
 			continue R
 		}
 		if err != storage.ErrFetching {
@@ -180,17 +182,21 @@ R:
 		}
 		select {
 		case <-chunk.ReqC:
+			log.Error("someone else delivered?", "hash", chunk.Key.Hex())
 			continue R
 		default:
 		}
+		go func() {
 			chunk.SData = req.SData
+			log.Error("received delivery", "hash", chunk.Key.Hex())
 			d.db.Put(chunk)
-		log.Warn("reecived delivery", "hash", chunk.Key)
-		close(chunk.ReqC)
+			log.Error("put to db", "hash", chunk.Key.Hex())
 			chunk.WaitToStore()
+			close(chunk.ReqC)
 			//log.Warn("received delivery stored", "hash", chunk.Key)
-		log.Warn("received delivery requesters notified", "hash", chunk.Key)
+			log.Error("requesters notified", "hash", chunk.Key.Hex())
 			d.counterDone++
+		}()
 	}
 }
 

swarm/network/stream/peer.go

@@ -28,7 +28,7 @@ import (
 	"github.com/ethereum/go-ethereum/swarm/storage"
 )
 
-var sendTimeout = 5 * time.Second
+var sendTimeout = 1 * time.Second
 
 // Peer is the Peer extention for the streaming protocol
 type Peer struct {
@@ -97,7 +97,11 @@ func (p *Peer) SendOfferedHashes(s *server, f, t uint64) error {
 		Stream:        s.stream,
 		Key:           s.key,
 	}
-	log.Warn("Swarm syncer offer batch", "peer", p.ID(), "stream", s.stream, "key", s.key, "len", len(hashes), "from", from, "to", to)
+	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)
+	}
 	return p.SendPriority(msg, s.priority)
 }
 

swarm/network/stream/stream.go

@@ -39,7 +39,7 @@ const (
 	High
 	Top
 	PriorityQueue        // number of queues
-	PriorityQueueCap = 32 // queue capacity
+	PriorityQueueCap = 3 // queue capacity
 	HashSize         = 32
 )
 
@@ -227,7 +227,7 @@ func (p *Peer) HandleMsg(msg interface{}) error {
 		return p.handleWantedHashesMsg(msg)
 
 	case *ChunkDeliveryMsg:
-		return p.streamer.delivery.handleChunkDeliveryMsg(msg)
+		return p.streamer.delivery.handleChunkDeliveryMsg(p, msg)
 
 	case *RetrieveRequestMsg:
 		return p.streamer.delivery.handleRetrieveRequestMsg(p, msg)

swarm/network/stream/syncer.go

@@ -186,7 +186,6 @@ func (s *SwarmSyncerClient) NeedData(key []byte) (wait func()) {
 	chunk, _ := s.db.GetOrCreateRequest(key)
 	// TODO: we may want to request from this peer anyway even if the request exists
 	if chunk.ReqC == nil {
-		log.Error("oops this is found")
 		return nil
 	}
 	// create request and wait until the chunk data arrives and is stored

swarm/network/stream/syncer_test.go

@@ -22,6 +22,7 @@ import (
 	"fmt"
 	"io"
 	"math"
+	"runtime/debug"
 	"testing"
 	"time"
 
@@ -203,7 +204,7 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
 
 	conf.Step = &simulations.Step{
 		Action:  action,
-		Trigger: streamTesting.PivotTrigger(100*time.Millisecond, checkC, sim.IDs[0]),
+		Trigger: streamTesting.PivotTrigger(500*time.Millisecond, checkC, sim.IDs[0]),
 		Expect: &simulations.Expectation{
 			Nodes: sim.IDs[0:1],
 			Check: check,
@@ -220,6 +221,7 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
 	}
 	if result.Error != nil {
 		t.Fatalf("Simulation failed: %s", result.Error)
-	}
 		streamTesting.CheckResult(t, result, startedAt, finishedAt)
+		debug.PrintStack()
+	}
 }

swarm/storage/dbstore.go

@@ -538,6 +538,7 @@ func (s *DbStore) CurrentStorageIndex() uint64 {
 }
 
 func (s *DbStore) Put(chunk *Chunk) {
+	log.Error("DbStore.Put", "hash", chunk.Key.Hex())
 	s.lock.Lock()
 	defer s.lock.Unlock()
 
@@ -549,17 +550,23 @@ func (s *DbStore) Put(chunk *Chunk) {
 	idata, err := s.db.Get(ikey)
 	if err != nil {
 		s.doPut(chunk, ikey, &index, po)
+		batchC := s.batchC
+		go func() {
+			<-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++
 	idata = encodeIndex(&index)
 	s.batch.Put(ikey, idata)
 	select {
-	case <-s.quit:
 	case s.batchesC <- struct{}{}:
 	default:
 	}
@@ -579,13 +586,6 @@ func (s *DbStore) doPut(chunk *Chunk, ikey []byte, index *dpaDBIndex, po uint8)
 	cntKey[1] = po
 	s.batch.Put(cntKey, U64ToBytes(s.bucketCnt[po]))
 
-	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() {

swarm/storage/localstore.go

@@ -96,9 +96,9 @@ func NewTestLocalStoreForAddr(path string, basekey []byte) (*LocalStore, error)
 func (self *LocalStore) Put(chunk *Chunk) {
 	chunk.Size = int64(binary.LittleEndian.Uint64(chunk.SData[0:8]))
 	self.memStore.Put(chunk)
-	go func() {
+	log.Error("put to memstore", "hash", chunk.Key.Hex())
 	self.DbStore.Put(chunk)
-	}()
+	log.Error("put to dbstore", "hash", chunk.Key.Hex())
 }
 
 // Get(chunk *Chunk) looks up a chunk in the local stores