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: added and cleaned up docs

Browse source
This commit is contained in:
Zsolt Felfoldi 2019-01-25 20:23:29 +01:00
parent e8a5843903
commit 0c3b122fb2
23 changed files with 116 additions and 85 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

87
les/api.go
View file

@ -13,6 +13,7 @@
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package les
import (
@ -34,25 +35,32 @@ var (
dropCapacityDelay = time.Second
)
// PublicLesServerAPI provides an API to access the les server.
// 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 PrivateLesServerAPI struct {
type PrivateLightServerAPI struct {
server *LesServer
}
// NewPublicLesServerAPI creates a new les server API.
func NewPrivateLesServerAPI(server *LesServer) *PrivateLesServerAPI {
return &PrivateLesServerAPI{
// NewPrivateLightServerAPI creates a new LES light server API.
func NewPrivateLightServerAPI(server *LesServer) *PrivateLightServerAPI {
return &PrivateLightServerAPI{
server: server,
}
}
// TotalCapacity queries total available capacity for all clients
func (api *PrivateLesServerAPI) TotalCapacity() hexutil.Uint64 {
func (api *PrivateLightServerAPI) TotalCapacity() hexutil.Uint64 {
return hexutil.Uint64(api.server.priorityClientPool.totalCapacity())
}
func (api *PrivateLesServerAPI) SubscribeTotalCapacity(ctx context.Context, onlyUnderrun bool) (*rpc.Subscription, error) {
// 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) {
notifier, supported := rpc.NotifierFromContext(ctx)
if !supported {
return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported
@ -63,6 +71,7 @@ func (api *PrivateLesServerAPI) SubscribeTotalCapacity(ctx context.Context, only
}
type (
// tcSubscription represents a total capacity subscription
tcSubscription struct {
notifier *rpc.Notifier
rpcSub *rpc.Subscription
@ -71,6 +80,7 @@ type (
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 {
@ -87,14 +97,37 @@ func (s tcSubs) send(tc uint64, underrun bool) {
}
// MinimumCapacity queries minimum assignable capacity for a single client
func (api *PrivateLesServerAPI) MinimumCapacity() hexutil.Uint64 {
func (api *PrivateLightServerAPI) MinimumCapacity() hexutil.Uint64 {
return hexutil.Uint64(minCapacity)
}
func (api *PrivateLesServerAPI) FreeClientCapacity() hexutil.Uint64 {
// 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 < 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)
}
type clientPool interface {
peerSetNotify
setLimits(count int, totalCap uint64)
@ -115,6 +148,7 @@ type priorityClientPool struct {
scheduleCounter uint64
}
// scheduledUpdate represents a delayed total capacity update
type scheduledUpdate struct {
time mclock.AbsTime
totalCap, id uint64
@ -127,28 +161,7 @@ type priorityClientInfo struct {
peer *peer
}
// 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 *PrivateLesServerAPI) SetClientCapacity(id enode.ID, cap uint64) error {
if cap != 0 && cap < minCapacity {
return ErrMinCap
}
return api.server.priorityClientPool.setClientCapacity(id, cap)
}
// GetClientCapacity returns the capacity assigned to a given client
func (api *PrivateLesServerAPI) 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)
}
// newPriorityClientPool creates a new priority client pool
func newPriorityClientPool(freeClientCap uint64, ps *peerSet, child clientPool) *priorityClientPool {
return &priorityClientPool{
clients: make(map[enode.ID]priorityClientInfo),
@ -223,6 +236,12 @@ func (v *priorityClientPool) unregisterPeer(p *peer) {
}
}
// 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()
@ -255,6 +274,8 @@ func (v *priorityClientPool) setLimits(count int, totalCap uint64) {
}
}
// 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()
@ -273,6 +294,7 @@ func (v *priorityClientPool) checkUpdate(id uint64) {
}
}
// 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 {
@ -296,6 +318,7 @@ func (v *priorityClientPool) setLimitsNow(count int, totalCap uint64) {
}
}
// totalCapacity queries total available capacity for all clients
func (v *priorityClientPool) totalCapacity() uint64 {
v.lock.Lock()
defer v.lock.Unlock()
@ -303,6 +326,7 @@ func (v *priorityClientPool) totalCapacity() uint64 {
return v.totalCapAnnounced
}
// subscribeTotalCapacity subscribes to changed total capacity events
func (v *priorityClientPool) subscribeTotalCapacity(sub *tcSubscription) {
v.lock.Lock()
defer v.lock.Unlock()
@ -310,6 +334,7 @@ func (v *priorityClientPool) subscribeTotalCapacity(sub *tcSubscription) {
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()

4
les/api_test.go
View file

@ -71,10 +71,10 @@ func TestCapacityAPI10(t *testing.T) {
}
// testCapacityAPI runs an end-to-end simulation test connecting one server with
// a given number of clients. It sets different priority capacitys to all clients
// a given number of clients. It sets different priority capacities to all clients
// except a randomly selected one which runs in free client mode. All clients send
// similar requests at the maximum allowed rate and the test verifies whether the
// ratio of processed requests is close enough to the ratio of assigned capacitys.
// ratio of processed requests is close enough to the ratio of assigned capacities.
// Running multiple rounds with different settings ensures that changing capacity
// while connected and going back and forth between free and priority mode with
// the supplied API calls is also thoroughly tested.

1
les/benchmark.go
View file

@ -13,6 +13,7 @@
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
package les
import (

2
les/distributor.go
View file

@ -14,8 +14,6 @@
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
// Package light implements on-demand retrieval capable state and chain objects
// for the Ethereum Light Client.
package les
import (

2
les/distributor_test.go
View file

@ -14,8 +14,6 @@
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
// Package light implements on-demand retrieval capable state and chain objects
// for the Ethereum Light Client.
package les
import (

1
les/fetcher.go
View file

@ -14,7 +14,6 @@
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
// Package les implements the Light Ethereum Subprotocol.
package les
import (

13
les/flowcontrol/control.go
View file

@ -46,6 +46,7 @@ type ServerParams struct {
BufLimit, MinRecharge uint64
}
// scheduledUpdate represents a delayed flow control parameter update
type scheduledUpdate struct {
time mclock.AbsTime
params ServerParams
@ -78,14 +79,17 @@ func NewClientNode(cm *ClientManager, params ServerParams) *ClientNode {
if keepLogs > 0 {
node.log = newLogger(keepLogs)
}
cm.init(node)
cm.connect(node)
return node
}
// Disconnect should be called when a client is disconnected
func (node *ClientNode) Disconnect() {
node.cm.disconnect(node)
}
// update recalculates the buffer value at a specified time while also performing
// scheduled flow control parameter updates if necessary
func (node *ClientNode) update(now mclock.AbsTime) {
for len(node.updateSchedule) > 0 && node.updateSchedule[0].time <= now {
node.recalcBV(node.updateSchedule[0].time)
@ -95,6 +99,7 @@ func (node *ClientNode) update(now mclock.AbsTime) {
node.recalcBV(now)
}
// recalcBV recalculates the buffer value at a specified time
func (node *ClientNode) recalcBV(now mclock.AbsTime) {
dt := uint64(now - node.lastTime)
if now < node.lastTime {
@ -110,6 +115,7 @@ func (node *ClientNode) recalcBV(now mclock.AbsTime) {
node.lastTime = now
}
// UpdateParams updates the flow control parameters of a client node
func (node *ClientNode) UpdateParams(params ServerParams) {
node.lock.Lock()
defer node.lock.Unlock()
@ -131,6 +137,7 @@ func (node *ClientNode) UpdateParams(params ServerParams) {
}
}
// updateParams updates the flow control parameters of the node
func (node *ClientNode) updateParams(params ServerParams, now mclock.AbsTime) {
diff := params.BufLimit - node.params.BufLimit
if int64(diff) > 0 {
@ -212,7 +219,7 @@ func NewServerNode(params ServerParams, clock mclock.Clock) *ServerNode {
return node
}
// UpdateParams updates flow control parameters
// UpdateParams updates the flow control parameters of the node
func (node *ServerNode) UpdateParams(params ServerParams) {
node.lock.Lock()
defer node.lock.Unlock()
@ -228,6 +235,8 @@ func (node *ServerNode) UpdateParams(params ServerParams) {
node.params = params
}
// recalcBLE recalculates the lowest estimate for the client's buffer value at
// the given server at the specified time
func (node *ServerNode) recalcBLE(now mclock.AbsTime) {
if now < node.lastTime {
return

1
les/flowcontrol/logger.go
View file

@ -14,7 +14,6 @@
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
// Package flowcontrol implements a client side flow control mechanism
package flowcontrol
import (

59
les/flowcontrol/manager.go
View file

@ -14,7 +14,6 @@
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
// Package flowcontrol implements a client side flow control mechanism
package flowcontrol
import (
@ -48,9 +47,9 @@ type cmNodeFields struct {
const FixedPointMultiplier = 1000000
var (
capFactorDropTC = time.Second * 10
capFactorRaiseTC = time.Hour
capFactorRaiseThreshold = 0.75
capFactorDropTC = 1 / float64(time.Second*10) // time constant for dropping the capacity factor
capFactorRaiseTC = 1 / float64(time.Hour) // time constant for raising the capacity factor
capFactorRaiseThreshold = 0.75 // connected / total capacity ratio threshold for raising the capacity factor
)
// ClientManager controls the capacity assigned to the clients of a server.
@ -131,8 +130,9 @@ func (cm *ClientManager) SetRechargeCurve(curve PieceWiseLinear) {
cm.refreshCapacity()
}
// init initializes the ClientManager specific fields of a ClientNode structure
func (cm *ClientManager) init(node *ClientNode) {
// connect should be called when a client is connected, before passing it to any
// other ClientManager funtion
func (cm *ClientManager) connect(node *ClientNode) {
cm.lock.Lock()
defer cm.lock.Unlock()
@ -145,6 +145,7 @@ func (cm *ClientManager) init(node *ClientNode) {
cm.totalConnected += node.params.MinRecharge
}
// disconnect should be called when a client is disconnected
func (cm *ClientManager) disconnect(node *ClientNode) {
cm.lock.Lock()
defer cm.lock.Unlock()
@ -155,8 +156,10 @@ func (cm *ClientManager) disconnect(node *ClientNode) {
cm.totalConnected -= node.params.MinRecharge
}
// accepted deduces the upper estimate for request cost from the buffer and returns a priority
// value based on current buffer status which is used by the serving queue.
// accepted is called when a request with given maximum cost is accepted.
// It returns a priority indicator for the request which is used to determine placement
// in the serving queue. Older requests have higher priority by default. If the client
// is almost out of buffer, request priority is reduced.
func (cm *ClientManager) accepted(node *ClientNode, maxCost uint64, now mclock.AbsTime) (priority int64) {
cm.lock.Lock()
defer cm.lock.Unlock()
@ -186,6 +189,7 @@ func (cm *ClientManager) processed(node *ClientNode, maxCost, realCost uint64, n
}
}
// updateParams updates the flow control parameters of a client node
func (cm *ClientManager) updateParams(node *ClientNode, params ServerParams, now mclock.AbsTime) {
cm.lock.Lock()
defer cm.lock.Unlock()
@ -280,25 +284,38 @@ func (cm *ClientManager) updateNodeRc(node *ClientNode, bvc int64, params *Serve
}
// updateCapFactor updates the total capacity factor. The capacity factor allows
// the total capacity of the system to go over the allowed total recharge value
// if the sum of momentarily recharging clients only exceeds the total recharge
// allowance in a very small fraction of time.
// The capacity factor is dropped quickly (with a small time constant) if sumRecharge
// exceeds totalRecharge. It is raised slowly (with a large time constant) if most
// of the total capacity is used by connected clients (totalConnected is larger than
// totalCapacity*capFactorRaiseThreshold) and sumRecharge stays under
// totalRecharge*totalConnected/totalCapacity.
func (cm *ClientManager) updateCapFactor(now mclock.AbsTime, refresh bool) {
if cm.totalRecharge == 0 {
return
}
dt := now - cm.capLastUpdate
cm.capLastUpdate = now
var d float64
if cm.sumRecharge > cm.totalRecharge {
d = (1 - float64(cm.sumRecharge)/float64(cm.totalRecharge)) / float64(capFactorDropTC)
d = (1 - float64(cm.sumRecharge)/float64(cm.totalRecharge)) * capFactorDropTC
} else {
r := float64(cm.totalConnected) / cm.totalCapacity
if r > 1 {
r = 1
totalConnected := float64(cm.totalConnected)
var connRatio float64
if totalConnected < cm.totalCapacity {
connRatio = totalConnected / cm.totalCapacity
} else {
connRatio = 1
}
if r > capFactorRaiseThreshold {
m := cm.totalRecharge
if cm.totalConnected < m {
m = cm.totalConnected
}
if cm.sumRecharge < m {
d = (1 - float64(cm.sumRecharge)/float64(m)) * (r - capFactorRaiseThreshold) / (1 - capFactorRaiseThreshold) / float64(capFactorRaiseTC)
if connRatio > capFactorRaiseThreshold {
sumRecharge := float64(cm.sumRecharge)
limit := float64(cm.totalRecharge) * connRatio
if sumRecharge < limit {
d = (1 - sumRecharge/limit) * (connRatio - capFactorRaiseThreshold) * (1 / (1 - capFactorRaiseThreshold)) * capFactorRaiseTC
}
}
}
@ -313,6 +330,8 @@ func (cm *ClientManager) updateCapFactor(now mclock.AbsTime, refresh bool) {
}
}
// refreshCapacity recalculates the total capacity value and sends an update to the subscription
// channel if the relative change of the value since the last update is more than 0.1 percent
func (cm *ClientManager) refreshCapacity() {
totalCapacity := float64(cm.totalRecharge) * math.Exp(cm.capLogFactor)
if totalCapacity >= cm.totalCapacity*0.999 && totalCapacity <= cm.totalCapacity*1.001 {
@ -327,6 +346,8 @@ func (cm *ClientManager) refreshCapacity() {
}
}
// SubscribeTotalCapacity returns all future updates to the total capacity value
// through a channel and also returns the current value
func (cm *ClientManager) SubscribeTotalCapacity(ch chan uint64) uint64 {
cm.lock.Lock()
defer cm.lock.Unlock()

7
les/flowcontrol/manager_test.go
View file

@ -14,7 +14,6 @@
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
// Package flowcontrol implements a client side flow control mechanism
package flowcontrol
import (
@ -26,10 +25,10 @@ import (
)
type testNode struct {
node *ClientNode
node *ClientNode
bufLimit, capacity uint64
waitUntil mclock.AbsTime
index, totalCost uint64
waitUntil mclock.AbsTime
index, totalCost uint64
}
const (

3
les/freeclient.go
View file

@ -14,11 +14,9 @@
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
// Package les implements the Light Ethereum Subprotocol.
package les
import (
"fmt"
"io"
"math"
"net"
@ -189,7 +187,6 @@ 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

2
les/freeclient_test.go
View file

@ -14,8 +14,6 @@
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
// Package light implements on-demand retrieval capable state and chain objects
// for the Ethereum Light Client.
package les
import (

1
les/handler.go
View file

@ -14,7 +14,6 @@
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
// Package les implements the Light Ethereum Subprotocol.
package les
import (

2
les/odr_requests.go
View file

@ -14,8 +14,6 @@
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
// Package light implements on-demand retrieval capable state and chain objects
// for the Ethereum Light Client.
package les
import (

1
les/peer.go
View file

@ -14,7 +14,6 @@
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
// Package les implements the Light Ethereum Subprotocol.
package les
import (

1
les/protocol.go
View file

@ -14,7 +14,6 @@
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
// Package les implements the Light Ethereum Subprotocol.
package les
import (

1
les/randselect.go
View file

@ -14,7 +14,6 @@
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
// Package les implements the Light Ethereum Subprotocol.
package les
import (

2
les/retrieve.go
View file

@ -14,8 +14,6 @@
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
// Package light implements on-demand retrieval capable state and chain objects
// for the Ethereum Light Client.
package les
import (

3
les/server.go
View file

@ -14,7 +14,6 @@
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
// Package les implements the Light Ethereum Subprotocol.
package les
import (
@ -137,7 +136,7 @@ func (s *LesServer) APIs() []rpc.API {
{
Namespace: "les",
Version: "1.0",
Service: NewPrivateLesServerAPI(s),
Service: NewPrivateLightServerAPI(s),
Public: false,
},
}

1
les/serverpool.go
View file

@ -14,7 +14,6 @@
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
// Package les implements the Light Ethereum Subprotocol.
package les
import (

3
les/servingqueue.go
View file

@ -14,7 +14,6 @@
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
// Package flowcontrol implements a client side flow control mechanism
package les
import (
@ -93,8 +92,6 @@ func (sq *servingQueue) addTask(task *servingTask) {
// Note: either blocking should be false or currentTask should be nil.
func (sq *servingQueue) getNewTask(currentTask *servingTask, blocking bool) *servingTask {
sq.lock.Lock()
if sq.stopCount != 0 {
}
if sq.stopCount == 0 {
if sq.best != nil && (currentTask == nil || sq.best.priority <= currentTask.priority-sq.suspendBias) {
best := sq.best

2
light/lightchain.go
View file

@ -14,6 +14,8 @@
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
// Package light implements on-demand retrieval capable state and chain objects
// for the Ethereum Light Client.
package light
import (

2
light/odr.go
View file

@ -14,8 +14,6 @@
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
// Package light implements on-demand retrieval capable state and chain objects
// for the Ethereum Light Client.
package light
import (