From 2de75e9e1c6f80c7124552de531516cb968bf479 Mon Sep 17 00:00:00 2001 From: Zsolt Felfoldi Date: Fri, 29 Mar 2019 14:13:18 +0100 Subject: [PATCH] les: implement server API --- internal/web3ext/web3ext.go | 19 + les/api.go | 1057 ++++++++++++++++++++++++----------- les/api_test.go | 99 ++-- les/handler.go | 1 + les/peer.go | 9 +- les/server.go | 10 +- 6 files changed, 802 insertions(+), 393 deletions(-) diff --git a/internal/web3ext/web3ext.go b/internal/web3ext/web3ext.go index 0abcd5a8a5..707771336e 100644 --- a/internal/web3ext/web3ext.go +++ b/internal/web3ext/web3ext.go @@ -445,6 +445,11 @@ web3._extend({ params: 2, inputFormatter:[null, null], }), + new web3._extend.Method({ + name: 'freezeClient', + call: 'debug_freezeClient', + params: 1, + }), ], properties: [] }); @@ -772,6 +777,16 @@ web3._extend({ call: 'les_getCheckpoint', params: 1 }), + new web3._extend.Method({ + name: 'clientInfo', + call: 'les_clientInfo', + params: 2 + }), + new web3._extend.Method({ + name: 'setClientParams', + call: 'les_setClientParams', + params: 3 + }), ], properties: [ @@ -783,6 +798,10 @@ web3._extend({ name: 'checkpointContractAddress', getter: 'les_getCheckpointContractAddress' }), + new web3._extend.Property({ + name: 'serverInfo', + getter: 'les_serverInfo' + }), ] }); ` diff --git a/les/api.go b/les/api.go index 9ce6ed4800..016c18ae7d 100644 --- a/les/api.go +++ b/les/api.go @@ -23,7 +23,6 @@ import ( "sync" "time" - "github.com/ethereum/go-ethereum/common/hexutil" "github.com/ethereum/go-ethereum/common/mclock" "github.com/ethereum/go-ethereum/les/csvlogger" "github.com/ethereum/go-ethereum/p2p/enode" @@ -31,17 +30,20 @@ import ( ) var ( - ErrMinCap = errors.New("capacity too small") - ErrTotalCap = errors.New("total capacity exceeded") - ErrUnknownBenchmarkType = errors.New("unknown benchmark type") ErrNoCheckpoint = errors.New("no local checkpoint provided") ErrNotActivated = errors.New("checkpoint registrar is not activated") + ErrInvalidTag = errors.New("invalid client tag") + ErrInvalidParam = errors.New("invalid client parameter") + ErrInvalidValue = errors.New("invalid parameter value") + ErrTotalCap = errors.New("total capacity exceeded") + ErrUnknownBenchmarkType = errors.New("unknown benchmark type") + ErrMultiple = errors.New("multiple errors") + ErrClientNotConnected = errors.New("client is not connected") dropCapacityDelay = time.Second // delay applied to decreasing capacity changes ) // PrivateLightServerAPI provides an API to access the LES light server. -// It offers only methods that operate on public data that is freely available to anyone. type PrivateLightServerAPI struct { server *LesServer } @@ -53,350 +55,128 @@ func NewPrivateLightServerAPI(server *LesServer) *PrivateLightServerAPI { } } -// TotalCapacity queries total available capacity for all clients -func (api *PrivateLightServerAPI) TotalCapacity() hexutil.Uint64 { - return hexutil.Uint64(api.server.priorityClientPool.totalCapacity()) +// ServerInfo returns global server parameters +func (api *PrivateLightServerAPI) ServerInfo() map[string]interface{} { + res := make(map[string]interface{}) + res["minimumCapacity"] = uint64(api.server.minCapacity) + res["freeClientCapacity"] = api.server.freeClientCap + res["totalCapacity"], res["totalConnectedCapacity"], res["priorityConnectedCapacity"], res["totalPriorityCapacity"] = api.server.priorityClientPool.capacityInfo() + return res } -// SubscribeTotalCapacity subscribes to changed total capacity events. -// If onlyUnderrun is true then notification is sent only if the total capacity -// drops under the total capacity of connected priority clients. -// -// Note: actually applying decreasing total capacity values is delayed while the -// notification is sent instantly. This allows lowering the capacity of a priority client -// or choosing which one to drop before the system drops some of them automatically. -func (api *PrivateLightServerAPI) SubscribeTotalCapacity(ctx context.Context, onlyUnderrun bool) (*rpc.Subscription, error) { +// ClientInfo returns information about clients listed in the ids list or matching the given tags +func (api *PrivateLightServerAPI) ClientInfo(ids []enode.ID, tags []string) map[enode.ID]map[string]interface{} { + res := make(map[enode.ID]map[string]interface{}) + api.server.priorityClientPool.matchClients(ids, tags, func(client *priorityClientInfo) { + res[client.id] = api.server.priorityClientPool.clientInfo(client) + }) + return res +} + +// SetClientParams sets client parameters for all clients listed in the ids list or matching the given tags +func (api *PrivateLightServerAPI) SetClientParams(ids []enode.ID, tags []string, params map[string]interface{}) error { + var finalErr error + api.server.priorityClientPool.matchClients(ids, tags, func(client *priorityClientInfo) { + var ( + err error + updatePrice bool + ) + for name, value := range params { + switch name { + case "userTags": + if t, ok := value.([]interface{}); ok { + tags := make([]string, len(t)) + for i, tag := range t { + if tt, ok := tag.(string); ok { + tags[i] = tt + } else { + err = ErrInvalidTag + } + } + if err == nil { + err = api.server.priorityClientPool.setClientTags(client, tags) + } + } else { + err = ErrInvalidTag + } + case "capacity": + if cap, ok := value.(float64); ok && (cap == 0 || uint64(cap) >= api.server.minCapacity) { + err = api.server.priorityClientPool.setClientCapacity(client, uint64(cap)) + updatePrice = true + } else { + err = ErrInvalidValue + } + case "pricing/timeFactor": + if val, ok := value.(float64); ok && val >= 0 { + client.timeFactor = val / 1000000000 + updatePrice = true + } else { + err = ErrInvalidValue + } + case "pricing/capacityFactor": + if val, ok := value.(float64); ok && val >= 0 { + client.capacityFactor = val / 1000000000 + updatePrice = true + } else { + err = ErrInvalidValue + } + case "pricing/requestCostFactor": + if val, ok := value.(float64); ok && val >= 0 { + client.requestCostFactor = val / 1000000000 + updatePrice = true + } else { + err = ErrInvalidValue + } + case "pricing/alert": + if val, ok := value.(float64); ok && val >= 0 { + api.server.priorityClientPool.setPriceUpdate(client, uint64(val), false) + } else { + err = ErrInvalidValue + } + case "pricing/periodicUpdate": + if val, ok := value.(float64); ok && val >= 0 { + api.server.priorityClientPool.setPriceUpdate(client, uint64(val), true) + } else { + err = ErrInvalidValue + } + default: + err = ErrInvalidParam + } + if err != nil { + if finalErr == nil { + finalErr = err + } else { + finalErr = ErrMultiple + } + } + } + if updatePrice && client.connected { + api.server.priorityClientPool.updatePriceFactors(client) + } + }) + return finalErr +} + +// eventSub represents an event subscription +type eventSub struct { + notifier *rpc.Notifier + rpcSub *rpc.Subscription + clientTags []string + totalCapUnderrun bool +} + +// SubscribeEvent subscribes to global events and client events related to the clients matching the given tags. +// If totalCapUnderrun is true then totalCapacity updates are only sent when totalCapacity drops under totalConnectedCapacity. +func (api *PrivateLightServerAPI) SubscribeEvent(ctx context.Context, clientTags []string, totalCapUnderrun bool) (*rpc.Subscription, error) { notifier, supported := rpc.NotifierFromContext(ctx) if !supported { return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported } rpcSub := notifier.CreateSubscription() - api.server.priorityClientPool.subscribeTotalCapacity(&tcSubscription{notifier, rpcSub, onlyUnderrun}) + api.server.priorityClientPool.subscribeEvents(&eventSub{notifier, rpcSub, clientTags, totalCapUnderrun}) return rpcSub, nil } -type ( - // tcSubscription represents a total capacity subscription - tcSubscription struct { - notifier *rpc.Notifier - rpcSub *rpc.Subscription - onlyUnderrun bool - } - tcSubs map[*tcSubscription]struct{} -) - -// send sends a changed total capacity event to the subscribers -func (s tcSubs) send(tc uint64, underrun bool) { - for sub := range s { - select { - case <-sub.rpcSub.Err(): - delete(s, sub) - case <-sub.notifier.Closed(): - delete(s, sub) - default: - if underrun || !sub.onlyUnderrun { - sub.notifier.Notify(sub.rpcSub.ID, tc) - } - } - } -} - -// MinimumCapacity queries minimum assignable capacity for a single client -func (api *PrivateLightServerAPI) MinimumCapacity() hexutil.Uint64 { - return hexutil.Uint64(api.server.minCapacity) -} - -// FreeClientCapacity queries the capacity provided for free clients -func (api *PrivateLightServerAPI) FreeClientCapacity() hexutil.Uint64 { - return hexutil.Uint64(api.server.freeClientCap) -} - -// SetClientCapacity sets the priority capacity assigned to a given client. -// If the assigned capacity is bigger than zero then connection is always -// guaranteed. The sum of capacity assigned to priority clients can not exceed -// the total available capacity. -// -// Note: assigned capacity can be changed while the client is connected with -// immediate effect. -func (api *PrivateLightServerAPI) SetClientCapacity(id enode.ID, cap uint64) error { - if cap != 0 && cap < api.server.minCapacity { - return ErrMinCap - } - return api.server.priorityClientPool.setClientCapacity(id, cap) -} - -// GetClientCapacity returns the capacity assigned to a given client -func (api *PrivateLightServerAPI) GetClientCapacity(id enode.ID) hexutil.Uint64 { - api.server.priorityClientPool.lock.Lock() - defer api.server.priorityClientPool.lock.Unlock() - - return hexutil.Uint64(api.server.priorityClientPool.clients[id].cap) -} - -// clientPool is implemented by both the free and priority client pools -type clientPool interface { - peerSetNotify - setLimits(count int, totalCap uint64) -} - -// priorityClientPool stores information about prioritized clients -type priorityClientPool struct { - lock sync.Mutex - child clientPool - ps *peerSet - clients map[enode.ID]priorityClientInfo - totalCap, totalCapAnnounced uint64 - totalConnectedCap, freeClientCap uint64 - maxPeers, priorityCount int - logger *csvlogger.Logger - logTotalPriConn *csvlogger.Channel - - subs tcSubs - updateSchedule []scheduledUpdate - scheduleCounter uint64 -} - -// scheduledUpdate represents a delayed total capacity update -type scheduledUpdate struct { - time mclock.AbsTime - totalCap, id uint64 -} - -// priorityClientInfo entries exist for all prioritized clients and currently connected non-priority clients -type priorityClientInfo struct { - cap uint64 // zero for non-priority clients - connected bool - peer *peer -} - -// newPriorityClientPool creates a new priority client pool -func newPriorityClientPool(freeClientCap uint64, ps *peerSet, child clientPool, metricsLogger, eventLogger *csvlogger.Logger) *priorityClientPool { - return &priorityClientPool{ - clients: make(map[enode.ID]priorityClientInfo), - freeClientCap: freeClientCap, - ps: ps, - child: child, - logger: eventLogger, - logTotalPriConn: metricsLogger.NewChannel("totalPriConn", 0), - } -} - -// registerPeer is called when a new client is connected. If the client has no -// priority assigned then it is passed to the child pool which may either keep it -// or disconnect it. -// -// Note: priorityClientPool also stores a record about free clients while they are -// connected in order to be able to assign priority to them later. -func (v *priorityClientPool) registerPeer(p *peer) { - v.lock.Lock() - defer v.lock.Unlock() - - id := p.ID() - c := v.clients[id] - v.logger.Event(fmt.Sprintf("priorityClientPool: registerPeer cap=%d connected=%v, %x", c.cap, c.connected, id.Bytes())) - if c.connected { - return - } - if c.cap == 0 && v.child != nil { - v.child.registerPeer(p) - } - if c.cap != 0 && v.totalConnectedCap+c.cap > v.totalCap { - v.logger.Event(fmt.Sprintf("priorityClientPool: rejected, %x", id.Bytes())) - go v.ps.Unregister(p.id) - return - } - - c.connected = true - c.peer = p - v.clients[id] = c - if c.cap != 0 { - v.priorityCount++ - v.totalConnectedCap += c.cap - v.logger.Event(fmt.Sprintf("priorityClientPool: accepted with %d capacity, %x", c.cap, id.Bytes())) - v.logTotalPriConn.Update(float64(v.totalConnectedCap)) - if v.child != nil { - v.child.setLimits(v.maxPeers-v.priorityCount, v.totalCap-v.totalConnectedCap) - } - p.updateCapacity(c.cap) - } -} - -// unregisterPeer is called when a client is disconnected. If the client has no -// priority assigned then it is also removed from the child pool. -func (v *priorityClientPool) unregisterPeer(p *peer) { - v.lock.Lock() - defer v.lock.Unlock() - - id := p.ID() - c := v.clients[id] - v.logger.Event(fmt.Sprintf("priorityClientPool: unregisterPeer cap=%d connected=%v, %x", c.cap, c.connected, id.Bytes())) - if !c.connected { - return - } - if c.cap != 0 { - c.connected = false - v.clients[id] = c - v.priorityCount-- - v.totalConnectedCap -= c.cap - v.logTotalPriConn.Update(float64(v.totalConnectedCap)) - if v.child != nil { - v.child.setLimits(v.maxPeers-v.priorityCount, v.totalCap-v.totalConnectedCap) - } - } else { - if v.child != nil { - v.child.unregisterPeer(p) - } - delete(v.clients, id) - } -} - -// setLimits updates the allowed peer count and total capacity of the priority -// client pool. Since the free client pool is a child of the priority pool the -// remaining peer count and capacity is assigned to the free pool by calling its -// own setLimits function. -// -// Note: a decreasing change of the total capacity is applied with a delay. -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) - return - } - v.setLimitsNow(count, v.totalCap) - if totalCap < v.totalCap { - v.subs.send(totalCap, totalCap < v.totalConnectedCap) - for i, s := range v.updateSchedule { - if totalCap >= s.totalCap { - s.totalCap = totalCap - v.updateSchedule = v.updateSchedule[:i+1] - return - } - } - v.updateSchedule = append(v.updateSchedule, scheduledUpdate{time: mclock.Now() + mclock.AbsTime(dropCapacityDelay), totalCap: totalCap}) - if len(v.updateSchedule) == 1 { - v.scheduleCounter++ - id := v.scheduleCounter - v.updateSchedule[0].id = id - time.AfterFunc(dropCapacityDelay, func() { v.checkUpdate(id) }) - } - } else { - v.updateSchedule = nil - } -} - -// checkUpdate performs the next scheduled update if possible and schedules -// the one after that -func (v *priorityClientPool) checkUpdate(id uint64) { - v.lock.Lock() - defer v.lock.Unlock() - - if len(v.updateSchedule) == 0 || v.updateSchedule[0].id != id { - return - } - v.setLimitsNow(v.maxPeers, v.updateSchedule[0].totalCap) - v.updateSchedule = v.updateSchedule[1:] - if len(v.updateSchedule) != 0 { - v.scheduleCounter++ - id := v.scheduleCounter - v.updateSchedule[0].id = id - dt := time.Duration(v.updateSchedule[0].time - mclock.Now()) - time.AfterFunc(dt, func() { v.checkUpdate(id) }) - } -} - -// setLimits updates the allowed peer count and total capacity immediately -func (v *priorityClientPool) setLimitsNow(count int, totalCap uint64) { - if v.priorityCount > count || v.totalConnectedCap > totalCap { - for id, c := range v.clients { - if c.connected { - v.logger.Event(fmt.Sprintf("priorityClientPool: setLimitsNow kicked out, %x", id.Bytes())) - c.connected = false - v.totalConnectedCap -= c.cap - v.logTotalPriConn.Update(float64(v.totalConnectedCap)) - v.priorityCount-- - v.clients[id] = c - go v.ps.Unregister(c.peer.id) - if v.priorityCount <= count && v.totalConnectedCap <= totalCap { - break - } - } - } - } - v.maxPeers = count - v.totalCap = totalCap - if v.child != nil { - v.child.setLimits(v.maxPeers-v.priorityCount, v.totalCap-v.totalConnectedCap) - } -} - -// totalCapacity queries total available capacity for all clients -func (v *priorityClientPool) totalCapacity() uint64 { - v.lock.Lock() - defer v.lock.Unlock() - - return v.totalCapAnnounced -} - -// subscribeTotalCapacity subscribes to changed total capacity events -func (v *priorityClientPool) subscribeTotalCapacity(sub *tcSubscription) { - v.lock.Lock() - defer v.lock.Unlock() - - v.subs[sub] = struct{}{} -} - -// setClientCapacity sets the priority capacity assigned to a given client -func (v *priorityClientPool) setClientCapacity(id enode.ID, cap uint64) error { - v.lock.Lock() - defer v.lock.Unlock() - - c := v.clients[id] - if c.cap == cap { - return nil - } - if c.connected { - if v.totalConnectedCap+cap > v.totalCap+c.cap { - return ErrTotalCap - } - if c.cap == 0 { - if v.child != nil { - v.child.unregisterPeer(c.peer) - } - v.priorityCount++ - } - if cap == 0 { - v.priorityCount-- - } - v.totalConnectedCap += cap - c.cap - v.logTotalPriConn.Update(float64(v.totalConnectedCap)) - if v.child != nil { - v.child.setLimits(v.maxPeers-v.priorityCount, v.totalCap-v.totalConnectedCap) - } - if cap == 0 { - if v.child != nil { - v.child.registerPeer(c.peer) - } - c.peer.updateCapacity(v.freeClientCap) - } else { - c.peer.updateCapacity(cap) - } - } - if cap != 0 || c.connected { - c.cap = cap - v.clients[id] = c - } else { - delete(v.clients, id) - } - if c.connected { - v.logger.Event(fmt.Sprintf("priorityClientPool: changed capacity to %d, %x", cap, id.Bytes())) - } - return nil -} - // Benchmark runs a request performance benchmark with a given set of measurement setups // in multiple passes specified by passCount. The measurement time for each setup in each // pass is specified in milliseconds by length. @@ -473,6 +253,605 @@ func (api *PrivateLightServerAPI) Benchmark(setups []map[string]interface{}, pas return result, nil } +// PrivateDebugAPI provides an API to debug LES light server functionality. +type PrivateDebugAPI struct { + server *LesServer +} + +// NewPrivateDebugAPI creates a new LES light server debug API. +func NewPrivateDebugAPI(server *LesServer) *PrivateDebugAPI { + return &PrivateDebugAPI{ + server: server, + } +} + +// FreezeClient forces a temporary client freeze which normally happens when the server is overloaded +func (api *PrivateDebugAPI) FreezeClient(id enode.ID) error { + return api.server.priorityClientPool.freezeClient(id) +} + +// priorityClientInfo entries exist for all prioritized clients and currently connected non-priority clients +type priorityClientInfo struct { + cap uint64 // zero for non-priority clients + connected bool + userTags map[string]struct{} + peer *peer + id enode.ID + timeFactor, capacityFactor, requestCostFactor float64 + priceUpdatePeriod uint64 + priceTracker priceTracker +} + +// matchTags checks whether the client matches the given tags +func (c *priorityClientInfo) matchTags(tags []string) bool { + for _, tag := range tags { + if len(tag) > 0 && tag[0] == '$' { + switch tag { + case "$all": + case "$connected": + if !c.connected { + return false + } + case "$disconnected": + if c.connected { + return false + } + case "$priority": + if c.cap == 0 { + return false + } + case "$free": + if c.cap != 0 { + return false + } + default: + return false + } + } else { + if _, ok := c.userTags[tag]; !ok { + return false + } + } + } + return true +} + +type ( + // clientPool is implemented by both the free and priority client pools + clientPool interface { + peerSetNotify + setLimits(count int, totalCap uint64) + } + // priorityClientPool stores information about prioritized clients + priorityClientPool struct { + lock sync.Mutex + child clientPool + ps *peerSet + clients map[enode.ID]*priorityClientInfo + totalCap, totalCapAnnounced uint64 + totalConnectedCap, totalAssignedCap, freeClientCap uint64 + maxPeers, freeCount, priorityCount int + logger *csvlogger.Logger + logTotalPriConn *csvlogger.Channel + + subs map[*eventSub]struct{} + updateSchedule []scheduledUpdate + scheduleCounter uint64 + } + // scheduledUpdate represents a delayed total capacity update + scheduledUpdate struct { + time mclock.AbsTime + totalCap, id uint64 + } +) + +// newPriorityClientPool creates a new priority client pool +func newPriorityClientPool(freeClientCap uint64, ps *peerSet, child clientPool, metricsLogger, eventLogger *csvlogger.Logger) *priorityClientPool { + return &priorityClientPool{ + clients: make(map[enode.ID]*priorityClientInfo), + freeClientCap: freeClientCap, + ps: ps, + child: child, + logger: eventLogger, + logTotalPriConn: metricsLogger.NewChannel("totalPriConn", 0), + } +} + +// registerPeer is called when a new client is connected. If the client has no +// priority assigned then it is passed to the child pool which may either keep it +// or disconnect it. +// +// Note: priorityClientPool also stores a record about free clients while they are +// connected in order to be able to assign priority to them later. +func (v *priorityClientPool) registerPeer(p *peer) { + v.lock.Lock() + defer v.lock.Unlock() + + id := p.ID() + c := v.clients[id] + if c == nil { + c = &priorityClientInfo{id: id} + v.clients[id] = c + } + v.logger.Event(fmt.Sprintf("priorityClientPool: registerPeer cap=%d connected=%v, %x", c.cap, c.connected, id.Bytes())) + if c.connected { + return + } + if c.cap == 0 && v.child != nil { + v.child.registerPeer(p) + v.freeCount++ + } + if c.cap != 0 && v.totalConnectedCap+c.cap > v.totalCap { + v.logger.Event(fmt.Sprintf("priorityClientPool: rejected, %x", id.Bytes())) + go v.ps.Unregister(p.id) + return + } + + c.connected = true + v.updatePriceFactors(c) + c.peer = p + p.priceTracker = &c.priceTracker + if c.cap != 0 { + v.priorityCount++ + v.totalConnectedCap += c.cap + v.logger.Event(fmt.Sprintf("priorityClientPool: accepted with %d capacity, %x", c.cap, id.Bytes())) + v.logTotalPriConn.Update(float64(v.totalConnectedCap)) + if v.child != nil { + v.child.setLimits(v.maxPeers-v.priorityCount, v.totalCap-v.totalConnectedCap) + } + p.updateCapacity(c.cap) + } + v.sendEvent("connect", c) +} + +// unregisterPeer is called when a client is disconnected. If the client has no +// priority assigned then it is also removed from the child pool. +func (v *priorityClientPool) unregisterPeer(p *peer) { + v.lock.Lock() + defer v.lock.Unlock() + + id := p.ID() + c := v.clients[id] + if c == nil { + return + } + v.logger.Event(fmt.Sprintf("priorityClientPool: unregisterPeer cap=%d connected=%v, %x", c.cap, c.connected, id.Bytes())) + if !c.connected { + return + } + c.priceTracker.setFactors(0, 0) + if c.cap != 0 { + c.connected = false + v.priorityCount-- + v.totalConnectedCap -= c.cap + v.logTotalPriConn.Update(float64(v.totalConnectedCap)) + if v.child != nil { + v.child.setLimits(v.maxPeers-v.priorityCount, v.totalCap-v.totalConnectedCap) + } + } else { + if v.child != nil { + v.child.unregisterPeer(p) + v.freeCount-- + } + delete(v.clients, id) + } + v.sendEvent("disconnect", c) +} + +// clientInfo creates a client info data structure +func (v *priorityClientPool) clientInfo(c *priorityClientInfo) map[string]interface{} { + clientInfo := make(map[string]interface{}) + clientInfo["isConnected"] = c.connected + cap := c.cap + pri := true + if cap == 0 { + cap = v.freeClientCap + pri = false + } + clientInfo["capacity"] = cap + clientInfo["hasPriority"] = pri + tags := make([]string, 0, len(c.userTags)) + for tag, _ := range c.userTags { + tags = append(tags, tag) + } + clientInfo["userTags"] = tags + clientInfo["pricing/totalAmount"] = c.priceTracker.getTotalAmount() + return clientInfo +} + +// sendEvent sends an event to the subscribers interested in it. For global events client == nil. +func (v *priorityClientPool) sendEvent(clientEvent string, client *priorityClientInfo) { + var event map[string]interface{} + makeEvent := func() { + event = make(map[string]interface{}) + event["totalCapacity"] = v.totalCap + event["totalConnectedCapacity"] = v.totalConnectedCap + uint64(v.freeCount)*v.freeClientCap + event["priorityConnectedCapacity"] = v.totalConnectedCap + if client != nil { + event["clientEvent"] = clientEvent + event["clientId"] = client.id + event["clientInfo"] = v.clientInfo(client) + } + } + + for sub := range v.subs { + select { + case <-sub.rpcSub.Err(): + delete(v.subs, sub) + case <-sub.notifier.Closed(): + delete(v.subs, sub) + default: + var send bool + if client == nil { + send = v.totalCap < v.totalConnectedCap || !sub.totalCapUnderrun + } else { + send = client.matchTags(sub.clientTags) + } + if send { + if event == nil { + makeEvent() + } + sub.notifier.Notify(sub.rpcSub.ID, event) + } + } + } +} + +// matchClients calls the given callback for all clients in the ids list or matching the +// given tags. If an unknown client is listed in ids it is temporarily created but only +// kept if the callback has assigned a priority to it. +func (v *priorityClientPool) matchClients(ids []enode.ID, tags []string, cb func(client *priorityClientInfo)) { + v.lock.Lock() + defer v.lock.Unlock() + + for _, id := range ids { + c := v.clients[id] + if c == nil { + c = &priorityClientInfo{id: id} + v.clients[id] = c + } + cb(c) + if c.cap == 0 && !c.connected { + delete(v.clients, id) + } + } + if len(tags) > 0 { + for _, info := range v.clients { + if info.matchTags(tags) { + cb(info) + } + } + } +} + +// priceUpdate sends a price update client event and schedules a new update with the +// price tracker if necessary. +func (v *priorityClientPool) priceUpdate(client *priorityClientInfo) { + v.lock.Lock() + defer v.lock.Unlock() + + v.sendEvent("priceUpdate", client) + if client.priceUpdatePeriod != 0 { + v.setPriceUpdate(client, client.priceUpdatePeriod, true) + } +} + +// setPriceUpdate schedules a price update when the total price reaches the given limit. +// If periodic is false then the limit is interpreted as an absolute value while if true +// it is relative to the current totalAmount value or the its value at the last future update. +func (v *priorityClientPool) setPriceUpdate(client *priorityClientInfo, value uint64, periodic bool) { + if value == 0 { + client.priceTracker.setCallback(0, nil) + return + } + if periodic { + client.priceUpdatePeriod = value + value += client.priceTracker.getTotalAmount() + } else { + client.priceUpdatePeriod = 0 + } + client.priceTracker.setCallback(value, func() { v.priceUpdate(client) }) +} + +// setLimits updates the allowed peer count and total capacity of the priority +// client pool. Since the free client pool is a child of the priority pool the +// remaining peer count and capacity is assigned to the free pool by calling its +// own setLimits function. +// +// Note: a decreasing change of the total capacity is applied with a delay. +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.sendEvent("", nil) + return + } + v.setLimitsNow(count, v.totalCap) + if totalCap < v.totalCap { + v.sendEvent("", nil) + for i, s := range v.updateSchedule { + if totalCap >= s.totalCap { + s.totalCap = totalCap + v.updateSchedule = v.updateSchedule[:i+1] + return + } + } + v.updateSchedule = append(v.updateSchedule, scheduledUpdate{time: mclock.Now() + mclock.AbsTime(dropCapacityDelay), totalCap: totalCap}) + if len(v.updateSchedule) == 1 { + v.scheduleCounter++ + id := v.scheduleCounter + v.updateSchedule[0].id = id + time.AfterFunc(dropCapacityDelay, func() { v.checkUpdate(id) }) + } + } else { + v.updateSchedule = nil + } +} + +// checkUpdate performs the next scheduled update if possible and schedules +// the one after that +func (v *priorityClientPool) checkUpdate(id uint64) { + v.lock.Lock() + defer v.lock.Unlock() + + if len(v.updateSchedule) == 0 || v.updateSchedule[0].id != id { + return + } + v.setLimitsNow(v.maxPeers, v.updateSchedule[0].totalCap) + v.updateSchedule = v.updateSchedule[1:] + if len(v.updateSchedule) != 0 { + v.scheduleCounter++ + id := v.scheduleCounter + v.updateSchedule[0].id = id + dt := time.Duration(v.updateSchedule[0].time - mclock.Now()) + time.AfterFunc(dt, func() { v.checkUpdate(id) }) + } +} + +// setLimits updates the allowed peer count and total capacity immediately +func (v *priorityClientPool) setLimitsNow(count int, totalCap uint64) { + if v.priorityCount > count || v.totalConnectedCap > totalCap { + for id, c := range v.clients { + if c.connected { + v.logger.Event(fmt.Sprintf("priorityClientPool: setLimitsNow kicked out, %x", id.Bytes())) + c.connected = false + v.totalConnectedCap -= c.cap + v.logTotalPriConn.Update(float64(v.totalConnectedCap)) + v.priorityCount-- + v.sendEvent("disconnect", c) + go v.ps.Unregister(c.peer.id) + if v.priorityCount <= count && v.totalConnectedCap <= totalCap { + break + } + } + } + } + v.maxPeers = count + v.totalCap = totalCap + if v.child != nil { + v.child.setLimits(v.maxPeers-v.priorityCount, v.totalCap-v.totalConnectedCap) + } +} + +// capacityInfo queries total available, connected and assigned capacity +func (v *priorityClientPool) capacityInfo() (total, conn, priConn, priAssigned uint64) { + v.lock.Lock() + defer v.lock.Unlock() + + return v.totalCapAnnounced, v.totalConnectedCap + uint64(v.freeCount)*v.freeClientCap, v.totalConnectedCap, v.totalAssignedCap +} + +// subscribeEvents subscribes to events +func (v *priorityClientPool) subscribeEvents(sub *eventSub) { + v.lock.Lock() + defer v.lock.Unlock() + + v.subs[sub] = struct{}{} +} + +// setClientTags sets the user tags associated with the client +func (v *priorityClientPool) setClientTags(c *priorityClientInfo, tags []string) error { + if len(tags) == 0 { + c.userTags = nil + } else { + c.userTags = make(map[string]struct{}) + for _, tag := range tags { + if len(tag) > 0 && tag[0] == '$' { + return ErrInvalidTag + } + c.userTags[tag] = struct{}{} + } + } + return nil +} + +// setClientCapacity sets the priority capacity assigned to a given client +func (v *priorityClientPool) setClientCapacity(c *priorityClientInfo, cap uint64) error { + if c.cap == cap { + return nil + } + v.totalAssignedCap += cap - c.cap + if c.connected { + if v.totalConnectedCap+cap > v.totalCap+c.cap { + return ErrTotalCap + } + if c.cap == 0 { + if v.child != nil { + v.child.unregisterPeer(c.peer) + v.freeCount-- + } + v.priorityCount++ + } + if cap == 0 { + v.priorityCount-- + } + v.totalConnectedCap += cap - c.cap + v.logTotalPriConn.Update(float64(v.totalConnectedCap)) + if v.child != nil { + v.child.setLimits(v.maxPeers-v.priorityCount, v.totalCap-v.totalConnectedCap) + } + if cap == 0 { + if v.child != nil { + v.child.registerPeer(c.peer) + v.freeCount++ + } + c.peer.updateCapacity(v.freeClientCap) + } else { + c.peer.updateCapacity(cap) + } + } + if cap != 0 || c.connected { + c.cap = cap + } else { + delete(v.clients, c.id) + } + v.sendEvent("updateCapacity", c) + if c.connected { + v.logger.Event(fmt.Sprintf("priorityClientPool: changed capacity to %d, %x", cap, c.id.Bytes())) + } + return nil +} + +// freezeClient forces a temporary client freeze which normally happens when the server is overloaded +func (v *priorityClientPool) freezeClient(id enode.ID) error { + v.lock.Lock() + defer v.lock.Unlock() + + c := v.clients[id] + if c != nil && c.peer != nil { + c.peer.freezeClient() + return nil + } else { + return ErrClientNotConnected + } +} + +// updatePriceFactors updates the price tracker for a client based on current parameters +func (v *priorityClientPool) updatePriceFactors(c *priorityClientInfo) { + cap := c.cap + if cap == 0 { + cap = v.freeClientCap + } + c.priceTracker.setFactors(c.timeFactor+c.capacityFactor*float64(cap)/1000000, c.requestCostFactor) +} + +// priceTracker calculates service price for a single client based on connection time +// and/or requests served. It also provides a callback function when a given total +// price threshold has been reached. +type priceTracker struct { + lock sync.Mutex + totalAmount, callbackThreshold uint64 + callback func() + timeFactor, requestFactor float64 + lastUpdate, nextUpdate mclock.AbsTime + updateTimer *time.Timer +} + +// update recalculates the total price, calls the callback and/or modifies +// the next scheduled update if necessary +func (pt *priceTracker) update() { + now := mclock.Now() + if pt.lastUpdate != 0 { + dt := now - pt.lastUpdate + if dt > 0 { + pt.totalAmount += uint64(pt.timeFactor * float64(dt)) + } + } + pt.lastUpdate = now + if pt.callbackThreshold == 0 || pt.callback == nil { + pt.nextUpdate = 0 + pt.updateAfter(0) + } else { + if pt.totalAmount >= pt.callbackThreshold { + pt.callbackThreshold = 0 + pt.nextUpdate = 0 + go pt.callback() + } else { + if pt.timeFactor > 1e-100 { + dt := float64(pt.callbackThreshold-pt.totalAmount) / pt.timeFactor + if dt > 1e15 { + dt = 1e15 + } + d := time.Duration(dt) + if pt.nextUpdate == 0 || pt.nextUpdate > now+mclock.AbsTime(d) { + if d > time.Second { + d = ((d - time.Second) * 7 / 8) + time.Second + } + pt.nextUpdate = now + mclock.AbsTime(d) + pt.updateAfter(d) + } + } else { + pt.nextUpdate = 0 + pt.updateAfter(0) + } + } + } +} + +// updateAfter schedules an update in the future +func (pt *priceTracker) updateAfter(dt time.Duration) { + if pt.updateTimer == nil || pt.updateTimer.Stop() { + if dt == 0 { + pt.updateTimer = nil + } else { + pt.updateTimer = time.AfterFunc(dt, func() { + pt.lock.Lock() + defer pt.lock.Unlock() + + if pt.callbackThreshold != 0 { + pt.update() + } + }) + } + } +} + +// requestCost should be called after serving a request for the given peer +func (pt *priceTracker) requestCost(cost uint64) { + pt.lock.Lock() + defer pt.lock.Unlock() + + if pt.requestFactor != 0 { + pt.totalAmount += uint64(float64(cost) * pt.requestFactor) + pt.update() + } +} + +// getTotalAmount returns the current total cost accumulated. +func (pt *priceTracker) getTotalAmount() uint64 { + pt.lock.Lock() + defer pt.lock.Unlock() + + pt.update() + return pt.totalAmount +} + +// setFactors sets the price factors. timeFactor is the price of a nanosecond of +// connection while requestFactor is the price of a "realCost" unit. +func (pt *priceTracker) setFactors(timeFactor, requestFactor float64) { + pt.lock.Lock() + defer pt.lock.Unlock() + + pt.update() + pt.timeFactor = timeFactor + pt.requestFactor = requestFactor +} + +// setCallback sets up a one-time callback to be called when totalAmount reaches +// the threshold. If it has already reached the threshold the callback is called +// immediately. +func (pt *priceTracker) setCallback(threshold uint64, callback func()) { + pt.lock.Lock() + defer pt.lock.Unlock() + + pt.callbackThreshold = threshold + pt.callback = callback + pt.update() +} + // PrivateLightAPI provides an API to access the LES light server or light client. type PrivateLightAPI struct { backend *lesCommons diff --git a/les/api_test.go b/les/api_test.go index cec9459625..4a592db22f 100644 --- a/les/api_test.go +++ b/les/api_test.go @@ -97,17 +97,14 @@ func testCapacityAPI(t *testing.T, clientCount int) { t.Fatalf("Failed to obtain rpc client: %v", err) } headNum, headHash := getHead(ctx, t, serverRpcClient) - totalCap := getTotalCap(ctx, t, serverRpcClient) - minCap := getMinCap(ctx, t, serverRpcClient) + minCap, freeCap, totalCap := getCapacityInfo(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 := getFreeCap(ctx, t, serverRpcClient) for i, client := range clients { var err error @@ -147,7 +144,7 @@ func testCapacityAPI(t *testing.T, clientCount int) { i, c := i, c go func() { queue := make(chan struct{}, 100) - var count uint64 + reqCount[i] = 0 for { select { case queue <- struct{}{}: @@ -165,8 +162,11 @@ func testCapacityAPI(t *testing.T, clientCount int) { wg.Done() <-queue if ok { - count++ - atomic.StoreUint64(&reqCount[i], count) + count := atomic.AddUint64(&reqCount[i], 1) + if count%10000 == 0 { + fmt.Println("freeze", clients[i].ID()) + freezeClient(ctx, t, serverRpcClient, clients[i].ID()) + } } }() } @@ -239,7 +239,7 @@ func testCapacityAPI(t *testing.T, clientCount int) { default: } - totalCap = getTotalCap(ctx, t, serverRpcClient) + _, _, totalCap = getCapacityInfo(ctx, t, serverRpcClient) if totalCap < testCap { fmt.Println("Total capacity underrun") close(stop) @@ -328,58 +328,61 @@ func testRequest(ctx context.Context, t *testing.T, client *rpc.Client) bool { return err == nil } +func freezeClient(ctx context.Context, t *testing.T, server *rpc.Client, clientID enode.ID) { + if err := server.CallContext(ctx, nil, "debug_freezeClient", clientID); err != nil { + t.Fatalf("Failed to freeze client: %v", err) + } + +} + 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 { + params := make(map[string]interface{}) + params["capacity"] = cap + if err := server.CallContext(ctx, nil, "les_setClientParams", []enode.ID{clientID}, []string{}, params); err != nil { t.Fatalf("Failed to set client capacity: %v", err) } } func getCapacity(ctx context.Context, t *testing.T, server *rpc.Client, clientID enode.ID) uint64 { - var s string - if err := server.CallContext(ctx, &s, "les_getClientCapacity", clientID); err != nil { - t.Fatalf("Failed to get client capacity: %v", err) + var res map[enode.ID]map[string]interface{} + if err := server.CallContext(ctx, &res, "les_clientInfo", []enode.ID{clientID}, []string{}); err != nil { + t.Fatalf("Failed to get client info: %v", err) } - cap, err := hexutil.DecodeUint64(s) - if err != nil { - t.Fatalf("Failed to decode client capacity: %v", err) + info, ok := res[clientID] + if !ok { + t.Fatalf("Missing client info") } - return cap + v, ok := info["capacity"] + if !ok { + t.Fatalf("Missing field in client info: capacity") + } + vv, ok := v.(float64) + if !ok { + t.Fatalf("Failed to decode capacity field") + } + return uint64(vv) } -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) +func getCapacityInfo(ctx context.Context, t *testing.T, server *rpc.Client) (minCap, freeCap, totalCap uint64) { + var res map[string]interface{} + if err := server.CallContext(ctx, &res, "les_serverInfo"); err != nil { + t.Fatalf("Failed to query server info: %v", err) } - total, err := hexutil.DecodeUint64(s) - if err != nil { - t.Fatalf("Failed to decode total capacity: %v", err) + decode := func(s string) uint64 { + v, ok := res[s] + if !ok { + t.Fatalf("Missing field in server info: %s", s) + } + vv, ok := v.(float64) + if !ok { + t.Fatalf("Failed to decode server info field: %s", s) + } + return uint64(vv) } - 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) - } - min, err := hexutil.DecodeUint64(s) - if err != nil { - t.Fatalf("Failed to decode minimum capacity: %v", err) - } - 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 + minCap = decode("minimumCapacity") + freeCap = decode("freeClientCapacity") + totalCap = decode("totalCapacity") + return } func init() { diff --git a/les/handler.go b/les/handler.go index c043755326..5acdd9a7c6 100644 --- a/les/handler.go +++ b/les/handler.go @@ -408,6 +408,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { realCost = pm.server.costTracker.realCost(servingTime, msg.Size, replySize) if amount != 0 { pm.server.costTracker.updateStats(msg.Code, amount, servingTime, realCost) + p.priceTracker.requestCost(realCost) } } else { realCost = maxCost diff --git a/les/peer.go b/les/peer.go index 879e9f74c7..a5e2d85c9c 100644 --- a/les/peer.go +++ b/les/peer.go @@ -103,10 +103,11 @@ type peer struct { updateTime mclock.AbsTime frozen uint32 // 1 if client is in frozen state - fcClient *flowcontrol.ClientNode // nil if the peer is server only - fcServer *flowcontrol.ServerNode // nil if the peer is client only - fcParams flowcontrol.ServerParams - fcCosts requestCostTable + fcClient *flowcontrol.ClientNode // nil if the peer is server only + fcServer *flowcontrol.ServerNode // nil if the peer is client only + fcParams flowcontrol.ServerParams + fcCosts requestCostTable + priceTracker *priceTracker isTrusted bool isOnlyAnnounce bool diff --git a/les/server.go b/les/server.go index 5b5c0cae84..4ac0b2a6d6 100644 --- a/les/server.go +++ b/les/server.go @@ -137,13 +137,19 @@ func (s *LesServer) APIs() []rpc.API { { Namespace: "les", Version: "1.0", - Service: NewPrivateLightServerAPI(s), + Service: NewPrivateLightAPI(&s.lesCommons, s.protocolManager.reg), Public: false, }, { Namespace: "les", Version: "1.0", - Service: NewPrivateLightAPI(&s.lesCommons, s.protocolManager.reg), + Service: NewPrivateLightServerAPI(s), + Public: false, + }, + { + Namespace: "debug", + Version: "1.0", + Service: NewPrivateDebugAPI(s), Public: false, }, }