les: fixed api sim test

les/api.go

@@ -91,6 +91,10 @@ func (api *PrivateLesServerAPI) MinimumCapacity() hexutil.Uint64 {
 	return hexutil.Uint64(minCapacity)
 }
 
+func (api *PrivateLesServerAPI) FreeClientCapacity() hexutil.Uint64 {
+	return hexutil.Uint64(api.server.freeClientCap)
+}
+
 type clientPool interface {
 	peerSetNotify
 	setLimits(count int, totalCap uint64)
@@ -98,12 +102,13 @@ type clientPool interface {
 
 // priorityClientPool stores information about prioritized clients
 type priorityClientPool struct {
-	lock                                       sync.Mutex
+	lock                             sync.Mutex
-	child                                      clientPool
+	child                            clientPool
-	ps                                         *peerSet
+	ps                               *peerSet
-	clients                                    map[enode.ID]priorityClientInfo
+	clients                          map[enode.ID]priorityClientInfo
-	totalCap, totalConnectedCap, freeClientCap uint64
+	totalCap, totalCapAnnounced      uint64
-	maxPeers, priorityCount                    int
+	totalConnectedCap, freeClientCap uint64
+	maxPeers, priorityCount          int
 
 	subs            tcSubs
 	updateSchedule  []scheduledUpdate
@@ -174,7 +179,7 @@ func (v *priorityClientPool) registerPeer(p *peer) {
 		v.child.registerPeer(p)
 	}
 	if c.cap != 0 && v.totalConnectedCap+c.cap > v.totalCap {
-		v.ps.Unregister(p.id)
+		go v.ps.Unregister(p.id)
 		return
 	}
 
@@ -222,6 +227,7 @@ func (v *priorityClientPool) setLimits(count int, totalCap uint64) {
 	v.lock.Lock()
 	defer v.lock.Unlock()
 
+	v.totalCapAnnounced = totalCap
 	if totalCap > v.totalCap {
 		v.setLimitsNow(count, totalCap)
 		v.subs.send(totalCap, false)
@@ -275,7 +281,7 @@ func (v *priorityClientPool) setLimitsNow(count int, totalCap uint64) {
 				v.totalConnectedCap -= c.cap
 				v.priorityCount--
 				v.clients[id] = c
-				v.ps.Unregister(c.peer.id)
+				go v.ps.Unregister(c.peer.id)
 				if v.priorityCount <= count && v.totalConnectedCap <= totalCap {
 					break
 				}
@@ -294,7 +300,7 @@ func (v *priorityClientPool) totalCapacity() uint64 {
 	v.lock.Lock()
 	defer v.lock.Unlock()
 
-	return v.totalCap
+	return v.totalCapAnnounced
 }
 
 func (v *priorityClientPool) subscribeTotalCapacity(sub *tcSubscription) {
@@ -321,7 +327,9 @@ func (v *priorityClientPool) setClientCapacity(id enode.ID, cap uint64) error {
 				v.child.unregisterPeer(c.peer)
 			}
 			v.priorityCount++
-			c.peer.updateCapacity(cap)
+		}
+		if cap == 0 {
+			v.priorityCount--
 		}
 		v.totalConnectedCap += cap - c.cap
 		if v.child != nil {
@@ -331,8 +339,9 @@ func (v *priorityClientPool) setClientCapacity(id enode.ID, cap uint64) error {
 			if v.child != nil {
 				v.child.registerPeer(c.peer)
 			}
-			v.priorityCount--
 			c.peer.updateCapacity(v.freeClientCap)
+		} else {
+			c.peer.updateCapacity(cap)
 		}
 	}
 	if cap != 0 || c.connected {

les/api_test.go

@@ -84,7 +84,7 @@ func testCapacityAPI(t *testing.T, clientCount int) {
 		return
 	}
 
-	testSim(t, 1, clientCount, []string{testServerDataDir}, nil, func(ctx context.Context, net *simulations.Network, servers []*simulations.Node, clients []*simulations.Node) {
+	for !testSim(t, 1, clientCount, []string{testServerDataDir}, nil, func(ctx context.Context, net *simulations.Network, servers []*simulations.Node, clients []*simulations.Node) bool {
 		if len(servers) != 1 {
 			t.Fatalf("Invalid number of servers: %d", len(servers))
 		}
@@ -97,15 +97,17 @@ func testCapacityAPI(t *testing.T, clientCount int) {
 			t.Fatalf("Failed to obtain rpc client: %v", err)
 		}
 		headNum, headHash := getHead(ctx, t, serverRpcClient)
-		totalCap, minCap := capacityLimits(ctx, t, serverRpcClient)
+		totalCap := getTotalCap(ctx, t, serverRpcClient)
-		fmt.Printf("Server totalCap: %d  minCap: %d  head number: %d  head hash: %064x\n", totalCap, minCap, headNum, headHash)
+		minCap := getMinCap(ctx, t, serverRpcClient)
-		reqMinCap := uint64(float64(totalCap) * minRelCap / (minRelCap + float64(len(clients)-1)))
+		testCap := totalCap * 3 / 4
+		fmt.Printf("Server testCap: %d  minCap: %d  head number: %d  head hash: %064x\n", testCap, minCap, headNum, headHash)
+		reqMinCap := uint64(float64(testCap) * minRelCap / (minRelCap + float64(len(clients)-1)))
 		if minCap > reqMinCap {
 			t.Fatalf("Minimum client capacity (%d) bigger than required minimum for this test (%d)", minCap, reqMinCap)
 		}
 
 		freeIdx := rand.Intn(len(clients))
-		freeCap := totalCap / testMaxClients
+		freeCap := getFreeCap(ctx, t, serverRpcClient)
 
 		for i, client := range clients {
 			var err error
@@ -116,7 +118,7 @@ func testCapacityAPI(t *testing.T, clientCount int) {
 
 			fmt.Println("connecting client", i)
 			if i != freeIdx {
-				setCapacity(ctx, t, serverRpcClient, client.ID(), totalCap/uint64(len(clients)))
+				setCapacity(ctx, t, serverRpcClient, client.ID(), testCap/uint64(len(clients)))
 			}
 			net.Connect(client.ID(), server.ID())
 
@@ -139,6 +141,7 @@ func testCapacityAPI(t *testing.T, clientCount int) {
 		stop := make(chan struct{})
 
 		reqCount := make([]uint64, len(clientRpcClients))
+
 		for i, c := range clientRpcClients {
 			wg.Add(1)
 			i, c := i, c
@@ -148,14 +151,25 @@ func testCapacityAPI(t *testing.T, clientCount int) {
 				for {
 					select {
 					case queue <- struct{}{}:
-						wg.Add(1)
+						select {
-						go func() {
+						case <-stop:
-							testRequest(ctx, t, c)
 							wg.Done()
-							<-queue
+							return
-							count++
+						case <-ctx.Done():
-							atomic.StoreUint64(&reqCount[i], count)
+							wg.Done()
-						}()
+							return
+						default:
+							wg.Add(1)
+							go func() {
+								ok := testRequest(ctx, t, c)
+								wg.Done()
+								<-queue
+								if ok {
+									count++
+									atomic.StoreUint64(&reqCount[i], count)
+								}
+							}()
+						}
 					case <-stop:
 						wg.Done()
 						return
@@ -192,7 +206,7 @@ func testCapacityAPI(t *testing.T, clientCount int) {
 				sum += weights[i]
 			}
 			for i, client := range clients {
-				weights[i] *= float64(totalCap-freeCap-100) / sum
+				weights[i] *= float64(testCap-freeCap-100) / sum
 				capacity := uint64(weights[i])
 				if i != freeIdx && capacity < getCapacity(ctx, t, serverRpcClient, client.ID()) {
 					setCapacity(ctx, t, serverRpcClient, client.ID(), capacity)
@@ -207,11 +221,16 @@ func testCapacityAPI(t *testing.T, clientCount int) {
 			}
 			weights[freeIdx] = float64(freeCap)
 			for i, _ := range clients {
-				weights[i] /= float64(totalCap)
+				weights[i] /= float64(testCap)
 			}
 
 			time.Sleep(flowcontrol.DecParamDelay)
-			fmt.Println("starting measurement")
+			fmt.Println("Starting measurement")
+			fmt.Printf("Relative weights:")
+			for i, _ := range clients {
+				fmt.Printf("  %f", weights[i])
+			}
+			fmt.Println()
 			start := processedSince(nil)
 			for {
 				select {
@@ -220,6 +239,14 @@ func testCapacityAPI(t *testing.T, clientCount int) {
 				default:
 				}
 
+				totalCap = getTotalCap(ctx, t, serverRpcClient)
+				if totalCap < testCap {
+					fmt.Println("Total capacity underrun")
+					close(stop)
+					wg.Wait()
+					return false
+				}
+
 				processed := processedSince(start)
 				var avg uint64
 				fmt.Printf("Processed")
@@ -242,7 +269,7 @@ func testCapacityAPI(t *testing.T, clientCount int) {
 							maxDev = dev
 						}
 					}
-					fmt.Printf("  max deviation: %f\n", maxDev)
+					fmt.Printf("  max deviation: %f  totalCap: %d\n", maxDev, totalCap)
 					if maxDev <= testTolerance {
 						fmt.Println("success")
 						break
@@ -260,7 +287,10 @@ func testCapacityAPI(t *testing.T, clientCount int) {
 		for i, count := range reqCount {
 			fmt.Println("client", i, "processed", count)
 		}
-	})
+		return true
+	}) {
+		fmt.Println("restarting test")
+	}
 }
 
 func getHead(ctx context.Context, t *testing.T, client *rpc.Client) (uint64, common.Hash) {
@@ -284,15 +314,18 @@ func getHead(ctx context.Context, t *testing.T, client *rpc.Client) (uint64, com
 	return num, hash
 }
 
-func testRequest(ctx context.Context, t *testing.T, client *rpc.Client) {
+func testRequest(ctx context.Context, t *testing.T, client *rpc.Client) bool {
 	//res := make(map[string]interface{})
 	var res string
 	var addr common.Address
 	rand.Read(addr[:])
+	c, _ := context.WithTimeout(ctx, time.Second*12)
 	//	if err := client.CallContext(ctx, &res, "eth_getProof", addr, nil, "latest"); err != nil {
-	if err := client.CallContext(ctx, &res, "eth_getBalance", addr, "latest"); err != nil {
+	err := client.CallContext(c, &res, "eth_getBalance", addr, "latest")
-		t.Fatalf("Failed to obtain Merkle proof: %v", err)
+	if err != nil {
+		fmt.Println("request error:", err)
 	}
+	return err == nil
 }
 
 func setCapacity(ctx context.Context, t *testing.T, server *rpc.Client, clientID enode.ID, cap uint64) {
@@ -313,7 +346,7 @@ func getCapacity(ctx context.Context, t *testing.T, server *rpc.Client, clientID
 	return cap
 }
 
-func capacityLimits(ctx context.Context, t *testing.T, server *rpc.Client) (uint64, uint64) {
+func getTotalCap(ctx context.Context, t *testing.T, server *rpc.Client) uint64 {
 	var s string
 	if err := server.CallContext(ctx, &s, "les_totalCapacity"); err != nil {
 		t.Fatalf("Failed to query total capacity: %v", err)
@@ -322,6 +355,11 @@ func capacityLimits(ctx context.Context, t *testing.T, server *rpc.Client) (uint
 	if err != nil {
 		t.Fatalf("Failed to decode total capacity: %v", err)
 	}
+	return total
+}
+
+func getMinCap(ctx context.Context, t *testing.T, server *rpc.Client) uint64 {
+	var s string
 	if err := server.CallContext(ctx, &s, "les_minimumCapacity"); err != nil {
 		t.Fatalf("Failed to query minimum capacity: %v", err)
 	}
@@ -329,7 +367,19 @@ func capacityLimits(ctx context.Context, t *testing.T, server *rpc.Client) (uint
 	if err != nil {
 		t.Fatalf("Failed to decode minimum capacity: %v", err)
 	}
-	return total, min
+	return min
+}
+
+func getFreeCap(ctx context.Context, t *testing.T, server *rpc.Client) uint64 {
+	var s string
+	if err := server.CallContext(ctx, &s, "les_freeClientCapacity"); err != nil {
+		t.Fatalf("Failed to query free client capacity: %v", err)
+	}
+	free, err := hexutil.DecodeUint64(s)
+	if err != nil {
+		t.Fatalf("Failed to decode free client capacity: %v", err)
+	}
+	return free
 }
 
 func init() {
@@ -396,7 +446,7 @@ func NewAdapter(adapterType string, services adapters.Services) (adapter adapter
 	return adapter, teardown, nil
 }
 
-func testSim(t *testing.T, serverCount, clientCount int, serverDir, clientDir []string, test func(ctx context.Context, net *simulations.Network, servers []*simulations.Node, clients []*simulations.Node)) {
+func testSim(t *testing.T, serverCount, clientCount int, serverDir, clientDir []string, test func(ctx context.Context, net *simulations.Network, servers []*simulations.Node, clients []*simulations.Node) bool) bool {
 	net, teardown, err := NewNetwork()
 	defer teardown()
 	if err != nil {
@@ -446,7 +496,7 @@ func testSim(t *testing.T, serverCount, clientCount int, serverDir, clientDir []
 		}
 	}
 
-	test(ctx, net, servers, clients)
+	return test(ctx, net, servers, clients)
 }
 
 func newLesClientService(ctx *adapters.ServiceContext) (node.Service, error) {

les/freeclient.go

@@ -18,6 +18,7 @@
 package les
 
 import (
+	"fmt"
 	"io"
 	"math"
 	"net"
@@ -108,7 +109,7 @@ func (f *freeClientPool) connect(address, id string) bool {
 
 	if f.connectedLimit == 0 {
 		log.Debug("Client rejected", "address", address)
-		f.removePeer(id)
+		go f.removePeer(id)
 		return false
 	}
 	e := f.addressMap[address]
@@ -120,7 +121,7 @@ func (f *freeClientPool) connect(address, id string) bool {
 	} else {
 		if e.connected {
 			log.Debug("Client already connected", "address", address)
-			f.removePeer(id)
+			go f.removePeer(id)
 			return false
 		}
 		recentUsage = int64(math.Exp(float64(e.logUsage-f.logOffset(now)) / fixedPointMultiplier))
@@ -136,7 +137,7 @@ func (f *freeClientPool) connect(address, id string) bool {
 			// keep the old client and reject the new one
 			f.connPool.Push(i, i.linUsage)
 			log.Debug("Client rejected", "address", address)
-			f.removePeer(id)
+			go f.removePeer(id)
 			return false
 		}
 	}
@@ -188,6 +189,7 @@ func (f *freeClientPool) setLimits(count int, totalCap uint64) {
 	f.lock.Lock()
 	defer f.lock.Unlock()
 
+	fmt.Println("setLimits", count, totalCap, f.freeClientCap)
 	f.connectedLimit = int(totalCap / f.freeClientCap)
 	if count < f.connectedLimit {
 		f.connectedLimit = count
@@ -207,7 +209,7 @@ func (f *freeClientPool) dropClient(i *freeClientPoolEntry, now mclock.AbsTime)
 	i.connected = false
 	f.disconnPool.Push(i, -i.logUsage)
 	log.Debug("Client kicked out", "address", i.address)
-	f.removePeer(i.id)
+	go f.removePeer(i.id)
 }
 
 // logOffset calculates the time-dependent offset for the logarithmic

les/helper_test.go

@@ -134,16 +134,6 @@ func testIndexers(db ethdb.Database, odr light.OdrBackend, iConfig *light.Indexe
 	return chtIndexer, bloomIndexer, bloomTrieIndexer
 }
 
-func testRCL() RequestCostList {
-	cl := make(RequestCostList, len(reqBenchMap))
-	for i, req := range reqBenchMap {
-		cl[i].MsgCode = req.code
-		cl[i].BaseCost = 0
-		cl[i].ReqCost = 0
-	}
-	return cl
-}
-
 // newTestProtocolManager creates a new protocol manager for testing purposes,
 // with the given number of blocks already known, potential notification
 // channels for different events and relative chain indexers array.
@@ -305,7 +295,7 @@ func (p *testPeer) handshake(t *testing.T, td *big.Int, head common.Hash, headNu
 	expList = expList.add("txRelay", nil)
 	expList = expList.add("flowControl/BL", testBufLimit)
 	expList = expList.add("flowControl/MRR", uint64(1))
-	expList = expList.add("flowControl/MRC", testRCL())
+	expList = expList.add("flowControl/MRC", testCostList())
 
 	if err := p2p.ExpectMsg(p.app, StatusMsg, expList); err != nil {
 		t.Fatalf("status recv: %v", err)

les/peer.go

@@ -523,7 +523,7 @@ func (p *peer) Handshake(td *big.Int, head common.Hash, headNum uint64, genesis
 		if server.costTracker != nil {
 			costList = server.costTracker.makeCostList()
 		} else {
-			costList = testRCL()
+			costList = testCostList()
 		}
 		send = send.add("flowControl/MRC", costList)
 		p.fcCosts = costList.decode()

les/server.go

@@ -185,7 +185,7 @@ func (s *LesServer) Start(srvr *p2p.Server) {
 	s.maxPeers = s.config.LightPeers
 	totalRecharge := s.costTracker.totalRecharge()
 	if s.maxPeers > 0 {
-		s.freeClientCap = totalRecharge / uint64(s.maxPeers)
+		s.freeClientCap = minCapacity //totalRecharge / uint64(s.maxPeers)
 		if s.freeClientCap < minCapacity {
 			s.freeClientCap = minCapacity
 		}
@@ -198,7 +198,7 @@ func (s *LesServer) Start(srvr *p2p.Server) {
 	}
 	freePeers := int(totalRecharge / s.freeClientCap)
 	if freePeers < s.maxPeers {
-		log.Warn("Light peer count limited", "specified", s.maxPeers, "allowed", freePeers) //qqq
+		log.Warn("Light peer count limited", "specified", s.maxPeers, "allowed", freePeers)
 	}
 
 	s.freeClientPool = newFreeClientPool(s.chainDb, s.freeClientCap, 10000, mclock.System{}, s.protocolManager.removePeer)