forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-20 10:52:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

les: renamed bw to cap

Browse source
This commit is contained in:
Zsolt Felfoldi 2019-01-16 15:37:45 +01:00
parent 3421811582
commit 8fcb060f10
5 changed files with 86 additions and 86 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

64
les/api.go
View file

@ -24,8 +24,8 @@ import (
) )
var ( var (
ErrMinBW = errors.New("capacity too small") ErrMinCap = errors.New("capacity too small")
ErrTotalBW = errors.New("total capacity exceeded") ErrTotalCap = errors.New("total capacity exceeded")
) )
// PublicLesServerAPI provides an API to access the les server. // PublicLesServerAPI provides an API to access the les server.
@ -40,7 +40,7 @@ type PrivateLesServerAPI struct {
func NewPrivateLesServerAPI(server *LesServer) *PrivateLesServerAPI { func NewPrivateLesServerAPI(server *LesServer) *PrivateLesServerAPI {
vip := &vipClientPool{ vip := &vipClientPool{
clients: make(map[enode.ID]vipClientInfo), clients: make(map[enode.ID]vipClientInfo),
totalBw: server.totalCapacity, totalCap: server.totalCapacity,
pm: server.protocolManager, pm: server.protocolManager,
} }
server.protocolManager.vipClientPool = vip server.protocolManager.vipClientPool = vip
@ -66,15 +66,15 @@ type vipClientPool struct {
lock sync.Mutex lock sync.Mutex
pm *ProtocolManager pm *ProtocolManager
clients map[enode.ID]vipClientInfo clients map[enode.ID]vipClientInfo
totalBw, totalVipBw, totalConnectedBw uint64 totalCap, totalVipCap, totalConnectedCap uint64
vipCount int vipCount int
} }
// vipClientInfo entries exist for all prioritized clients and currently connected free clients // vipClientInfo entries exist for all prioritized clients and currently connected free clients
type vipClientInfo struct { type vipClientInfo struct {
bw uint64 // zero for non-vip clients cap uint64 // zero for non-vip clients
connected bool connected bool
updateBw func(uint64) updateCap func(uint64)
} }
// SetClientCapacity sets the priority capacity assigned to a given client. // 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 // Note: assigned capacity can be changed while the client is connected with
// immediate effect. // immediate effect.
func (api *PrivateLesServerAPI) SetClientCapacity(id enode.ID, bw uint64) error { func (api *PrivateLesServerAPI) SetClientCapacity(id enode.ID, cap uint64) error {
if bw != 0 && bw < api.server.minCapacity { if cap != 0 && cap < api.server.minCapacity {
return ErrMinBW return ErrMinCap
} }
api.vip.lock.Lock() api.vip.lock.Lock()
defer api.vip.lock.Unlock() defer api.vip.lock.Unlock()
c := api.vip.clients[id] c := api.vip.clients[id]
if api.vip.totalVipBw+bw > api.vip.totalBw+c.bw { if api.vip.totalVipCap+cap > api.vip.totalCap+c.cap {
return ErrTotalBW return ErrTotalCap
} }
api.vip.totalVipBw += bw - c.bw api.vip.totalVipCap += cap - c.cap
if c.updateBw != nil && bw != 0 { if c.updateCap != nil && cap != 0 {
c.updateBw(bw) c.updateCap(cap)
} }
if c.connected { if c.connected {
if c.bw != 0 { if c.cap != 0 {
api.vip.vipCount-- api.vip.vipCount--
} }
if bw != 0 { if cap != 0 {
api.vip.vipCount++ api.vip.vipCount++
} }
api.vip.totalConnectedBw += bw - c.bw api.vip.totalConnectedCap += cap - c.cap
api.pm.clientPool.setConnLimit(api.pm.maxFreePeers(api.vip.vipCount, api.vip.totalConnectedBw)) api.pm.clientPool.setConnLimit(api.pm.maxFreePeers(api.vip.vipCount, api.vip.totalConnectedCap))
} }
if c.updateBw != nil && bw == 0 { if c.updateCap != nil && cap == 0 {
c.updateBw(bw) c.updateCap(cap)
} }
if bw != 0 || c.connected { if cap != 0 || c.connected {
c.bw = bw c.cap = cap
api.vip.clients[id] = c api.vip.clients[id] = c
} else { } else {
delete(api.vip.clients, id) delete(api.vip.clients, id)
@ -127,7 +127,7 @@ func (api *PrivateLesServerAPI) GetClientCapacity(id enode.ID) hexutil.Uint64 {
api.vip.lock.Lock() api.vip.lock.Lock()
defer api.vip.lock.Unlock() 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 // 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 // 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 // connected in order to be able to assign priority to them later with the callback
// function if necessary. // 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() v.lock.Lock()
defer v.lock.Unlock() defer v.lock.Unlock()
@ -147,14 +147,14 @@ func (v *vipClientPool) connect(id enode.ID, updateBw func(uint64)) (uint64, boo
return 0, false return 0, false
} }
c.connected = true c.connected = true
c.updateBw = updateBw c.updateCap = updateCap
v.clients[id] = c v.clients[id] = c
if c.bw != 0 { if c.cap != 0 {
v.vipCount++ v.vipCount++
} }
v.totalConnectedBw += c.bw v.totalConnectedCap += c.cap
v.pm.clientPool.setConnLimit(v.pm.maxFreePeers(v.vipCount, v.totalConnectedBw)) v.pm.clientPool.setConnLimit(v.pm.maxFreePeers(v.vipCount, v.totalConnectedCap))
return c.bw, true return c.cap, true
} }
// disconnect should be called when a client is disconnected. // 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 := v.clients[id]
c.connected = false c.connected = false
if c.bw != 0 { if c.cap != 0 {
v.clients[id] = c v.clients[id] = c
v.vipCount-- v.vipCount--
} else { } else {
delete(v.clients, id) delete(v.clients, id)
} }
v.totalConnectedBw -= c.bw v.totalConnectedCap -= c.cap
v.pm.clientPool.setConnLimit(v.pm.maxFreePeers(v.vipCount, v.totalConnectedBw)) v.pm.clientPool.setConnLimit(v.pm.maxFreePeers(v.vipCount, v.totalConnectedCap))
} }

34
les/api_test.go
View file

@ -55,7 +55,7 @@ const (
testServerCapacity = 200 testServerCapacity = 200
testMaxClients = 10 testMaxClients = 10
testTolerance = 0.1 testTolerance = 0.1
minRelBw = 0.2 minRelCap = 0.2
) )
func TestCapacityAPI3(t *testing.T) { 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) t.Fatalf("Failed to obtain rpc client: %v", err)
} }
headNum, headHash := getHead(ctx, t, serverRpcClient) headNum, headHash := getHead(ctx, t, serverRpcClient)
totalBw, minBw := capacityLimits(ctx, t, serverRpcClient) totalCap, minCap := capacityLimits(ctx, t, serverRpcClient)
fmt.Printf("Server totalBw: %d minBw: %d head number: %d head hash: %064x\n", totalBw, minBw, headNum, headHash) fmt.Printf("Server totalCap: %d minCap: %d head number: %d head hash: %064x\n", totalCap, minCap, headNum, headHash)
reqMinBw := uint64(float64(totalBw) * minRelBw / (minRelBw + float64(len(clients)-1))) reqMinCap := uint64(float64(totalCap) * minRelCap / (minRelCap + float64(len(clients)-1)))
if minBw > reqMinBw { if minCap > reqMinCap {
t.Fatalf("Minimum client capacity (%d) bigger than required minimum for this test (%d)", minBw, reqMinBw) t.Fatalf("Minimum client capacity (%d) bigger than required minimum for this test (%d)", minCap, reqMinCap)
} }
freeIdx := rand.Intn(len(clients)) freeIdx := rand.Intn(len(clients))
freeBw := totalBw / testMaxClients freeCap := totalCap / testMaxClients
for i, client := range clients { for i, client := range clients {
var err error var err error
@ -116,7 +116,7 @@ func testCapacityAPI(t *testing.T, clientCount int) {
fmt.Println("connecting client", i) fmt.Println("connecting client", i)
if i != freeIdx { 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()) net.Connect(client.ID(), server.ID())
@ -180,19 +180,19 @@ func testCapacityAPI(t *testing.T, clientCount int) {
weights := make([]float64, len(clients)) weights := make([]float64, len(clients))
for c := 0; c < 5; c++ { 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)) freeIdx = rand.Intn(len(clients))
var sum float64 var sum float64
for i, _ := range clients { for i, _ := range clients {
if i == freeIdx { if i == freeIdx {
weights[i] = 0 weights[i] = 0
} else { } else {
weights[i] = rand.Float64()*(1-minRelBw) + minRelBw weights[i] = rand.Float64()*(1-minRelCap) + minRelCap
} }
sum += weights[i] sum += weights[i]
} }
for i, client := range clients { for i, client := range clients {
weights[i] *= float64(totalBw-freeBw-100) / sum weights[i] *= float64(totalCap-freeCap-100) / sum
capacity := uint64(weights[i]) capacity := uint64(weights[i])
if i != freeIdx && capacity < getCapacity(ctx, t, serverRpcClient, client.ID()) { if i != freeIdx && capacity < getCapacity(ctx, t, serverRpcClient, client.ID()) {
setCapacity(ctx, t, serverRpcClient, client.ID(), capacity) 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) setCapacity(ctx, t, serverRpcClient, client.ID(), capacity)
} }
} }
weights[freeIdx] = float64(freeBw) weights[freeIdx] = float64(freeCap)
for i, _ := range clients { for i, _ := range clients {
weights[i] /= float64(totalBw) weights[i] /= float64(totalCap)
} }
time.Sleep(flowcontrol.DecParamDelay) 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) { 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, bw); err != nil { if err := server.CallContext(ctx, nil, "les_setClientCapacity", clientID, cap); err != nil {
t.Fatalf("Failed to set client capacity: %v", err) 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 { if err := server.CallContext(ctx, &s, "les_getClientCapacity", clientID); err != nil {
t.Fatalf("Failed to get client capacity: %v", err) t.Fatalf("Failed to get client capacity: %v", err)
} }
bw, err := hexutil.DecodeUint64(s) cap, err := hexutil.DecodeUint64(s)
if err != nil { if err != nil {
t.Fatalf("Failed to decode client capacity: %v", err) 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) { func capacityLimits(ctx context.Context, t *testing.T, server *rpc.Client) (uint64, uint64) {

38
les/handler.go
View file

@ -103,7 +103,7 @@ type ProtocolManager struct {
server *LesServer server *LesServer
serverPool *serverPool serverPool *serverPool
clientPool *freeClientPool clientPool *freeClientPool
freeClientBw uint64 freeClientCap uint64
vipClientPool *vipClientPool vipClientPool *vipClientPool
lesTopic discv5.Topic lesTopic discv5.Topic
reqDist *requestDistributor 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 // maxFreePeers returns the maximum number of free client slots based on the number
// and total capacity of other clients // and total capacity of other clients
func (pm *ProtocolManager) maxFreePeers(otherPeers int, otherBw uint64) int { func (pm *ProtocolManager) maxFreePeers(otherPeers int, otherCap uint64) int {
if otherPeers >= pm.maxPeers || otherBw >= pm.server.totalCapacity { if otherPeers >= pm.maxPeers || otherCap >= pm.server.totalCapacity {
return 0 return 0
} }
maxPeers := int((pm.server.totalCapacity - otherBw) / pm.freeClientBw) maxPeers := int((pm.server.totalCapacity - otherCap) / pm.freeClientCap)
if maxPeers <= pm.maxPeers-otherPeers { if maxPeers <= pm.maxPeers-otherPeers {
return maxPeers return maxPeers
} }
@ -212,14 +212,14 @@ func (pm *ProtocolManager) maxFreePeers(otherPeers int, otherBw uint64) int {
func (pm *ProtocolManager) Start(maxPeers int) { func (pm *ProtocolManager) Start(maxPeers int) {
pm.maxPeers = maxPeers pm.maxPeers = maxPeers
if pm.server != nil && maxPeers > 0 { if pm.server != nil && maxPeers > 0 {
pm.freeClientBw = pm.server.totalCapacity / uint64(maxPeers) pm.freeClientCap = pm.server.totalCapacity / uint64(maxPeers)
if pm.freeClientBw < pm.server.minCapacity { if pm.freeClientCap < pm.server.minCapacity {
pm.freeClientBw = pm.server.minCapacity pm.freeClientCap = pm.server.minCapacity
} }
if pm.freeClientBw > 0 { if pm.freeClientCap > 0 {
pm.server.defParams = flowcontrol.ServerParams{ pm.server.defParams = flowcontrol.ServerParams{
BufLimit: pm.freeClientBw * bufLimitRatio, BufLimit: pm.freeClientCap * bufLimitRatio,
MinRecharge: pm.freeClientBw, MinRecharge: pm.freeClientCap,
} }
} }
} }
@ -354,21 +354,21 @@ func (pm *ProtocolManager) handle(p *peer) error {
lock.Unlock() lock.Unlock()
}() }()
updateBw := func(bw uint64) { updateCap := func(cap uint64) {
lock.Lock() lock.Lock()
defer lock.Unlock() defer lock.Unlock()
if !vip && bw != 0 { if !vip && cap != 0 {
// switch to vip mode // switch to vip mode
if free { if free {
pm.clientPool.disconnect(freeId) pm.clientPool.disconnect(freeId)
free = false free = false
} }
vip = true vip = true
p.updateCapacity(bw) p.updateCapacity(cap)
} }
if vip { if vip {
if bw == 0 { if cap == 0 {
// priority revoked; switch to free client mode or drop // priority revoked; switch to free client mode or drop
if freeId != "" { if freeId != "" {
if !pm.clientPool.connect(freeId, func() { go pm.removePeer(p.id) }) { if !pm.clientPool.connect(freeId, func() { go pm.removePeer(p.id) }) {
@ -378,10 +378,10 @@ func (pm *ProtocolManager) handle(p *peer) error {
free = true free = true
} }
vip = false vip = false
p.updateCapacity(pm.freeClientBw) p.updateCapacity(pm.freeClientCap)
} else { } else {
// just update vip capacity // 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 { if pm.vipClientPool != nil {
// the vip client pool registers currently connected non-vip clients too // 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 // 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 { if !ok {
lock.Unlock() lock.Unlock()
return p2p.DiscAlreadyConnected return p2p.DiscAlreadyConnected
} }
// always unregister // always unregister
defer pm.vipClientPool.disconnect(p.ID()) defer pm.vipClientPool.disconnect(p.ID())
if vipBw != 0 { if vipCap != 0 {
vip = true vip = true
p.updateCapacity(vipBw) p.updateCapacity(vipCap)
} }
} }

8
les/peer.go
View file

@ -188,15 +188,15 @@ func (p *peer) waitBefore(maxCost uint64) (time.Duration, float64) {
// updateCapacity updates the request serving capacity assigned to a given client // updateCapacity updates the request serving capacity assigned to a given client
// and also sends an announcement about the updated flow control parameters // 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() p.responseLock.Lock()
defer p.responseLock.Unlock() 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) p.fcClient.UpdateParams(p.fcParams)
var kvList keyValueList var kvList keyValueList
kvList = kvList.add("flowControl/MRR", bw) kvList = kvList.add("flowControl/MRR", cap)
kvList = kvList.add("flowControl/BL", bw*bufLimitRatio) kvList = kvList.add("flowControl/BL", cap*bufLimitRatio)
p.queueSend(func() { p.SendAnnounce(announceData{Update: kvList}) }) p.queueSend(func() { p.SendAnnounce(announceData{Update: kvList}) })
} }

22
les/server.go
View file

@ -55,7 +55,7 @@ type LesServer struct {
onlyAnnounce bool onlyAnnounce bool
totalCapacity, minCapacity, minBufLimit, bufLimitRatio uint64 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 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() logger := log.New()
pm.server = srv pm.server = srv
bwNormal := uint64(config.LightServ) * flowcontrol.FixedPointMultiplier / 100 capNormal := uint64(config.LightServ) * flowcontrol.FixedPointMultiplier / 100
srv.bwcNormal = flowcontrol.PieceWiseLinear{{0, 0} /*{bwNormal / 10, bwNormal}, */, {bwNormal, bwNormal}} srv.rcNormal = flowcontrol.PieceWiseLinear{{0, 0} /*{capNormal / 10, capNormal}, */, {capNormal, capNormal}}
// limit the serving thread count to at least 4 times the targeted average // limit the serving thread count to at least 4 times the targeted average
// capacity, allowing more paralellization in short-term load spikes but // capacity, allowing more paralellization in short-term load spikes but
// still limiting the total thread count at a reasonable level // 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 { if srv.thcNormal < 4 {
srv.thcNormal = 4 srv.thcNormal = 4
} }
// while processing blocks use half of the normal target capacity // while processing blocks use half of the normal target capacity
bwBlockProcessing := bwNormal / 2 capBlockProcessing := capNormal / 2
srv.bwcBlockProcessing = flowcontrol.PieceWiseLinear{{0, 0} /*{bwBlockProcessing / 10, bwBlockProcessing}, */, {bwBlockProcessing, bwBlockProcessing}} srv.rcBlockProcessing = flowcontrol.PieceWiseLinear{{0, 0} /*{capBlockProcessing / 10, capBlockProcessing}, */, {capBlockProcessing, capBlockProcessing}}
// limit the serving thread count just above the targeted average capacity, // limit the serving thread count just above the targeted average capacity,
// ensuring that block processing is minimally hindered // 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) 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 { if config.LightBandwidthIn > 0 {
pm.inSizeCostFactor = float64(srv.totalCapacity) / float64(config.LightBandwidthIn) pm.inSizeCostFactor = float64(srv.totalCapacity) / float64(config.LightBandwidthIn)
} }
@ -183,10 +183,10 @@ func (s *LesServer) blockProcLoop(pm *ProtocolManager) {
case processing := <-procFeedback: case processing := <-procFeedback:
if processing { if processing {
pm.servingQueue.setThreads(s.thcBlockProcessing) pm.servingQueue.setThreads(s.thcBlockProcessing)
s.fcManager.SetRechargeCurve(s.bwcBlockProcessing) s.fcManager.SetRechargeCurve(s.rcBlockProcessing)
} else { } else {
pm.servingQueue.setThreads(s.thcNormal) pm.servingQueue.setThreads(s.thcNormal)
s.fcManager.SetRechargeCurve(s.bwcNormal) s.fcManager.SetRechargeCurve(s.rcNormal)
} }
case <-pm.quitSync: case <-pm.quitSync:
pm.wg.Done() pm.wg.Done()