// Copyright 2016 The go-ethereum Authors // This file is part of the go-ethereum library. // // The go-ethereum library is free software: you can redistribute it and/or modify // it under the terms of the GNU Lesser General Public License as published by // the Free Software Foundation, either version 3 of the License, or // (at your option) any later version. // // The go-ethereum library is distributed in the hope that it will be useful, // but WITHOUT ANY WARRANTY; without even the implied warranty of // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the // GNU Lesser General Public License for more details. // // 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 ( "fmt" "sync" "time" "github.com/ethereum/go-ethereum/common/mclock" "github.com/ethereum/go-ethereum/log" ) const ( // fcTimeConst is the time constant applied for MinRecharge during linear // buffer recharge period fcTimeConst = time.Millisecond // DecParamDelay is applied at server side when decreasing capacity in order to // avoid a buffer underrun error due to requests sent by the client before // receiving the capacity update announcement DecParamDelay = time.Second * 2 // keepLogs is the duration of keeping logs; logging is not used if zero keepLogs = 0 ) // ServerParams are the flow control parameters specified by a server for a client // // Note: a server can assign different amounts of capacity to each client by giving // different parameters to them. type ServerParams struct { BufLimit, MinRecharge uint64 } // scheduledUpdate represents a delayed flow control parameter update type scheduledUpdate struct { time mclock.AbsTime params ServerParams } // ClientNode is the flow control system's representation of a client // (used in server mode only) type ClientNode struct { params ServerParams bufValue int64 lastTime mclock.AbsTime updateSchedule []scheduledUpdate sumCost uint64 // sum of req costs received from this client accepted map[uint64]uint64 // value = sumCost after accepting the given req connected bool lock sync.Mutex cm *ClientManager log *logger cmNodeFields } // NewClientNode returns a new ClientNode func NewClientNode(cm *ClientManager, params ServerParams) *ClientNode { node := &ClientNode{ cm: cm, params: params, bufValue: int64(params.BufLimit), lastTime: cm.clock.Now(), accepted: make(map[uint64]uint64), connected: true, } if keepLogs > 0 { node.log = newLogger(keepLogs) } cm.connect(node) return node } // Disconnect should be called when a client is disconnected func (node *ClientNode) Disconnect() { node.lock.Lock() defer node.lock.Unlock() node.connected = false node.cm.disconnect(node) } // BufferStatus returns the current buffer value and limit func (node *ClientNode) BufferStatus() (uint64, uint64) { node.lock.Lock() defer node.lock.Unlock() if !node.connected { return 0, 0 } now := node.cm.clock.Now() node.update(now) node.cm.updateBuffer(node, 0, now) bv := node.bufValue if bv < 0 { bv = 0 } return uint64(bv), node.params.BufLimit } // OneTimeCost subtracts the given amount from the node's buffer. // // Note: this call can take the buffer into the negative region internally. // In this case zero buffer value is returned by exported calls and no requests // are accepted. func (node *ClientNode) OneTimeCost(cost uint64) { node.lock.Lock() defer node.lock.Unlock() now := node.cm.clock.Now() node.update(now) node.bufValue -= int64(cost) node.cm.updateBuffer(node, -int64(cost), now) } // Freeze notifies the client manager about a client freeze event in which case // the total capacity allowance is slightly reduced. func (node *ClientNode) Freeze() { node.lock.Lock() frozenCap := node.params.MinRecharge node.lock.Unlock() node.cm.reduceTotalCapacity(frozenCap) } // 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) node.updateParams(node.updateSchedule[0].params, now) node.updateSchedule = node.updateSchedule[1:] } 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 { dt = 0 } node.bufValue += int64(node.params.MinRecharge * dt / uint64(fcTimeConst)) if node.bufValue > int64(node.params.BufLimit) { node.bufValue = int64(node.params.BufLimit) } if node.log != nil { node.log.add(now, fmt.Sprintf("updated bv=%d MRR=%d BufLimit=%d", node.bufValue, node.params.MinRecharge, node.params.BufLimit)) } 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() now := node.cm.clock.Now() node.update(now) if params.MinRecharge >= node.params.MinRecharge { node.updateSchedule = nil node.updateParams(params, now) } else { for i, s := range node.updateSchedule { if params.MinRecharge >= s.params.MinRecharge { s.params = params node.updateSchedule = node.updateSchedule[:i+1] return } } node.updateSchedule = append(node.updateSchedule, scheduledUpdate{time: now + mclock.AbsTime(DecParamDelay), params: params}) } } // updateParams updates the flow control parameters of the node func (node *ClientNode) updateParams(params ServerParams, now mclock.AbsTime) { diff := int64(params.BufLimit - node.params.BufLimit) if diff > 0 { node.bufValue += diff } else if node.bufValue > int64(params.BufLimit) { node.bufValue = int64(params.BufLimit) } node.cm.updateParams(node, params, now) } // AcceptRequest returns whether a new request can be accepted and the missing // buffer amount if it was rejected due to a buffer underrun. If accepted, maxCost // is deducted from the flow control buffer. func (node *ClientNode) AcceptRequest(reqID, index, maxCost uint64) (accepted bool, bufShort uint64, priority int64) { node.lock.Lock() defer node.lock.Unlock() now := node.cm.clock.Now() node.update(now) if int64(maxCost) > node.bufValue { if node.log != nil { node.log.add(now, fmt.Sprintf("rejected reqID=%d bv=%d maxCost=%d", reqID, node.bufValue, maxCost)) node.log.dump(now) } return false, maxCost - uint64(node.bufValue), 0 } node.bufValue -= int64(maxCost) node.sumCost += maxCost if node.log != nil { node.log.add(now, fmt.Sprintf("accepted reqID=%d bv=%d maxCost=%d sumCost=%d", reqID, node.bufValue, maxCost, node.sumCost)) } node.accepted[index] = node.sumCost return true, 0, node.cm.accepted(node, maxCost, now) } // RequestProcessed should be called when the request has been processed func (node *ClientNode) RequestProcessed(reqID, index, maxCost, realCost uint64) uint64 { node.lock.Lock() defer node.lock.Unlock() now := node.cm.clock.Now() node.update(now) node.cm.processed(node, maxCost, realCost, now) bv := node.bufValue + int64(node.sumCost-node.accepted[index]) if node.log != nil { node.log.add(now, fmt.Sprintf("processed reqID=%d bv=%d maxCost=%d realCost=%d sumCost=%d oldSumCost=%d reportedBV=%d", reqID, node.bufValue, maxCost, realCost, node.sumCost, node.accepted[index], bv)) } delete(node.accepted, index) if bv < 0 { return 0 } return uint64(bv) } // ServerNode is the flow control system's representation of a server // (used in client mode only) type ServerNode struct { clock mclock.Clock bufEstimate uint64 bufRecharge bool lastTime mclock.AbsTime params ServerParams sumCost uint64 // sum of req costs sent to this server pending map[uint64]uint64 // value = sumCost after sending the given req log *logger lock sync.RWMutex } // NewServerNode returns a new ServerNode func NewServerNode(params ServerParams, clock mclock.Clock) *ServerNode { node := &ServerNode{ clock: clock, bufEstimate: params.BufLimit, bufRecharge: false, lastTime: clock.Now(), params: params, pending: make(map[uint64]uint64), } if keepLogs > 0 { node.log = newLogger(keepLogs) } return node } // UpdateParams updates the flow control parameters of the node func (node *ServerNode) UpdateParams(params ServerParams) { node.lock.Lock() defer node.lock.Unlock() node.recalcBLE(mclock.Now()) if params.BufLimit > node.params.BufLimit { node.bufEstimate += params.BufLimit - node.params.BufLimit } else { if node.bufEstimate > params.BufLimit { node.bufEstimate = params.BufLimit } } 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 } if node.bufRecharge { dt := uint64(now - node.lastTime) node.bufEstimate += node.params.MinRecharge * dt / uint64(fcTimeConst) if node.bufEstimate >= node.params.BufLimit { node.bufEstimate = node.params.BufLimit node.bufRecharge = false } } node.lastTime = now if node.log != nil { node.log.add(now, fmt.Sprintf("updated bufEst=%d MRR=%d BufLimit=%d", node.bufEstimate, node.params.MinRecharge, node.params.BufLimit)) } } // safetyMargin is added to the flow control waiting time when estimated buffer value is low const safetyMargin = time.Millisecond // CanSend returns the minimum waiting time required before sending a request // with the given maximum estimated cost. Second return value is the relative // estimated buffer level after sending the request (divided by BufLimit). func (node *ServerNode) CanSend(maxCost uint64) (time.Duration, float64) { node.lock.RLock() defer node.lock.RUnlock() now := node.clock.Now() node.recalcBLE(now) maxCost += uint64(safetyMargin) * node.params.MinRecharge / uint64(fcTimeConst) if maxCost > node.params.BufLimit { maxCost = node.params.BufLimit } if node.bufEstimate >= maxCost { relBuf := float64(node.bufEstimate-maxCost) / float64(node.params.BufLimit) if node.log != nil { node.log.add(now, fmt.Sprintf("canSend bufEst=%d maxCost=%d true relBuf=%f", node.bufEstimate, maxCost, relBuf)) } return 0, relBuf } timeLeft := time.Duration((maxCost - node.bufEstimate) * uint64(fcTimeConst) / node.params.MinRecharge) if node.log != nil { node.log.add(now, fmt.Sprintf("canSend bufEst=%d maxCost=%d false timeLeft=%v", node.bufEstimate, maxCost, timeLeft)) } return timeLeft, 0 } // QueuedRequest should be called when the request has been assigned to the given // server node, before putting it in the send queue. It is mandatory that requests // are sent in the same order as the QueuedRequest calls are made. func (node *ServerNode) QueuedRequest(reqID, maxCost uint64) { node.lock.Lock() defer node.lock.Unlock() now := node.clock.Now() node.recalcBLE(now) // Note: we do not know when requests actually arrive to the server so bufRecharge // is not turned on here if buffer was full; in this case it is going to be turned // on by the first reply's bufValue feedback if node.bufEstimate >= maxCost { node.bufEstimate -= maxCost } else { log.Error("Queued request with insufficient buffer estimate") node.bufEstimate = 0 } node.sumCost += maxCost node.pending[reqID] = node.sumCost if node.log != nil { node.log.add(now, fmt.Sprintf("queued reqID=%d bufEst=%d maxCost=%d sumCost=%d", reqID, node.bufEstimate, maxCost, node.sumCost)) } } // ReceivedReply adjusts estimated buffer value according to the value included in // the latest request reply. func (node *ServerNode) ReceivedReply(reqID, bv uint64) { node.lock.Lock() defer node.lock.Unlock() now := node.clock.Now() node.recalcBLE(now) if bv > node.params.BufLimit { bv = node.params.BufLimit } sc, ok := node.pending[reqID] if !ok { return } delete(node.pending, reqID) cc := node.sumCost - sc newEstimate := uint64(0) if bv > cc { newEstimate = bv - cc } if newEstimate > node.bufEstimate { // Note: we never reduce the buffer estimate based on the reported value because // this can only happen because of the delayed delivery of the latest reply. // The lowest estimate based on the previous reply can still be considered valid. node.bufEstimate = newEstimate } node.bufRecharge = node.bufEstimate < node.params.BufLimit node.lastTime = now if node.log != nil { node.log.add(now, fmt.Sprintf("received reqID=%d bufEst=%d reportedBv=%d sumCost=%d oldSumCost=%d", reqID, node.bufEstimate, bv, node.sumCost, sc)) } } // ResumeFreeze cleans all pending requests and sets the buffer estimate to the // reported value after resuming from a frozen state func (node *ServerNode) ResumeFreeze(bv uint64) { node.lock.Lock() defer node.lock.Unlock() for reqID := range node.pending { delete(node.pending, reqID) } now := node.clock.Now() node.recalcBLE(now) if bv > node.params.BufLimit { bv = node.params.BufLimit } node.bufEstimate = bv node.bufRecharge = node.bufEstimate < node.params.BufLimit node.lastTime = now if node.log != nil { node.log.add(now, fmt.Sprintf("unfreeze bv=%d sumCost=%d", bv, node.sumCost)) } } // DumpLogs dumps the event log if logging is used func (node *ServerNode) DumpLogs() { node.lock.Lock() defer node.lock.Unlock() if node.log != nil { node.log.dump(node.clock.Now()) } }