diff --git a/les/api.go b/les/api.go index e6979acf27..1b5c14365e 100644 --- a/les/api.go +++ b/les/api.go @@ -24,8 +24,8 @@ import ( ) var ( - ErrMinBW = errors.New("capacity too small") - ErrTotalBW = errors.New("total capacity exceeded") + ErrMinCap = errors.New("capacity too small") + ErrTotalCap = errors.New("total capacity exceeded") ) // PublicLesServerAPI provides an API to access the les server. @@ -40,7 +40,7 @@ type PrivateLesServerAPI struct { func NewPrivateLesServerAPI(server *LesServer) *PrivateLesServerAPI { vip := &vipClientPool{ clients: make(map[enode.ID]vipClientInfo), - totalBw: server.totalCapacity, + totalCap: server.totalCapacity, pm: server.protocolManager, } server.protocolManager.vipClientPool = vip @@ -66,15 +66,15 @@ type vipClientPool struct { lock sync.Mutex pm *ProtocolManager clients map[enode.ID]vipClientInfo - totalBw, totalVipBw, totalConnectedBw uint64 + totalCap, totalVipCap, totalConnectedCap uint64 vipCount int } // vipClientInfo entries exist for all prioritized clients and currently connected free clients type vipClientInfo struct { - bw uint64 // zero for non-vip clients + cap uint64 // zero for non-vip clients connected bool - updateBw func(uint64) + updateCap func(uint64) } // SetClientCapacity sets the priority capacity assigned to a given client. @@ -84,37 +84,37 @@ type vipClientInfo struct { // // Note: assigned capacity can be changed while the client is connected with // immediate effect. -func (api *PrivateLesServerAPI) SetClientCapacity(id enode.ID, bw uint64) error { - if bw != 0 && bw < api.server.minCapacity { - return ErrMinBW +func (api *PrivateLesServerAPI) SetClientCapacity(id enode.ID, cap uint64) error { + if cap != 0 && cap < api.server.minCapacity { + return ErrMinCap } api.vip.lock.Lock() defer api.vip.lock.Unlock() c := api.vip.clients[id] - if api.vip.totalVipBw+bw > api.vip.totalBw+c.bw { - return ErrTotalBW + if api.vip.totalVipCap+cap > api.vip.totalCap+c.cap { + return ErrTotalCap } - api.vip.totalVipBw += bw - c.bw - if c.updateBw != nil && bw != 0 { - c.updateBw(bw) + api.vip.totalVipCap += cap - c.cap + if c.updateCap != nil && cap != 0 { + c.updateCap(cap) } if c.connected { - if c.bw != 0 { + if c.cap != 0 { api.vip.vipCount-- } - if bw != 0 { + if cap != 0 { api.vip.vipCount++ } - api.vip.totalConnectedBw += bw - c.bw - api.pm.clientPool.setConnLimit(api.pm.maxFreePeers(api.vip.vipCount, api.vip.totalConnectedBw)) + api.vip.totalConnectedCap += cap - c.cap + api.pm.clientPool.setConnLimit(api.pm.maxFreePeers(api.vip.vipCount, api.vip.totalConnectedCap)) } - if c.updateBw != nil && bw == 0 { - c.updateBw(bw) + if c.updateCap != nil && cap == 0 { + c.updateCap(cap) } - if bw != 0 || c.connected { - c.bw = bw + if cap != 0 || c.connected { + c.cap = cap api.vip.clients[id] = c } else { delete(api.vip.clients, id) @@ -127,7 +127,7 @@ func (api *PrivateLesServerAPI) GetClientCapacity(id enode.ID) hexutil.Uint64 { api.vip.lock.Lock() defer api.vip.lock.Unlock() - return hexutil.Uint64(api.vip.clients[id].bw) + return hexutil.Uint64(api.vip.clients[id].cap) } // connect should be called when a new client is connected. The callback function @@ -138,7 +138,7 @@ func (api *PrivateLesServerAPI) GetClientCapacity(id enode.ID) hexutil.Uint64 { // Note: vipClientPool also stores a record about free clients while they are // connected in order to be able to assign priority to them later with the callback // function if necessary. -func (v *vipClientPool) connect(id enode.ID, updateBw func(uint64)) (uint64, bool) { +func (v *vipClientPool) connect(id enode.ID, updateCap func(uint64)) (uint64, bool) { v.lock.Lock() defer v.lock.Unlock() @@ -147,14 +147,14 @@ func (v *vipClientPool) connect(id enode.ID, updateBw func(uint64)) (uint64, boo return 0, false } c.connected = true - c.updateBw = updateBw + c.updateCap = updateCap v.clients[id] = c - if c.bw != 0 { + if c.cap != 0 { v.vipCount++ } - v.totalConnectedBw += c.bw - v.pm.clientPool.setConnLimit(v.pm.maxFreePeers(v.vipCount, v.totalConnectedBw)) - return c.bw, true + v.totalConnectedCap += c.cap + v.pm.clientPool.setConnLimit(v.pm.maxFreePeers(v.vipCount, v.totalConnectedCap)) + return c.cap, true } // disconnect should be called when a client is disconnected. @@ -165,12 +165,12 @@ func (v *vipClientPool) disconnect(id enode.ID) { c := v.clients[id] c.connected = false - if c.bw != 0 { + if c.cap != 0 { v.clients[id] = c v.vipCount-- } else { delete(v.clients, id) } - v.totalConnectedBw -= c.bw - v.pm.clientPool.setConnLimit(v.pm.maxFreePeers(v.vipCount, v.totalConnectedBw)) + v.totalConnectedCap -= c.cap + v.pm.clientPool.setConnLimit(v.pm.maxFreePeers(v.vipCount, v.totalConnectedCap)) } diff --git a/les/api_test.go b/les/api_test.go index f33491bbb6..2371da4035 100644 --- a/les/api_test.go +++ b/les/api_test.go @@ -51,11 +51,11 @@ request performance test. When testServerDataDir is empty, the test is skipped. */ const ( - testServerDataDir = "" // should always be empty on the master branch + testServerDataDir = "" // should always be empty on the master branch testServerCapacity = 200 - testMaxClients = 10 - testTolerance = 0.1 - minRelBw = 0.2 + testMaxClients = 10 + testTolerance = 0.1 + minRelCap = 0.2 ) func TestCapacityAPI3(t *testing.T) { @@ -97,15 +97,15 @@ func testCapacityAPI(t *testing.T, clientCount int) { t.Fatalf("Failed to obtain rpc client: %v", err) } headNum, headHash := getHead(ctx, t, serverRpcClient) - totalBw, minBw := capacityLimits(ctx, t, serverRpcClient) - fmt.Printf("Server totalBw: %d minBw: %d head number: %d head hash: %064x\n", totalBw, minBw, headNum, headHash) - reqMinBw := uint64(float64(totalBw) * minRelBw / (minRelBw + float64(len(clients)-1))) - if minBw > reqMinBw { - t.Fatalf("Minimum client capacity (%d) bigger than required minimum for this test (%d)", minBw, reqMinBw) + 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))) + if minCap > reqMinCap { + t.Fatalf("Minimum client capacity (%d) bigger than required minimum for this test (%d)", minCap, reqMinCap) } freeIdx := rand.Intn(len(clients)) - freeBw := totalBw / testMaxClients + freeCap := totalCap / testMaxClients for i, client := range clients { var err error @@ -116,7 +116,7 @@ func testCapacityAPI(t *testing.T, clientCount int) { fmt.Println("connecting client", i) if i != freeIdx { - setCapacity(ctx, t, serverRpcClient, client.ID(), totalBw/uint64(len(clients))) + setCapacity(ctx, t, serverRpcClient, client.ID(), totalCap/uint64(len(clients))) } net.Connect(client.ID(), server.ID()) @@ -180,19 +180,19 @@ func testCapacityAPI(t *testing.T, clientCount int) { weights := make([]float64, len(clients)) for c := 0; c < 5; c++ { - setCapacity(ctx, t, serverRpcClient, clients[freeIdx].ID(), freeBw) + setCapacity(ctx, t, serverRpcClient, clients[freeIdx].ID(), freeCap) freeIdx = rand.Intn(len(clients)) var sum float64 for i, _ := range clients { if i == freeIdx { weights[i] = 0 } else { - weights[i] = rand.Float64()*(1-minRelBw) + minRelBw + weights[i] = rand.Float64()*(1-minRelCap) + minRelCap } sum += weights[i] } for i, client := range clients { - weights[i] *= float64(totalBw-freeBw-100) / sum + weights[i] *= float64(totalCap-freeCap-100) / sum capacity := uint64(weights[i]) if i != freeIdx && capacity < getCapacity(ctx, t, serverRpcClient, client.ID()) { setCapacity(ctx, t, serverRpcClient, client.ID(), capacity) @@ -205,9 +205,9 @@ func testCapacityAPI(t *testing.T, clientCount int) { setCapacity(ctx, t, serverRpcClient, client.ID(), capacity) } } - weights[freeIdx] = float64(freeBw) + weights[freeIdx] = float64(freeCap) for i, _ := range clients { - weights[i] /= float64(totalBw) + weights[i] /= float64(totalCap) } time.Sleep(flowcontrol.DecParamDelay) @@ -295,8 +295,8 @@ func testRequest(ctx context.Context, t *testing.T, client *rpc.Client) { } } -func setCapacity(ctx context.Context, t *testing.T, server *rpc.Client, clientID enode.ID, bw uint64) { - if err := server.CallContext(ctx, nil, "les_setClientCapacity", clientID, bw); err != nil { +func setCapacity(ctx context.Context, t *testing.T, server *rpc.Client, clientID enode.ID, cap uint64) { + if err := server.CallContext(ctx, nil, "les_setClientCapacity", clientID, cap); err != nil { t.Fatalf("Failed to set client capacity: %v", err) } } @@ -306,11 +306,11 @@ func getCapacity(ctx context.Context, t *testing.T, server *rpc.Client, clientID if err := server.CallContext(ctx, &s, "les_getClientCapacity", clientID); err != nil { t.Fatalf("Failed to get client capacity: %v", err) } - bw, err := hexutil.DecodeUint64(s) + cap, err := hexutil.DecodeUint64(s) if err != nil { t.Fatalf("Failed to decode client capacity: %v", err) } - return bw + return cap } func capacityLimits(ctx context.Context, t *testing.T, server *rpc.Client) (uint64, uint64) { diff --git a/les/handler.go b/les/handler.go index 54e9bbf7a8..907aea2a31 100644 --- a/les/handler.go +++ b/les/handler.go @@ -103,7 +103,7 @@ type ProtocolManager struct { server *LesServer serverPool *serverPool clientPool *freeClientPool - freeClientBw uint64 + freeClientCap uint64 vipClientPool *vipClientPool lesTopic discv5.Topic reqDist *requestDistributor @@ -198,11 +198,11 @@ func (pm *ProtocolManager) removePeer(id string) { // maxFreePeers returns the maximum number of free client slots based on the number // and total capacity of other clients -func (pm *ProtocolManager) maxFreePeers(otherPeers int, otherBw uint64) int { - if otherPeers >= pm.maxPeers || otherBw >= pm.server.totalCapacity { +func (pm *ProtocolManager) maxFreePeers(otherPeers int, otherCap uint64) int { + if otherPeers >= pm.maxPeers || otherCap >= pm.server.totalCapacity { return 0 } - maxPeers := int((pm.server.totalCapacity - otherBw) / pm.freeClientBw) + maxPeers := int((pm.server.totalCapacity - otherCap) / pm.freeClientCap) if maxPeers <= pm.maxPeers-otherPeers { return maxPeers } @@ -212,14 +212,14 @@ func (pm *ProtocolManager) maxFreePeers(otherPeers int, otherBw uint64) int { func (pm *ProtocolManager) Start(maxPeers int) { pm.maxPeers = maxPeers if pm.server != nil && maxPeers > 0 { - pm.freeClientBw = pm.server.totalCapacity / uint64(maxPeers) - if pm.freeClientBw < pm.server.minCapacity { - pm.freeClientBw = pm.server.minCapacity + pm.freeClientCap = pm.server.totalCapacity / uint64(maxPeers) + if pm.freeClientCap < pm.server.minCapacity { + pm.freeClientCap = pm.server.minCapacity } - if pm.freeClientBw > 0 { + if pm.freeClientCap > 0 { pm.server.defParams = flowcontrol.ServerParams{ - BufLimit: pm.freeClientBw * bufLimitRatio, - MinRecharge: pm.freeClientBw, + BufLimit: pm.freeClientCap * bufLimitRatio, + MinRecharge: pm.freeClientCap, } } } @@ -354,21 +354,21 @@ func (pm *ProtocolManager) handle(p *peer) error { lock.Unlock() }() - updateBw := func(bw uint64) { + updateCap := func(cap uint64) { lock.Lock() defer lock.Unlock() - if !vip && bw != 0 { + if !vip && cap != 0 { // switch to vip mode if free { pm.clientPool.disconnect(freeId) free = false } vip = true - p.updateCapacity(bw) + p.updateCapacity(cap) } if vip { - if bw == 0 { + if cap == 0 { // priority revoked; switch to free client mode or drop if freeId != "" { if !pm.clientPool.connect(freeId, func() { go pm.removePeer(p.id) }) { @@ -378,10 +378,10 @@ func (pm *ProtocolManager) handle(p *peer) error { free = true } vip = false - p.updateCapacity(pm.freeClientBw) + p.updateCapacity(pm.freeClientCap) } else { // just update vip capacity - p.updateCapacity(bw) + p.updateCapacity(cap) } } } @@ -390,16 +390,16 @@ func (pm *ProtocolManager) handle(p *peer) error { if pm.vipClientPool != nil { // the vip client pool registers currently connected non-vip clients too // in order to be able to notify them if they get priority while connected - vipBw, ok := pm.vipClientPool.connect(p.ID(), updateBw) + vipCap, ok := pm.vipClientPool.connect(p.ID(), updateCap) if !ok { lock.Unlock() return p2p.DiscAlreadyConnected } // always unregister defer pm.vipClientPool.disconnect(p.ID()) - if vipBw != 0 { + if vipCap != 0 { vip = true - p.updateCapacity(vipBw) + p.updateCapacity(vipCap) } } diff --git a/les/peer.go b/les/peer.go index f228651900..dc45107913 100644 --- a/les/peer.go +++ b/les/peer.go @@ -188,15 +188,15 @@ func (p *peer) waitBefore(maxCost uint64) (time.Duration, float64) { // updateCapacity updates the request serving capacity assigned to a given client // and also sends an announcement about the updated flow control parameters -func (p *peer) updateCapacity(bw uint64) { +func (p *peer) updateCapacity(cap uint64) { p.responseLock.Lock() defer p.responseLock.Unlock() - p.fcParams = flowcontrol.ServerParams{MinRecharge: bw, BufLimit: bw * bufLimitRatio} + p.fcParams = flowcontrol.ServerParams{MinRecharge: cap, BufLimit: cap * bufLimitRatio} p.fcClient.UpdateParams(p.fcParams) var kvList keyValueList - kvList = kvList.add("flowControl/MRR", bw) - kvList = kvList.add("flowControl/BL", bw*bufLimitRatio) + kvList = kvList.add("flowControl/MRR", cap) + kvList = kvList.add("flowControl/BL", cap*bufLimitRatio) p.queueSend(func() { p.SendAnnounce(announceData{Update: kvList}) }) } diff --git a/les/server.go b/les/server.go index 137f2f14b0..dc909acbf6 100644 --- a/les/server.go +++ b/les/server.go @@ -55,7 +55,7 @@ type LesServer struct { onlyAnnounce bool totalCapacity, minCapacity, minBufLimit, bufLimitRatio uint64 - bwcNormal, bwcBlockProcessing flowcontrol.PieceWiseLinear // capacity curve for normal operation and block processing mode + rcNormal, rcBlockProcessing flowcontrol.PieceWiseLinear // buffer recharge curve for normal operation and block processing mode thcNormal, thcBlockProcessing int // serving thread count for normal operation and block processing mode } @@ -104,26 +104,26 @@ func NewLesServer(eth *eth.Ethereum, config *eth.Config) (*LesServer, error) { logger := log.New() pm.server = srv - bwNormal := uint64(config.LightServ) * flowcontrol.FixedPointMultiplier / 100 - srv.bwcNormal = flowcontrol.PieceWiseLinear{{0, 0} /*{bwNormal / 10, bwNormal}, */, {bwNormal, bwNormal}} + capNormal := uint64(config.LightServ) * flowcontrol.FixedPointMultiplier / 100 + srv.rcNormal = flowcontrol.PieceWiseLinear{{0, 0} /*{capNormal / 10, capNormal}, */, {capNormal, capNormal}} // limit the serving thread count to at least 4 times the targeted average // capacity, allowing more paralellization in short-term load spikes but // still limiting the total thread count at a reasonable level - srv.thcNormal = int(bwNormal * 4 / flowcontrol.FixedPointMultiplier) + srv.thcNormal = int(capNormal * 4 / flowcontrol.FixedPointMultiplier) if srv.thcNormal < 4 { srv.thcNormal = 4 } // while processing blocks use half of the normal target capacity - bwBlockProcessing := bwNormal / 2 - srv.bwcBlockProcessing = flowcontrol.PieceWiseLinear{{0, 0} /*{bwBlockProcessing / 10, bwBlockProcessing}, */, {bwBlockProcessing, bwBlockProcessing}} + capBlockProcessing := capNormal / 2 + srv.rcBlockProcessing = flowcontrol.PieceWiseLinear{{0, 0} /*{capBlockProcessing / 10, capBlockProcessing}, */, {capBlockProcessing, capBlockProcessing}} // limit the serving thread count just above the targeted average capacity, // ensuring that block processing is minimally hindered - srv.thcBlockProcessing = int(bwBlockProcessing/flowcontrol.FixedPointMultiplier) + 1 + srv.thcBlockProcessing = int(capBlockProcessing/flowcontrol.FixedPointMultiplier) + 1 pm.servingQueue.setThreads(srv.thcNormal) - srv.fcManager = flowcontrol.NewClientManager(srv.bwcNormal, &mclock.System{}) + srv.fcManager = flowcontrol.NewClientManager(srv.rcNormal, &mclock.System{}) - srv.totalCapacity = bwNormal + srv.totalCapacity = capNormal if config.LightBandwidthIn > 0 { pm.inSizeCostFactor = float64(srv.totalCapacity) / float64(config.LightBandwidthIn) } @@ -183,10 +183,10 @@ func (s *LesServer) blockProcLoop(pm *ProtocolManager) { case processing := <-procFeedback: if processing { pm.servingQueue.setThreads(s.thcBlockProcessing) - s.fcManager.SetRechargeCurve(s.bwcBlockProcessing) + s.fcManager.SetRechargeCurve(s.rcBlockProcessing) } else { pm.servingQueue.setThreads(s.thcNormal) - s.fcManager.SetRechargeCurve(s.bwcNormal) + s.fcManager.SetRechargeCurve(s.rcNormal) } case <-pm.quitSync: pm.wg.Done()