From e8a5843903cff76d75ec9a4444692ed92b3bd6b5 Mon Sep 17 00:00:00 2001 From: Zsolt Felfoldi Date: Wed, 23 Jan 2019 18:06:47 +0100 Subject: [PATCH] les: fixed api sim test --- les/api.go | 31 +++++++++----- les/api_test.go | 100 +++++++++++++++++++++++++++++++++------------ les/freeclient.go | 10 +++-- les/helper_test.go | 12 +----- les/peer.go | 2 +- les/server.go | 4 +- 6 files changed, 105 insertions(+), 54 deletions(-) diff --git a/les/api.go b/les/api.go index 86fa8887cf..e876278c27 100644 --- a/les/api.go +++ b/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 - child clientPool - ps *peerSet - clients map[enode.ID]priorityClientInfo - totalCap, totalConnectedCap, freeClientCap uint64 - maxPeers, priorityCount int + lock sync.Mutex + child clientPool + ps *peerSet + clients map[enode.ID]priorityClientInfo + totalCap, totalCapAnnounced uint64 + 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 { diff --git a/les/api_test.go b/les/api_test.go index 2371da4035..420f30efe8 100644 --- a/les/api_test.go +++ b/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) - fmt.Printf("Server totalCap: %d minCap: %d head number: %d head hash: %064x\n", totalCap, minCap, headNum, headHash) - reqMinCap := uint64(float64(totalCap) * minRelCap / (minRelCap + float64(len(clients)-1))) + totalCap := getTotalCap(ctx, t, serverRpcClient) + minCap := getMinCap(ctx, t, serverRpcClient) + 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) - go func() { - testRequest(ctx, t, c) + select { + case <-stop: wg.Done() - <-queue - count++ - atomic.StoreUint64(&reqCount[i], count) - }() + return + case <-ctx.Done(): + 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 { - t.Fatalf("Failed to obtain Merkle proof: %v", err) + err := client.CallContext(c, &res, "eth_getBalance", addr, "latest") + 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) { diff --git a/les/freeclient.go b/les/freeclient.go index d11231af48..8e02df0369 100644 --- a/les/freeclient.go +++ b/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 diff --git a/les/helper_test.go b/les/helper_test.go index 053f752de1..a8097708ee 100644 --- a/les/helper_test.go +++ b/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) diff --git a/les/peer.go b/les/peer.go index 6572370ba3..28e81b952b 100644 --- a/les/peer.go +++ b/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() diff --git a/les/server.go b/les/server.go index ec61b9e05b..962e20a050 100644 --- a/les/server.go +++ b/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)