pot, swarm/network: merge bug, kad deadlock

pot/pot.go

@@ -23,7 +23,6 @@ import (
 )
 
 const (
-	keylen    = 256
 	maxkeylen = 256
 )
 
@@ -277,22 +276,25 @@ func Swap(t *Pot, k Val, pof Pof, f func(v Val) Val) (r *Pot, po int, found bool
 	if t.pin == nil {
 		val = f(nil)
 		if val == nil {
-			return t, t.po, false, false
+			return nil, 0, false, false
 		}
-		return NewPot(val, t.po), t.po, false, true
+		return NewPot(val, t.po), 0, false, true
 	}
 	size := t.size
 	po, found = pof(k, t.pin, t.po)
 	if found {
 		val = f(t.pin)
+		// remove element
 		if val == nil {
 			size--
 			if size == 0 {
 				r = &Pot{
 					po: t.po,
 				}
+				// return empty pot
 				return r, po, true, true
 			}
+			// actually remove pin, by merging last bin
 			i := len(t.bins) - 1
 			last := t.bins[i]
 			r = &Pot{
@@ -301,55 +303,71 @@ func Swap(t *Pot, k Val, pof Pof, f func(v Val) Val) (r *Pot, po int, found bool
 				size: size,
 				po:   t.po,
 			}
-			return r, t.po, true, true
-			// remove element
-		} else if val == t.pin {
-			return t, po, true, false
-		} else { // add element
-			r = t.clone()
-			r.pin = val
 			return r, po, true, true
 		}
+		// element found but no change
+		if val == t.pin {
+			return t, po, true, false
+		}
+		// actually modify the pinned element, but no change in structure
+		r = t.clone()
+		r.pin = val
+		return r, po, true, true
 	}
+
 	// recursive step
 	var p *Pot
 	n, i := t.getPos(po)
 	if n != nil {
 		p, po, found, change = Swap(n, k, pof, f)
+		// recursive no change
 		if !change {
 			return t, po, found, false
 		}
-		size += p.size - n.size
+		// recursive change
-	} else {
+		//size += p.size - n.size
-		val = f(nil)
+		bins := append([]*Pot{}, t.bins[:i]...)
-		if val == nil {
+		if p.size == 0 {
-			return t, po, false, false
+			size--
+		} else {
+			size += p.size - n.size
+			bins = append(bins, p)
 		}
-		if _, eq := pof(val, k, po); !eq {
+		i++
-			panic("invalid value")
+		if i < len(t.bins) {
-		}
+			bins = append(bins, t.bins[i:]...)
-		size++
-		p = &Pot{
-			pin:  val,
-			size: 1,
-			po:   po,
 		}
+		r = t.clone()
+		r.bins = bins
+		r.size = size
+		return r, po, found, true
+	}
+	// key does not exist
+	val = f(nil)
+	if val == nil {
+		// and it should not be created
+		return t, po, false, false
+	}
+	// otherwise check val if equal to k
+	if _, eq := pof(val, k, po); !eq {
+		panic("invalid value")
+	}
+	///
+	size++
+	p = &Pot{
+		pin:  val,
+		size: 1,
+		po:   po,
 	}
 
 	bins := append([]*Pot{}, t.bins[:i]...)
-	if p.pin != nil {
+	bins = append(bins, p)
-		bins = append(bins, p)
-	}
 	if i < len(t.bins) {
-		bins = append(bins, t.bins[i+1:]...)
+		bins = append(bins, t.bins[i:]...)
 	}
-	r = &Pot{
+	r = t.clone()
-		pin:  f(t.pin),
+	r.bins = bins
-		size: size,
+	r.size = size
-		po:   t.po,
-		bins: bins,
-	}
-
 	return r, po, found, true
 }
 
@@ -394,6 +412,7 @@ func union(t0, t1 *Pot, pof Pof) (*Pot, int) {
 	var bins []*Pot
 	var mis []int
 	wg := &sync.WaitGroup{}
+	wg.Add(1)
 	pin0 := t0.pin
 	pin1 := t1.pin
 	bins0 := t0.bins
@@ -445,11 +464,12 @@ func union(t0, t1 *Pot, pof Pof) (*Pot, int) {
 				bins = append(bins, nil)
 				ml := len(mis)
 				mis = append(mis, 0)
-				wg.Add(1)
+				// wg.Add(1)
-				go func(b, m int, m0, m1 *Pot) {
+				// go func(b, m int, m0, m1 *Pot) {
-					defer wg.Done()
+				// 	defer wg.Done()
-					bins[b], mis[m] = union(m0, m1, pof)
+				// bins[b], mis[m] = union(m0, m1, pof)
-				}(bl, ml, n0, n1)
+				// }(bl, ml, n0, n1)
+				bins[bl], mis[ml] = union(n0, n1, pof)
 				i0++
 				i1++
 				n0 = nil
@@ -499,6 +519,7 @@ func union(t0, t1 *Pot, pof Pof) (*Pot, int) {
 
 	}
 
+	wg.Done()
 	wg.Wait()
 	for _, c := range mis {
 		common += c
@@ -534,6 +555,9 @@ func (t *Pot) each(f func(Val, int) bool) bool {
 			return false
 		}
 	}
+	if t.size == 0 {
+		return false
+	}
 	return f(t.pin, t.po)
 }
 

pot/pot_test.go

@@ -30,26 +30,19 @@ import (
 const (
 	maxEachNeighbourTests = 420
 	maxEachNeighbour      = 420
+	maxSwap               = 420
+	maxSwapTests          = 420
 )
 
-func init() {
+// func init() {
-	// log.Root().SetHandler(log.LvlFilterHandler(log.LvlTrace, log.StreamHandler(os.Stderr, log.TerminalFormat(false))))
+// 	log.Root().SetHandler(log.LvlFilterHandler(log.LvlTrace, log.StreamHandler(os.Stderr, log.TerminalFormat(false))))
-}
+// }
 
 type testAddr struct {
 	a []byte
 	i int
 }
 
-//
-// func newTestAddr(s string, i int) *testAddr {
-// 	b := NewAddressFromString(s)
-// 	h := &common.Hash{}
-// 	copy((*h)[:], b)
-// 	a := Address(*h)
-// 	return &testAddr{&a, i}
-// }
-
 func newTestAddr(s string, i int) *testAddr {
 	return &testAddr{NewAddressFromString(s), i}
 }
@@ -60,7 +53,6 @@ func (a *testAddr) Address() []byte {
 
 func (a *testAddr) String() string {
 	return Label(a.a)
-	// return a.Address.String()[:keylen]
 }
 
 func randomTestAddr(n int, i int) *testAddr {
@@ -86,12 +78,10 @@ func indexes(t *Pot) (i []int, po []int) {
 func testAdd(t *Pot, pof Pof, j int, values ...string) (_ *Pot, n int, f bool) {
 	for i, val := range values {
 		t, n, f = Add(t, newTestAddr(val, i+j), pof)
-		// t.Add(newTestAddr(val, i+n))
 	}
 	return t, n, f
 }
 
-// func RandomBoolAddress()
 func TestPotAdd(t *testing.T) {
 	pof := DefaultPof(8)
 	n := NewPot(newTestAddr("00111100", 0), 0)
@@ -128,7 +118,6 @@ func TestPotAdd(t *testing.T) {
 	}
 }
 
-// func RandomBoolAddress()
 func TestPotRemove(t *testing.T) {
 	pof := DefaultPof(8)
 	n := NewPot(newTestAddr("00111100", 0), 0)
@@ -168,61 +157,69 @@ func TestPotRemove(t *testing.T) {
 }
 
 func TestPotSwap(t *testing.T) {
-	pof := DefaultPof(8)
+	for i := 0; i < maxSwapTests; i++ {
-	max := maxEachNeighbour
+		// alen := maxkeylen
-	n := NewPot(nil, 0)
+		alen := maxkeylen
-	var m []*testAddr
+		pof := DefaultPof(alen)
-	var found bool
+		max := rand.Intn(maxSwap)
-	for j := 0; j < 2*max; {
+
-		v := randomtestAddr(keylen, j)
+		n := NewPot(nil, 0)
-		n, _, found = Add(n, v, pof)
+		var m []*testAddr
-		if !found {
+		var found bool
-			m = append(m, v)
+		for j := 0; j < 2*max; {
-			j++
+			v := randomtestAddr(alen, j)
-		}
+			n, _, found = Add(n, v, pof)
-	}
+			if !found {
-	k := make(map[string]*testAddr)
+				m = append(m, v)
-	for j := 0; j < max; {
+				j++
-		v := randomtestAddr(keylen, 1)
-		_, found := k[Label(v)]
-		if !found {
-			k[Label(v)] = v
-			j++
-		}
-	}
-	for _, v := range k {
-		m = append(m, v)
-	}
-	f := func(v Val) Val {
-		tv := v.(*testAddr)
-		if tv.i < max {
-			return nil
-		}
-		tv.i = 0
-		return v
-	}
-	for _, val := range m {
-		n, _, _, _ = Swap(n, val, pof, func(v Val) Val {
-			if v == nil {
-				return val
 			}
-			return f(v)
-		})
-	}
-	sum := 0
-	n.Each(func(v Val, i int) bool {
-		sum++
-		tv := v.(*testAddr)
-		if tv.i > 1 {
-			t.Fatalf("item value incorrect, expected 0, got %v", tv.i)
 		}
-		return true
+		k := make(map[string]*testAddr)
-	})
+		for j := 0; j < max; {
-	if sum != 2*max {
+			v := randomtestAddr(alen, 1)
-		t.Fatalf("incorrect number of elements. expected %v, got %v", 2*max, sum)
+			_, found := k[Label(v)]
-	}
+			if !found {
-	if sum != n.Size() {
+				k[Label(v)] = v
-		t.Fatalf("incorrect size. expected %v, got %v", sum, n.Size())
+				j++
+			}
+		}
+		for _, v := range k {
+			m = append(m, v)
+		}
+		f := func(v Val) Val {
+			tv := v.(*testAddr)
+			if tv.i < max {
+				return nil
+			}
+			tv.i = 0
+			return v
+		}
+		for _, val := range m {
+			n, _, _, _ = Swap(n, val, pof, func(v Val) Val {
+				if v == nil {
+					return val
+				}
+				return f(v)
+			})
+		}
+		sum := 0
+		n.Each(func(v Val, i int) bool {
+			if v == nil {
+				return true
+			}
+			sum++
+			tv := v.(*testAddr)
+			if tv.i > 1 {
+				t.Fatalf("item value incorrect, expected 0, got %v", tv.i)
+			}
+			return true
+		})
+		if sum != 2*max {
+			t.Fatalf("incorrect number of elements. expected %v, got %v", 2*max, sum)
+		}
+		if sum != n.Size() {
+			t.Fatalf("incorrect size. expected %v, got %v", sum, n.Size())
+		}
 	}
 }
 
@@ -238,7 +235,7 @@ func checkPo(val Val, pof Pof) func(Val, int) error {
 }
 
 func checkOrder(val Val) func(Val, int) error {
-	po := keylen
+	po := maxkeylen
 	return func(v Val, p int) error {
 		if po < p {
 			return fmt.Errorf("incorrect order for item %v in neighbour iteration for %v. PO %v > %v (previous max)", v, val, p, po)
@@ -291,31 +288,15 @@ const (
 	mergeTestChoose = 5
 )
 
-//
-// func TestPotMergeOne(t *testing.T) {
-// 	pot1 := NewPot(nil, 0, DefaultPof(2))
-// 	pot1.Add(newTestAddr("10", 0))
-// 	pot1.Add(newTestAddr("00", 0))
-// 	pot2 := NewPot(nil, 0, DefaultPof(2))
-// 	pot2.Add(newTestAddr("01", 0))
-// 	pot1.Merge(pot2)
-// 	count := 0
-// 	pot1.Each(func(val Val, i int) bool {
-// 		count++
-// 		return true
-// 	})
-// 	if count != 3 {
-// 		t.Fatalf("expected count to be 3, got %d\n%v\n%v", count, pot2, pot1)
-// 	}
-// }
-
 func TestPotMergeCommon(t *testing.T) {
 	vs := make([]*testAddr, mergeTestCount)
-	pof := DefaultPof(maxkeylen)
 	for i := 0; i < maxEachNeighbourTests; i++ {
+		alen := maxkeylen
+		// alen := maxkeylen
+		pof := DefaultPof(alen)
 
 		for j := 0; j < len(vs); j++ {
-			vs[j] = randomtestAddr(keylen, j)
+			vs[j] = randomtestAddr(alen, j)
 		}
 		max0 := rand.Intn(mergeTestChoose) + 1
 		max1 := rand.Intn(mergeTestChoose) + 1
@@ -369,17 +350,18 @@ func TestPotMergeCommon(t *testing.T) {
 			t.Fatalf("%v: merged pot contains duplicates: \n%v", i, n)
 		}
 		for k := range m {
-			n, _, found := Add(n, newTestAddr(k, 0), pof)
+			_, _, found = Add(n, newTestAddr(k, 0), pof)
 			if !found {
-				t.Fatalf("%v: merged pot (size:%v, added: %v) missing element %v\n%v", i, size, added, k, n)
+				t.Fatalf("%v: merged pot (size:%v, added: %v) missing element %v", i, size, added, k)
 			}
 		}
 	}
 }
 
 func TestPotMergeScale(t *testing.T) {
-	pof := DefaultPof(maxkeylen)
 	for i := 0; i < maxEachNeighbourTests; i++ {
+		alen := maxkeylen
+		pof := DefaultPof(alen)
 		max0 := rand.Intn(maxEachNeighbour) + 1
 		max1 := rand.Intn(maxEachNeighbour) + 1
 		n0 := NewPot(nil, 0)
@@ -388,10 +370,8 @@ func TestPotMergeScale(t *testing.T) {
 		m := make(map[string]bool)
 		var found bool
 		for j := 0; j < max0; {
-			v := randomtestAddr(keylen, j)
+			v := randomtestAddr(alen, j)
-			// v := randomtestAddr(keylen, j)
 			n0, _, found = Add(n0, v, pof)
-			// _, found := n0.Add(v)
 			if !found {
 				m[Label(v)] = false
 				j++
@@ -400,10 +380,8 @@ func TestPotMergeScale(t *testing.T) {
 		expAdded := 0
 
 		for j := 0; j < max1; {
-			v := randomtestAddr(keylen, j)
+			v := randomtestAddr(alen, j)
-			// v := randomtestAddr(keylen, j)
 			n1, _, found = Add(n1, v, pof)
-			// _, found := n1.Add(v)
 			if !found {
 				j++
 			}
@@ -425,18 +403,18 @@ func TestPotMergeScale(t *testing.T) {
 		size := n.Size()
 
 		if expSize != size {
-			t.Fatalf("%v: incorrect number of elements in merged pot, expected %v, got %v\n%v", i, expSize, size, n)
+			t.Fatalf("%v: incorrect number of elements in merged pot, expected %v, got %v", i, expSize, size)
 		}
 		if expAdded != added {
 			t.Fatalf("%v: incorrect number of added elements in merged pot, expected %v, got %v", i, expAdded, added)
 		}
 		if !checkDuplicates(n) {
-			t.Fatalf("%v: merged pot contains duplicates: \n%v", i, n)
+			t.Fatalf("%v: merged pot contains duplicates", i)
 		}
 		for k := range m {
-			n, _, found := Add(n, newTestAddr(k, 0), pof)
+			_, _, found = Add(n, newTestAddr(k, 0), pof)
 			if !found {
-				t.Fatalf("%v: merged pot (size:%v, added: %v) missing element %v\n%v", i, size, added, k, n)
+				t.Fatalf("%v: merged pot (size:%v, added: %v) missing element %v", i, size, added, k)
 			}
 		}
 	}
@@ -457,15 +435,16 @@ func checkDuplicates(t *Pot) bool {
 }
 
 func TestPotEachNeighbourSync(t *testing.T) {
-	pof := DefaultPof(maxkeylen)
 	for i := 0; i < maxEachNeighbourTests; i++ {
+		alen := maxkeylen
+		pof := DefaultPof(maxkeylen)
 		max := rand.Intn(maxEachNeighbour/2) + maxEachNeighbour/2
-		pin := randomTestAddr(keylen, 0)
+		pin := randomTestAddr(alen, 0)
 		n := NewPot(pin, 0)
 		m := make(map[string]bool)
 		m[Label(pin)] = false
 		for j := 1; j <= max; j++ {
-			v := randomTestAddr(keylen, j)
+			v := randomTestAddr(alen, j)
 			n, _, _ = Add(n, v, pof)
 			m[Label(v)] = false
 		}
@@ -475,13 +454,13 @@ func TestPotEachNeighbourSync(t *testing.T) {
 			continue
 		}
 		count := rand.Intn(size/2) + size/2
-		val := randomTestAddr(keylen, max+1)
+		val := randomTestAddr(alen, max+1)
 		log.Trace(fmt.Sprintf("%v: pin: %v, size: %v, val: %v, count: %v", i, n.Pin(), size, val, count))
 		err := testPotEachNeighbour(n, pof, val, count, checkPo(val, pof), checkOrder(val), checkValues(m, val))
 		if err != nil {
 			t.Fatal(err)
 		}
-		minPoFound := keylen
+		minPoFound := alen
 		maxPoNotFound := 0
 		for k, found := range m {
 			po, _ := pof(val, newTestAddr(k, 0), 0)
@@ -502,14 +481,15 @@ func TestPotEachNeighbourSync(t *testing.T) {
 }
 
 func TestPotEachNeighbourAsync(t *testing.T) {
-	pof := DefaultPof(maxkeylen)
 	for i := 0; i < maxEachNeighbourTests; i++ {
 		max := rand.Intn(maxEachNeighbour/2) + maxEachNeighbour/2
-		n := NewPot(randomTestAddr(keylen, 0), 0)
+		alen := maxkeylen
+		pof := DefaultPof(alen)
+		n := NewPot(randomTestAddr(alen, 0), 0)
 		size := 1
 		var found bool
 		for j := 1; j <= max; j++ {
-			v := randomTestAddr(keylen, j)
+			v := randomTestAddr(alen, j)
 			n, _, found = Add(n, v, pof)
 			if !found {
 				size++
@@ -522,11 +502,11 @@ func TestPotEachNeighbourAsync(t *testing.T) {
 			continue
 		}
 		count := rand.Intn(size/2) + size/2
-		val := randomTestAddr(keylen, max+1)
+		val := randomTestAddr(alen, max+1)
 
 		mu := sync.Mutex{}
 		m := make(map[string]bool)
-		maxPos := rand.Intn(keylen)
+		maxPos := rand.Intn(alen)
 		log.Trace(fmt.Sprintf("%v: pin: %v, size: %v, val: %v, count: %v, maxPos: %v", i, n.Pin(), size, val, count, maxPos))
 		msize := 0
 		remember := func(v Val, po int) error {
@@ -561,12 +541,13 @@ func TestPotEachNeighbourAsync(t *testing.T) {
 
 func benchmarkEachNeighbourSync(t *testing.B, max, count int, d time.Duration) {
 	t.ReportAllocs()
-	pof := DefaultPof(maxkeylen)
+	alen := maxkeylen
-	pin := randomTestAddr(keylen, 0)
+	pof := DefaultPof(alen)
+	pin := randomTestAddr(alen, 0)
 	n := NewPot(pin, 0)
 	var found bool
 	for j := 1; j <= max; {
-		v := randomTestAddr(keylen, j)
+		v := randomTestAddr(alen, j)
 		n, _, found = Add(n, v, pof)
 		if !found {
 			j++
@@ -574,7 +555,7 @@ func benchmarkEachNeighbourSync(t *testing.B, max, count int, d time.Duration) {
 	}
 	t.ResetTimer()
 	for i := 0; i < t.N; i++ {
-		val := randomTestAddr(keylen, max+1)
+		val := randomTestAddr(alen, max+1)
 		m := 0
 		n.EachNeighbour(val, pof, func(v Val, po int) bool {
 			time.Sleep(d)
@@ -588,17 +569,17 @@ func benchmarkEachNeighbourSync(t *testing.B, max, count int, d time.Duration) {
 	t.StopTimer()
 	stats := new(runtime.MemStats)
 	runtime.ReadMemStats(stats)
-	// fmt.Println(stats.Sys)
 }
 
 func benchmarkEachNeighbourAsync(t *testing.B, max, count int, d time.Duration) {
 	t.ReportAllocs()
-	pof := DefaultPof(maxkeylen)
+	alen := maxkeylen
-	pin := randomTestAddr(keylen, 0)
+	pof := DefaultPof(alen)
+	pin := randomTestAddr(alen, 0)
 	n := NewPot(pin, 0)
 	var found bool
 	for j := 1; j <= max; {
-		v := randomTestAddr(keylen, j)
+		v := randomTestAddr(alen, j)
 		n, _, found = Add(n, v, pof)
 		if !found {
 			j++
@@ -606,15 +587,14 @@ func benchmarkEachNeighbourAsync(t *testing.B, max, count int, d time.Duration)
 	}
 	t.ResetTimer()
 	for i := 0; i < t.N; i++ {
-		val := randomTestAddr(keylen, max+1)
+		val := randomTestAddr(alen, max+1)
-		n.EachNeighbourAsync(val, pof, count, keylen, func(v Val, po int) {
+		n.EachNeighbourAsync(val, pof, count, alen, func(v Val, po int) {
 			time.Sleep(d)
 		}, true)
 	}
 	t.StopTimer()
 	stats := new(runtime.MemStats)
 	runtime.ReadMemStats(stats)
-	// fmt.Println(stats.Sys)
 }
 
 func BenchmarkEachNeighbourSync_3_1_0(t *testing.B) {

swarm/network/hive_test.go

@@ -17,7 +17,7 @@ func newHiveTester(t *testing.T, params *HiveParams) (*bzzTester, *Hive) {
 }
 
 func TestRegisterAndConnect(t *testing.T) {
-	t.Skip("deadlocked")
+	//t.Skip("deadlocked")
 	params := NewHiveParams()
 	s, pp := newHiveTester(t, params)
 	defer s.Stop()
@@ -47,6 +47,11 @@ func TestRegisterAndConnect(t *testing.T) {
 	s.TestExchanges(p2ptest.Exchange{
 		Label: "getPeersMsg message",
 		Expects: []p2ptest.Expect{
+			p2ptest.Expect{
+				Code: 2,
+				Msg:  &subPeersMsg{0},
+				Peer: id,
+			},
 			p2ptest.Expect{
 				Code: 1,
 				Msg:  &getPeersMsg{uint8(o), 5},