server

Documentation

Index

• Variables
• type AtomicBalanceOperator
• type ClientPool
  • func NewClientPool(balanceDb ethdb.KeyValueStore, minCap uint64, connectedBias time.Duration, ...) *ClientPool
  • func (pp ClientPool) Active() (uint64, uint64)
  • func (bt ClientPool) BalanceOperation(id enode.ID, connAddress string, cb func(AtomicBalanceOperator))
  • func (bt ClientPool) GetExpirationTCs() (pos, neg uint64)
  • func (bt ClientPool) GetPosBalanceIDs(start, stop enode.ID, maxCount int) (result []enode.ID)
  • func (cp *ClientPool) Handle(id enode.ID, address string, name string, data []byte) []byte
  • func (pp ClientPool) Limits() (uint64, uint64)
  • func (cp *ClientPool) Register(peer clientPeer) ConnectedBalance
  • func (cp *ClientPool) SetCapacity(node *enode.Node, reqCap uint64, bias time.Duration, requested bool) (capacity uint64, err error)
  • func (cp *ClientPool) SetConnectedBias(bias time.Duration)
  • func (bt ClientPool) SetDefaultFactors(posFactors, negFactors PriceFactors)
  • func (bt ClientPool) SetExpirationTCs(pos, neg uint64)
  • func (pp ClientPool) SetLimits(maxCount, maxCap uint64)
  • func (cp *ClientPool) Start()
  • func (cp *ClientPool) Stop()
  • func (bt ClientPool) TotalTokenAmount() uint64
  • func (cp *ClientPool) Unregister(peer clientPeer)
• type ConnectedBalance
• type PriceFactors
• type ReadOnlyBalance
• type Server
  • func NewServer(delayPerRequest time.Duration) *Server
  • func (s *Server) Register(b Service, id, desc string)
  • func (s *Server) Serve(id enode.ID, address string, requests vflux.Requests) vflux.Replies
  • func (s *Server) ServeEncoded(id enode.ID, addr *net.UDPAddr, req []byte) []byte
  • func (s *Server) Stop()
• type Service

Variables

var (
	ErrNotConnected    = errors.New("client not connected")
	ErrNoPriority      = errors.New("priority too low to raise capacity")
	ErrCantFindMaximum = errors.New("Unable to find maximum allowed capacity")
)

Types

type AtomicBalanceOperator

type AtomicBalanceOperator interface {
	ReadOnlyBalance
	AddBalance(amount int64) (uint64, uint64, error)
	SetBalance(pos, neg uint64) error
}

AtomicBalanceOperator provides operations permitted in an atomic BalanceOperation

type ClientPool

type ClientPool struct {
	// contains filtered or unexported fields
}

ClientPool implements a client database that assigns a priority to each client based on a positive and negative balance. Positive balance is externally assigned to prioritized clients and is decreased with connection time and processed requests (unless the price factors are zero). If the positive balance is zero then negative balance is accumulated.

Balance tracking and priority calculation for connected clients is done by balanceTracker. PriorityQueue ensures that clients with the lowest positive or highest negative balance get evicted when the total capacity allowance is full and new clients with a better balance want to connect.

Already connected nodes receive a small bias in their favor in order to avoid accepting and instantly kicking out clients. In theory, we try to ensure that each client can have several minutes of connection time.

Balances of disconnected clients are stored in nodeDB including positive balance and negative banalce. Boeth positive balance and negative balance will decrease exponentially. If the balance is low enough, then the record will be dropped.

func NewClientPool

func NewClientPool(balanceDb ethdb.KeyValueStore, minCap uint64, connectedBias time.Duration, clock mclock.Clock, synced func() bool) *ClientPool

NewClientPool creates a new client pool

func (ClientPool) Active

func (pp ClientPool) Active() (uint64, uint64)

Active returns the number and total capacity of currently active nodes

func (ClientPool) BalanceOperation

func (bt ClientPool) BalanceOperation(id enode.ID, connAddress string, cb func(AtomicBalanceOperator))

BalanceOperation allows atomic operations on the balance of a node regardless of whether it is currently connected or not

func (ClientPool) GetExpirationTCs

func (bt ClientPool) GetExpirationTCs() (pos, neg uint64)

GetExpirationTCs returns the current positive and negative token expiration time constants

func (ClientPool) GetPosBalanceIDs

func (bt ClientPool) GetPosBalanceIDs(start, stop enode.ID, maxCount int) (result []enode.ID)

GetPosBalanceIDs lists node IDs with an associated positive balance

func (*ClientPool) Handle

func (cp *ClientPool) Handle(id enode.ID, address string, name string, data []byte) []byte

Handle implements Service

func (ClientPool) Limits

func (pp ClientPool) Limits() (uint64, uint64)

Limits returns the maximum allowed number and total capacity of active nodes

func (*ClientPool) Register

func (cp *ClientPool) Register(peer clientPeer) ConnectedBalance

Register registers the peer into the client pool. If the peer has insufficient priority and remains inactive for longer than the allowed timeout then it will be disconnected by calling the Disconnect function of the clientPeer interface.

func (*ClientPool) SetCapacity

func (cp *ClientPool) SetCapacity(node *enode.Node, reqCap uint64, bias time.Duration, requested bool) (capacity uint64, err error)

SetCapacity sets the assigned capacity of a connected client

func (*ClientPool) SetConnectedBias

func (cp *ClientPool) SetConnectedBias(bias time.Duration)

setConnectedBias sets the connection bias, which is applied to already connected clients So that already connected client won't be kicked out very soon and we can ensure all connected clients can have enough time to request or sync some data.

func (ClientPool) SetDefaultFactors

func (bt ClientPool) SetDefaultFactors(posFactors, negFactors PriceFactors)

SetDefaultFactors sets the default price factors applied to subsequently connected clients

func (ClientPool) SetExpirationTCs

func (bt ClientPool) SetExpirationTCs(pos, neg uint64)

SetExpirationTCs sets positive and negative token expiration time constants. Specified in seconds, 0 means infinite (no expiration).

func (ClientPool) SetLimits

func (pp ClientPool) SetLimits(maxCount, maxCap uint64)

SetLimits sets the maximum number and total capacity of simultaneously active nodes

func (*ClientPool) Start

func (cp *ClientPool) Start()

Start starts the client pool. Should be called before Register/Unregister.

func (*ClientPool) Stop

func (cp *ClientPool) Stop()

Stop shuts the client pool down. The clientPeer interface callbacks will not be called after Stop. Register calls will return nil.

func (ClientPool) TotalTokenAmount

func (bt ClientPool) TotalTokenAmount() uint64

TotalTokenAmount returns the current total amount of service tokens in existence

func (*ClientPool) Unregister

func (cp *ClientPool) Unregister(peer clientPeer)

Unregister removes the peer from the client pool

type ConnectedBalance

type ConnectedBalance interface {
	ReadOnlyBalance
	SetPriceFactors(posFactor, negFactor PriceFactors)
	RequestServed(cost uint64) uint64
}

ConnectedBalance provides operations permitted on connected nodes (non-read-only operations are not permitted inside a BalanceOperation)

type PriceFactors

type PriceFactors struct {
	TimeFactor, CapacityFactor, RequestFactor float64
}

PriceFactors determine the pricing policy (may apply either to positive or negative balances which may have different factors). - TimeFactor is cost unit per nanosecond of connection time - CapacityFactor is cost unit per nanosecond of connection time per 1000000 capacity - RequestFactor is cost unit per request "realCost" unit

type ReadOnlyBalance

type ReadOnlyBalance interface {
	GetBalance() (uint64, uint64)
	GetRawBalance() (utils.ExpiredValue, utils.ExpiredValue)
	GetPriceFactors() (posFactor, negFactor PriceFactors)
	// contains filtered or unexported methods
}

ReadOnlyBalance provides read-only operations on the node balance

type Server

type Server struct {
	// contains filtered or unexported fields
}

Server serves vflux requests

func NewServer

func NewServer(delayPerRequest time.Duration) *Server

NewServer creates a new Server

func (*Server) Register

func (s *Server) Register(b Service, id, desc string)

Register registers a Service

func (*Server) Serve

func (s *Server) Serve(id enode.ID, address string, requests vflux.Requests) vflux.Replies

Serve serves a vflux request batch Note: requests are served by the Handle functions of the registered services. Serve may be called concurrently but the Handle functions are called sequentially and therefore thread safety is guaranteed.

func (*Server) ServeEncoded

func (s *Server) ServeEncoded(id enode.ID, addr *net.UDPAddr, req []byte) []byte

ServeEncoded serves an encoded vflux request batch and returns the encoded replies

func (*Server) Stop

func (s *Server) Stop()

Stop shuts down the server

type Service

type Service interface {
	Handle(id enode.ID, address string, name string, data []byte) []byte // never called concurrently
}

Service is a service registered at the Server and identified by a string id