From c02170de4e523751e08b1557007acfbd1da3a55a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Wed, 1 Jun 2016 18:07:25 +0300 Subject: [PATCH 1/6] eth/downloader: adaptive quality of service tuning --- eth/downloader/downloader.go | 133 ++++++++++++++++++++++++------ eth/downloader/downloader_test.go | 2 +- eth/downloader/peer.go | 65 +++++++++++---- 3 files changed, 159 insertions(+), 41 deletions(-) diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index f6dbb4610f..2f371d8f06 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -54,14 +54,14 @@ var ( blockTargetRTT = 3 * time.Second / 2 // [eth/61] Target time for completing a block retrieval request blockTTL = 3 * blockTargetRTT // [eth/61] Maximum time allowance before a block request is considered expired - headerTargetRTT = time.Second // [eth/62] Target time for completing a header retrieval request (only for measurements for now) - headerTTL = 3 * time.Second // [eth/62] Time it takes for a header request to time out - bodyTargetRTT = 3 * time.Second / 2 // [eth/62] Target time for completing a block body retrieval request - bodyTTL = 3 * bodyTargetRTT // [eth/62] Maximum time allowance before a block body request is considered expired - receiptTargetRTT = 3 * time.Second / 2 // [eth/63] Target time for completing a receipt retrieval request - receiptTTL = 3 * receiptTargetRTT // [eth/63] Maximum time allowance before a receipt request is considered expired - stateTargetRTT = 2 * time.Second / 2 // [eth/63] Target time for completing a state trie retrieval request - stateTTL = 3 * stateTargetRTT // [eth/63] Maximum time allowance before a node data request is considered expired + rttMinEstimate = 2 * time.Second // Minimum round-trip time to target for download requests + rttMaxEstimate = 20 * time.Second // Maximum rount-trip time to target for download requests + rttMinConfidence = 0.1 // Worse confidence factor in our estimated RTT value + rttTTLScaling = 3 // Constant scaling factor for RTT -> TTL conversion + + qosTuningPeers = 5 // Number of peers to tune based on (best peers) + qosConfidenceCap = 10 // Number of peers above which not to modify RTT confidence + qosTuningImpact = 0.25 // Impact that a new tuning target has on the previous value maxQueuedHashes = 32 * 1024 // [eth/61] Maximum number of hashes to queue for import (DOS protection) maxQueuedHeaders = 32 * 1024 // [eth/62] Maximum number of headers to queue for import (DOS protection) @@ -110,7 +110,9 @@ type Downloader struct { queue *queue // Scheduler for selecting the hashes to download peers *peerSet // Set of active peers from which download can proceed - interrupt int32 // Atomic boolean to signal termination + rttEstimate uint64 // Round trip time to target for download requests + rttConfidence uint64 // Confidence in the estimated RTT (unit: millionths to allow atomic ops) + interrupt int32 // Atomic boolean to signal termination // Statistics syncStatsChainOrigin uint64 // Origin block number where syncing started at @@ -169,11 +171,13 @@ func New(stateDb ethdb.Database, mux *event.TypeMux, hasHeader headerCheckFn, ha headFastBlock headFastBlockRetrievalFn, commitHeadBlock headBlockCommitterFn, getTd tdRetrievalFn, insertHeaders headerChainInsertFn, insertBlocks blockChainInsertFn, insertReceipts receiptChainInsertFn, rollback chainRollbackFn, dropPeer peerDropFn) *Downloader { - return &Downloader{ + dl := &Downloader{ mode: FullSync, mux: mux, queue: newQueue(stateDb), peers: newPeerSet(), + rttEstimate: uint64(rttMaxEstimate), + rttConfidence: uint64(1000000), hasHeader: hasHeader, hasBlockAndState: hasBlockAndState, getHeader: getHeader, @@ -201,6 +205,8 @@ func New(stateDb ethdb.Database, mux *event.TypeMux, hasHeader headerCheckFn, ha stateWakeCh: make(chan bool, 1), headerProcCh: make(chan []*types.Header, 1), } + go dl.qosTuner() + return dl } // Progress retrieves the synchronisation boundaries, specifically the origin @@ -247,6 +253,8 @@ func (d *Downloader) RegisterPeer(id string, version int, head common.Hash, glog.V(logger.Error).Infoln("Register failed:", err) return err } + d.qosReduceConfidence() + return nil } @@ -915,7 +923,7 @@ func (d *Downloader) fetchBlocks61(from uint64) error { // Reserve a chunk of hashes for a peer. A nil can mean either that // no more hashes are available, or that the peer is known not to // have them. - request := d.queue.ReserveBlocks(peer, peer.BlockCapacity()) + request := d.queue.ReserveBlocks(peer, peer.BlockCapacity(blockTargetRTT)) if request == nil { continue } @@ -956,7 +964,7 @@ func (d *Downloader) fetchHeight(p *peer) (*types.Header, error) { // Request the advertised remote head block and wait for the response go p.getRelHeaders(p.head, 1, 0, false) - timeout := time.After(headerTTL) + timeout := time.After(d.requestTTL()) for { select { case <-d.cancelCh: @@ -1024,7 +1032,7 @@ func (d *Downloader) findAncestor(p *peer, height uint64) (uint64, error) { // Wait for the remote response to the head fetch number, hash := uint64(0), common.Hash{} - timeout := time.After(hashTTL) + timeout := time.After(d.requestTTL()) for finished := false; !finished; { select { @@ -1101,7 +1109,7 @@ func (d *Downloader) findAncestor(p *peer, height uint64) (uint64, error) { // Split our chain interval in two, and request the hash to cross check check := (start + end) / 2 - timeout := time.After(hashTTL) + timeout := time.After(d.requestTTL()) go p.getAbsHeaders(uint64(check), 1, 0, false) // Wait until a reply arrives to this request @@ -1182,7 +1190,7 @@ func (d *Downloader) fetchHeaders(p *peer, from uint64) error { getHeaders := func(from uint64) { request = time.Now() - timeout.Reset(headerTTL) + timeout.Reset(d.requestTTL()) if skeleton { glog.V(logger.Detail).Infof("%v: fetching %d skeleton headers from #%d", p, MaxHeaderFetch, from) @@ -1290,13 +1298,13 @@ func (d *Downloader) fillHeaderSkeleton(from uint64, skeleton []*types.Header) ( pack := packet.(*headerPack) return d.queue.DeliverHeaders(pack.peerId, pack.headers, d.headerProcCh) } - expire = func() map[string]int { return d.queue.ExpireHeaders(headerTTL) } + expire = func() map[string]int { return d.queue.ExpireHeaders(d.requestTTL()) } throttle = func() bool { return false } reserve = func(p *peer, count int) (*fetchRequest, bool, error) { return d.queue.ReserveHeaders(p, count), false, nil } fetch = func(p *peer, req *fetchRequest) error { return p.FetchHeaders(req.From, MaxHeaderFetch) } - capacity = func(p *peer) int { return p.HeaderCapacity() } + capacity = func(p *peer) int { return p.HeaderCapacity(d.requestRTT()) } setIdle = func(p *peer, accepted int) { p.SetHeadersIdle(accepted) } ) err := d.fetchParts(errCancelHeaderFetch, d.headerCh, deliver, d.queue.headerContCh, expire, @@ -1320,9 +1328,9 @@ func (d *Downloader) fetchBodies(from uint64) error { pack := packet.(*bodyPack) return d.queue.DeliverBodies(pack.peerId, pack.transactions, pack.uncles) } - expire = func() map[string]int { return d.queue.ExpireBodies(bodyTTL) } + expire = func() map[string]int { return d.queue.ExpireBodies(d.requestTTL()) } fetch = func(p *peer, req *fetchRequest) error { return p.FetchBodies(req) } - capacity = func(p *peer) int { return p.BlockCapacity() } + capacity = func(p *peer) int { return p.BlockCapacity(d.requestRTT()) } setIdle = func(p *peer, accepted int) { p.SetBodiesIdle(accepted) } ) err := d.fetchParts(errCancelBodyFetch, d.bodyCh, deliver, d.bodyWakeCh, expire, @@ -1344,9 +1352,9 @@ func (d *Downloader) fetchReceipts(from uint64) error { pack := packet.(*receiptPack) return d.queue.DeliverReceipts(pack.peerId, pack.receipts) } - expire = func() map[string]int { return d.queue.ExpireReceipts(receiptTTL) } + expire = func() map[string]int { return d.queue.ExpireReceipts(d.requestTTL()) } fetch = func(p *peer, req *fetchRequest) error { return p.FetchReceipts(req) } - capacity = func(p *peer) int { return p.ReceiptCapacity() } + capacity = func(p *peer) int { return p.ReceiptCapacity(d.requestRTT()) } setIdle = func(p *peer, accepted int) { p.SetReceiptsIdle(accepted) } ) err := d.fetchParts(errCancelReceiptFetch, d.receiptCh, deliver, d.receiptWakeCh, expire, @@ -1396,13 +1404,13 @@ func (d *Downloader) fetchNodeData() error { } }) } - expire = func() map[string]int { return d.queue.ExpireNodeData(stateTTL) } + expire = func() map[string]int { return d.queue.ExpireNodeData(d.requestTTL()) } throttle = func() bool { return false } reserve = func(p *peer, count int) (*fetchRequest, bool, error) { return d.queue.ReserveNodeData(p, count), false, nil } fetch = func(p *peer, req *fetchRequest) error { return p.FetchNodeData(req) } - capacity = func(p *peer) int { return p.NodeDataCapacity() } + capacity = func(p *peer) int { return p.NodeDataCapacity(d.requestRTT()) } setIdle = func(p *peer, accepted int) { p.SetNodeDataIdle(accepted) } ) err := d.fetchParts(errCancelStateFetch, d.stateCh, deliver, d.stateWakeCh, expire, @@ -1864,3 +1872,82 @@ func (d *Downloader) deliver(id string, destCh chan dataPack, packet dataPack, i return errNoSyncActive } } + +// qosTuner is the quality of service tuning loop that occasionally gathers the +// peer latency statistics and updates the estimated request round trip time. +func (d *Downloader) qosTuner() { + for atomic.LoadInt32(&d.interrupt) == 0 { + // Gather all the active RTTs and set the target RTT to the median + rtts := d.peers.RTTs() + + var rtt time.Duration + if qosTuningPeers <= len(rtts) { + rtt = rtts[qosTuningPeers/2] // Median of our tuning peers + } else if len(rtts) > 0 { + rtt = rtts[len(rtts)/2] // Median of our connected peers (maintain even like this some baseline qos) + } else { + rtt = rttMaxEstimate // No peers, reset to some sane default + } + if rtt < rttMinEstimate { + rtt = rttMinEstimate + } + if rtt > rttMaxEstimate { + rtt = rttMaxEstimate + } + rtt = time.Duration(float64(1-qosTuningImpact)*float64(atomic.LoadUint64(&d.rttEstimate)) + qosTuningImpact*float64(rtt)) + atomic.StoreUint64(&d.rttEstimate, uint64(rtt)) + + // A new RTT cycle passed, increase our confidence in the estimated RTT + conf := atomic.LoadUint64(&d.rttConfidence) + conf = conf + (1000000-conf)/2 + atomic.StoreUint64(&d.rttConfidence, conf) + + // Log the new QoS values and sleep until the next RTT + glog.V(logger.Debug).Infof("quality of service: rtt %v, conf %.3f, ttl %v", rtt, float64(conf)/1000000.0, d.requestTTL()) + time.Sleep(rtt) + } +} + +// qosReduceConfidence is meant to be called when a new peer joins the downloader's +// peer set, needing to reduce the confidence we have in out QoS estimates. +func (d *Downloader) qosReduceConfidence() { + // If we have a single peer, confidence is always 1 + peers := uint64(d.peers.Len()) + if peers == 1 { + atomic.StoreUint64(&d.rttConfidence, 1000000) + return + } + // If we have a ton of peers, don't drop confidence) + if peers >= uint64(qosConfidenceCap) { + return + } + // Otherwise drop the confidence factor + conf := atomic.LoadUint64(&d.rttConfidence) * (peers - 1) / peers + if float64(conf)/1000000 < rttMinConfidence { + conf = uint64(rttMinConfidence * 1000000) + } + atomic.StoreUint64(&d.rttConfidence, conf) + + rtt := time.Duration(atomic.LoadUint64(&d.rttEstimate)) + glog.V(logger.Debug).Infof("quality of service: rtt %v, conf %.3f, ttl %v", rtt, float64(conf)/1000000.0, d.requestTTL()) +} + +// requestRTT returns the current target round trip time for a download request +// to complete in. +// +// Note, the returned RTT is .9 of the actually estimated RTT. The reason is that +// the downloader tries to adapt queries to the RTT, so multiple RTT values can +// be adapted to, but smaller ones are preffered (stabler download stream). +func (d *Downloader) requestRTT() time.Duration { + return time.Duration(atomic.LoadUint64(&d.rttEstimate)) * 9 / 10 +} + +// requestTTL returns the current timeout allowance for a single download request +// to finish under. +func (d *Downloader) requestTTL() time.Duration { + var ( + rtt = time.Duration(atomic.LoadUint64(&d.rttEstimate)) + conf = float64(atomic.LoadUint64(&d.rttConfidence)) / 1000000.0 + ) + return time.Duration(rttTTLScaling) * time.Duration(float64(rtt)/conf) +} diff --git a/eth/downloader/downloader_test.go b/eth/downloader/downloader_test.go index 1cf0e7cd31..9eecf87628 100644 --- a/eth/downloader/downloader_test.go +++ b/eth/downloader/downloader_test.go @@ -1734,7 +1734,7 @@ func testDeliverHeadersHang(t *testing.T, protocol int, mode SyncMode) { impl := tester.peerGetAbsHeadersFn("peer", 0) go impl(from, count, skip, reverse) // None of the extra deliveries should block. - timeout := time.After(5 * time.Second) + timeout := time.After(15 * time.Second) for i := 0; i < cap(deliveriesDone); i++ { select { case <-deliveriesDone: diff --git a/eth/downloader/peer.go b/eth/downloader/peer.go index 6aab907d7e..32bce1628f 100644 --- a/eth/downloader/peer.go +++ b/eth/downloader/peer.go @@ -31,8 +31,8 @@ import ( ) const ( - maxLackingHashes = 4096 // Maximum number of entries allowed on the list or lacking items - throughputImpact = 0.1 // The impact a single measurement has on a peer's final throughput value. + maxLackingHashes = 4096 // Maximum number of entries allowed on the list or lacking items + measurementImpact = 0.1 // The impact a single measurement has on a peer's final throughput value. ) // Hash and block fetchers belonging to eth/61 and below @@ -68,6 +68,8 @@ type peer struct { receiptThroughput float64 // Number of receipts measured to be retrievable per second stateThroughput float64 // Number of node data pieces measured to be retrievable per second + rtt time.Duration // Request round trip time to track responsiveness (QoS) + headerStarted time.Time // Time instance when the last header fetch was started blockStarted time.Time // Time instance when the last block (body) fetch was started receiptStarted time.Time // Time instance when the last receipt fetch was started @@ -290,44 +292,47 @@ func (p *peer) setIdle(started time.Time, delivered int, throughput *float64, id return } // Otherwise update the throughput with a new measurement - measured := float64(delivered) / (float64(time.Since(started)+1) / float64(time.Second)) // +1 (ns) to ensure non-zero divisor - *throughput = (1-throughputImpact)*(*throughput) + throughputImpact*measured + elapsed := time.Since(started) + 1 // +1 (ns) to ensure non-zero divisor + measured := float64(delivered) / (float64(elapsed) / float64(time.Second)) + + *throughput = (1-measurementImpact)*(*throughput) + measurementImpact*measured + p.rtt = time.Duration((1-measurementImpact)*float64(p.rtt) + measurementImpact*float64(elapsed)) } // HeaderCapacity retrieves the peers header download allowance based on its // previously discovered throughput. -func (p *peer) HeaderCapacity() int { +func (p *peer) HeaderCapacity(targetRTT time.Duration) int { p.lock.RLock() defer p.lock.RUnlock() - return int(math.Max(1, math.Min(p.headerThroughput*float64(headerTargetRTT)/float64(time.Second), float64(MaxHeaderFetch)))) + return int(math.Min(1+math.Max(1, p.headerThroughput*float64(targetRTT)/float64(time.Second)), float64(MaxHeaderFetch))) } // BlockCapacity retrieves the peers block download allowance based on its // previously discovered throughput. -func (p *peer) BlockCapacity() int { +func (p *peer) BlockCapacity(targetRTT time.Duration) int { p.lock.RLock() defer p.lock.RUnlock() - return int(math.Max(1, math.Min(p.blockThroughput*float64(blockTargetRTT)/float64(time.Second), float64(MaxBlockFetch)))) + return int(math.Min(1+math.Max(1, p.blockThroughput*float64(targetRTT)/float64(time.Second)), float64(MaxBlockFetch))) } // ReceiptCapacity retrieves the peers receipt download allowance based on its // previously discovered throughput. -func (p *peer) ReceiptCapacity() int { +func (p *peer) ReceiptCapacity(targetRTT time.Duration) int { p.lock.RLock() defer p.lock.RUnlock() - return int(math.Max(1, math.Min(p.receiptThroughput*float64(receiptTargetRTT)/float64(time.Second), float64(MaxReceiptFetch)))) + return int(math.Min(1+math.Max(1, p.receiptThroughput*float64(targetRTT)/float64(time.Second)), float64(MaxReceiptFetch))) } // NodeDataCapacity retrieves the peers state download allowance based on its // previously discovered throughput. -func (p *peer) NodeDataCapacity() int { +func (p *peer) NodeDataCapacity(targetRTT time.Duration) int { p.lock.RLock() defer p.lock.RUnlock() - return int(math.Max(1, math.Min(p.stateThroughput*float64(stateTargetRTT)/float64(time.Second), float64(MaxStateFetch)))) + return int(math.Min(1+math.Max(1, p.stateThroughput*float64(targetRTT)/float64(time.Second)), float64(MaxStateFetch))) } // MarkLacking appends a new entity to the set of items (blocks, receipts, states) @@ -362,11 +367,12 @@ func (p *peer) String() string { defer p.lock.RUnlock() return fmt.Sprintf("Peer %s [%s]", p.id, - fmt.Sprintf("headers %3.2f/s, ", p.headerThroughput)+ - fmt.Sprintf("blocks %3.2f/s, ", p.blockThroughput)+ - fmt.Sprintf("receipts %3.2f/s, ", p.receiptThroughput)+ - fmt.Sprintf("states %3.2f/s, ", p.stateThroughput)+ - fmt.Sprintf("lacking %4d", len(p.lacking)), + fmt.Sprintf("hs %3.2f/s, ", p.headerThroughput)+ + fmt.Sprintf("bs %3.2f/s, ", p.blockThroughput)+ + fmt.Sprintf("rs %3.2f/s, ", p.receiptThroughput)+ + fmt.Sprintf("ss %3.2f/s, ", p.stateThroughput)+ + fmt.Sprintf("miss %4d, ", len(p.lacking))+ + fmt.Sprintf("rtt %v", p.rtt), ) } @@ -410,6 +416,7 @@ func (ps *peerSet) Register(p *peer) error { } if len(ps.peers) > 0 { p.headerThroughput, p.blockThroughput, p.receiptThroughput, p.stateThroughput = 0, 0, 0, 0 + p.rtt = rttMaxEstimate for _, peer := range ps.peers { peer.lock.RLock() @@ -564,3 +571,27 @@ func (ps *peerSet) idlePeers(minProtocol, maxProtocol int, idleCheck func(*peer) } return idle, total } + +// RTTs retrieves all the round trip times measured for the current set of peers, +// sorted ascending. +func (ps *peerSet) RTTs() []time.Duration { + // Gather all the currnetly measured round trip times + ps.lock.RLock() + defer ps.lock.RUnlock() + + rtts := make([]time.Duration, 0, len(ps.peers)) + for _, p := range ps.peers { + p.lock.RLock() + rtts = append(rtts, p.rtt) + p.lock.RUnlock() + } + // Sort into ascending order and return them + for i := 0; i < len(rtts); i++ { + for j := i + 1; j < len(rtts); j++ { + if rtts[i] > rtts[j] { + rtts[i], rtts[j] = rtts[j], rtts[i] + } + } + } + return rtts +} From 2925dc3f07d6dca317c63a9ca9f511b0379cdb5a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Fri, 3 Jun 2016 11:05:25 +0300 Subject: [PATCH 2/6] eth/downloader: fix string formatting per review requests --- eth/downloader/downloader.go | 4 ++-- eth/downloader/peer.go | 17 +++++++++-------- 2 files changed, 11 insertions(+), 10 deletions(-) diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index 2f371d8f06..08b685ff8a 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -1903,7 +1903,7 @@ func (d *Downloader) qosTuner() { atomic.StoreUint64(&d.rttConfidence, conf) // Log the new QoS values and sleep until the next RTT - glog.V(logger.Debug).Infof("quality of service: rtt %v, conf %.3f, ttl %v", rtt, float64(conf)/1000000.0, d.requestTTL()) + glog.V(logger.Debug).Infof("Quality of service: rtt %v, conf %.3f, ttl %v", rtt, float64(conf)/1000000.0, d.requestTTL()) time.Sleep(rtt) } } @@ -1929,7 +1929,7 @@ func (d *Downloader) qosReduceConfidence() { atomic.StoreUint64(&d.rttConfidence, conf) rtt := time.Duration(atomic.LoadUint64(&d.rttEstimate)) - glog.V(logger.Debug).Infof("quality of service: rtt %v, conf %.3f, ttl %v", rtt, float64(conf)/1000000.0, d.requestTTL()) + glog.V(logger.Debug).Infof("Quality of service: rtt %v, conf %.3f, ttl %v", rtt, float64(conf)/1000000.0, d.requestTTL()) } // requestRTT returns the current target round trip time for a download request diff --git a/eth/downloader/peer.go b/eth/downloader/peer.go index 32bce1628f..a538cad886 100644 --- a/eth/downloader/peer.go +++ b/eth/downloader/peer.go @@ -23,6 +23,7 @@ import ( "errors" "fmt" "math" + "strings" "sync" "sync/atomic" "time" @@ -366,14 +367,14 @@ func (p *peer) String() string { p.lock.RLock() defer p.lock.RUnlock() - return fmt.Sprintf("Peer %s [%s]", p.id, - fmt.Sprintf("hs %3.2f/s, ", p.headerThroughput)+ - fmt.Sprintf("bs %3.2f/s, ", p.blockThroughput)+ - fmt.Sprintf("rs %3.2f/s, ", p.receiptThroughput)+ - fmt.Sprintf("ss %3.2f/s, ", p.stateThroughput)+ - fmt.Sprintf("miss %4d, ", len(p.lacking))+ - fmt.Sprintf("rtt %v", p.rtt), - ) + return fmt.Sprintf("Peer %s [%s]", p.id, strings.Join([]string{ + fmt.Sprintf("hs %3.2f/s", p.headerThroughput), + fmt.Sprintf("bs %3.2f/s", p.blockThroughput), + fmt.Sprintf("rs %3.2f/s", p.receiptThroughput), + fmt.Sprintf("ss %3.2f/s", p.stateThroughput), + fmt.Sprintf("miss %4d", len(p.lacking)), + fmt.Sprintf("rtt %v", p.rtt), + }, ", ")) } // peerSet represents the collection of active peer participating in the chain From ad8378a8d1c229433aeefcfd5da3d49c88e9fb36 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Fri, 3 Jun 2016 11:25:35 +0300 Subject: [PATCH 3/6] eth/downloader: cap the TTL to a reasonable 1 minute --- eth/downloader/downloader.go | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index 08b685ff8a..542934f963 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -57,7 +57,8 @@ var ( rttMinEstimate = 2 * time.Second // Minimum round-trip time to target for download requests rttMaxEstimate = 20 * time.Second // Maximum rount-trip time to target for download requests rttMinConfidence = 0.1 // Worse confidence factor in our estimated RTT value - rttTTLScaling = 3 // Constant scaling factor for RTT -> TTL conversion + ttlScaling = 3 // Constant scaling factor for RTT -> TTL conversion + ttlLimit = time.Minute // Maximum TTL allowance to prevent reaching crazy timeouts qosTuningPeers = 5 // Number of peers to tune based on (best peers) qosConfidenceCap = 10 // Number of peers above which not to modify RTT confidence @@ -1949,5 +1950,9 @@ func (d *Downloader) requestTTL() time.Duration { rtt = time.Duration(atomic.LoadUint64(&d.rttEstimate)) conf = float64(atomic.LoadUint64(&d.rttConfidence)) / 1000000.0 ) - return time.Duration(rttTTLScaling) * time.Duration(float64(rtt)/conf) + ttl := time.Duration(ttlScaling) * time.Duration(float64(rtt)/conf) + if ttl > ttlLimit { + ttl = ttlLimit + } + return ttl } From 5255d200546c87c990e0767de66f05c2fb541798 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Fri, 3 Jun 2016 14:06:39 +0300 Subject: [PATCH 4/6] eth/downloader: use median RTT as the default for new peers --- eth/downloader/downloader.go | 20 ++------------------ eth/downloader/peer.go | 28 ++++++++++++++++++++++------ 2 files changed, 24 insertions(+), 24 deletions(-) diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index 542934f963..37cb04e05d 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -1878,24 +1878,8 @@ func (d *Downloader) deliver(id string, destCh chan dataPack, packet dataPack, i // peer latency statistics and updates the estimated request round trip time. func (d *Downloader) qosTuner() { for atomic.LoadInt32(&d.interrupt) == 0 { - // Gather all the active RTTs and set the target RTT to the median - rtts := d.peers.RTTs() - - var rtt time.Duration - if qosTuningPeers <= len(rtts) { - rtt = rtts[qosTuningPeers/2] // Median of our tuning peers - } else if len(rtts) > 0 { - rtt = rtts[len(rtts)/2] // Median of our connected peers (maintain even like this some baseline qos) - } else { - rtt = rttMaxEstimate // No peers, reset to some sane default - } - if rtt < rttMinEstimate { - rtt = rttMinEstimate - } - if rtt > rttMaxEstimate { - rtt = rttMaxEstimate - } - rtt = time.Duration(float64(1-qosTuningImpact)*float64(atomic.LoadUint64(&d.rttEstimate)) + qosTuningImpact*float64(rtt)) + // Retrieve the current median RTT and integrate into the previoust target RTT + rtt := time.Duration(float64(1-qosTuningImpact)*float64(atomic.LoadUint64(&d.rttEstimate)) + qosTuningImpact*float64(d.peers.medianRTT())) atomic.StoreUint64(&d.rttEstimate, uint64(rtt)) // A new RTT cycle passed, increase our confidence in the estimated RTT diff --git a/eth/downloader/peer.go b/eth/downloader/peer.go index a538cad886..57f7e008ad 100644 --- a/eth/downloader/peer.go +++ b/eth/downloader/peer.go @@ -409,6 +409,10 @@ func (ps *peerSet) Reset() { // average of all existing peers, to give it a realistic chance of being used // for data retrievals. func (ps *peerSet) Register(p *peer) error { + // Retrieve the current median RTT as a sane default + p.rtt = ps.medianRTT() + + // Register the new peer with some meaningful defaults ps.lock.Lock() defer ps.lock.Unlock() @@ -417,7 +421,6 @@ func (ps *peerSet) Register(p *peer) error { } if len(ps.peers) > 0 { p.headerThroughput, p.blockThroughput, p.receiptThroughput, p.stateThroughput = 0, 0, 0, 0 - p.rtt = rttMaxEstimate for _, peer := range ps.peers { peer.lock.RLock() @@ -573,9 +576,9 @@ func (ps *peerSet) idlePeers(minProtocol, maxProtocol int, idleCheck func(*peer) return idle, total } -// RTTs retrieves all the round trip times measured for the current set of peers, -// sorted ascending. -func (ps *peerSet) RTTs() []time.Duration { +// medianRTT returns the median RTT of te peerset, considering only the tuning +// peers if there are more peers available. +func (ps *peerSet) medianRTT() time.Duration { // Gather all the currnetly measured round trip times ps.lock.RLock() defer ps.lock.RUnlock() @@ -586,7 +589,7 @@ func (ps *peerSet) RTTs() []time.Duration { rtts = append(rtts, p.rtt) p.lock.RUnlock() } - // Sort into ascending order and return them + // Sort into ascending order and retrieve the median for i := 0; i < len(rtts); i++ { for j := i + 1; j < len(rtts); j++ { if rtts[i] > rtts[j] { @@ -594,5 +597,18 @@ func (ps *peerSet) RTTs() []time.Duration { } } } - return rtts + median := rttMaxEstimate + if qosTuningPeers <= len(rtts) { + median = rtts[qosTuningPeers/2] // Median of our tuning peers + } else if len(rtts) > 0 { + median = rtts[len(rtts)/2] // Median of our connected peers (maintain even like this some baseline qos) + } + // Restrict the RTT into some QoS defaults, irrelevant of true RTT + if median < rttMinEstimate { + median = rttMinEstimate + } + if median > rttMaxEstimate { + median = rttMaxEstimate + } + return median } From 708cca9ddeb058396919f999abd1eb022964fbc5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Thu, 2 Jun 2016 15:54:07 +0300 Subject: [PATCH 5/6] eth: don't accept transactions until we sync up with the network --- eth/handler.go | 14 ++++++++++---- eth/protocol_test.go | 1 + eth/sync.go | 2 ++ 3 files changed, 13 insertions(+), 4 deletions(-) diff --git a/eth/handler.go b/eth/handler.go index 58869a2eec..b309c0756e 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -59,7 +59,9 @@ type blockFetcherFn func([]common.Hash) error type ProtocolManager struct { networkId int - fastSync uint32 + fastSync uint32 // Flag whether fast sync is still enabled (no transactions, fast sync algo) + synced uint32 // Flag whether we're considered synchronised (enables transaction processing) + txpool txPool blockchain *core.BlockChain chaindb ethdb.Database @@ -161,7 +163,11 @@ func NewProtocolManager(config *core.ChainConfig, fastSync bool, networkId int, heighter := func() uint64 { return blockchain.CurrentBlock().NumberU64() } - manager.fetcher = fetcher.New(blockchain.GetBlock, validator, manager.BroadcastBlock, heighter, manager.insertChain, manager.removePeer) + inserter := func(blocks types.Blocks) (int, error) { + atomic.StoreUint32(&manager.synced, 1) // Mark initial sync done on any fetcher import + return manager.insertChain(blocks) + } + manager.fetcher = fetcher.New(blockchain.GetBlock, validator, manager.BroadcastBlock, heighter, inserter, manager.removePeer) if blockchain.Genesis().Hash().Hex() == defaultGenesisHash && networkId == 1 { glog.V(logger.Debug).Infoln("Bad Block Reporting is enabled") @@ -698,8 +704,8 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } case msg.Code == TxMsg: - // Transactions arrived, make sure we have a valid chain to handle them - if atomic.LoadUint32(&pm.fastSync) == 1 { + // Transactions arrived, make sure we have a valid and fresh chain to handle them + if atomic.LoadUint32(&pm.fastSync) == 1 || atomic.LoadUint32(&pm.synced) == 0 { break } // Transactions can be processed, parse all of them and deliver to the pool diff --git a/eth/protocol_test.go b/eth/protocol_test.go index 0a82e2e795..f860d0a35a 100644 --- a/eth/protocol_test.go +++ b/eth/protocol_test.go @@ -97,6 +97,7 @@ func TestRecvTransactions63(t *testing.T) { testRecvTransactions(t, 63) } func testRecvTransactions(t *testing.T, protocol int) { txAdded := make(chan []*types.Transaction) pm := newTestProtocolManagerMust(t, false, 0, nil, txAdded) + pm.synced = 1 // mark synced to accept transactions p, _ := newTestPeer("peer", protocol, pm, true) defer pm.Stop() defer p.close() diff --git a/eth/sync.go b/eth/sync.go index 4b16c13226..52f7e90e7b 100644 --- a/eth/sync.go +++ b/eth/sync.go @@ -174,6 +174,8 @@ func (pm *ProtocolManager) synchronise(peer *peer) { if err := pm.downloader.Synchronise(peer.id, peer.Head(), peer.Td(), mode); err != nil { return } + atomic.StoreUint32(&pm.synced, 1) // Mark initial sync done + // If fast sync was enabled, and we synced up, disable it if atomic.LoadUint32(&pm.fastSync) == 1 { // Disable fast sync if we indeed have something in our chain From 7c0c63540bf79e8ebe961a7634aa770390a2ce8a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?P=C3=A9ter=20Szil=C3=A1gyi?= Date: Thu, 2 Jun 2016 12:37:14 +0300 Subject: [PATCH 6/6] eth/downloader: make fast sync resilient to critical section fails --- eth/downloader/downloader.go | 68 +++++++++++++++++++++++------- eth/downloader/downloader_test.go | 70 +++++++++++++++++++++++++------ 2 files changed, 111 insertions(+), 27 deletions(-) diff --git a/eth/downloader/downloader.go b/eth/downloader/downloader.go index 37cb04e05d..60f3881750 100644 --- a/eth/downloader/downloader.go +++ b/eth/downloader/downloader.go @@ -74,6 +74,7 @@ var ( fsHeaderForceVerify = 24 // Number of headers to verify before and after the pivot to accept it fsPivotInterval = 512 // Number of headers out of which to randomize the pivot point fsMinFullBlocks = 1024 // Number of blocks to retrieve fully even in fast sync + fsCriticalTrials = 10 // Number of times to retry in the cricical section before bailing ) var ( @@ -104,16 +105,19 @@ var ( ) type Downloader struct { - mode SyncMode // Synchronisation mode defining the strategy used (per sync cycle) - noFast bool // Flag to disable fast syncing in case of a security error - mux *event.TypeMux // Event multiplexer to announce sync operation events + mode SyncMode // Synchronisation mode defining the strategy used (per sync cycle) + mux *event.TypeMux // Event multiplexer to announce sync operation events queue *queue // Scheduler for selecting the hashes to download peers *peerSet // Set of active peers from which download can proceed + fsPivotLock *types.Header // Pivot header on critical section entry (cannot change between retries) + fsPivotFails int // Number of fast sync failures in the critical section + rttEstimate uint64 // Round trip time to target for download requests rttConfidence uint64 // Confidence in the estimated RTT (unit: millionths to allow atomic ops) - interrupt int32 // Atomic boolean to signal termination + + interrupt int32 // Atomic boolean to signal termination // Statistics syncStatsChainOrigin uint64 // Origin block number where syncing started at @@ -323,6 +327,15 @@ func (d *Downloader) synchronise(id string, hash common.Hash, td *big.Int, mode default: } } + for _, ch := range []chan dataPack{d.hashCh, d.blockCh, d.headerCh, d.bodyCh, d.receiptCh, d.stateCh} { + for empty := false; !empty; { + select { + case <-ch: + default: + empty = true + } + } + } for empty := false; !empty; { select { case <-d.headerProcCh: @@ -339,7 +352,7 @@ func (d *Downloader) synchronise(id string, hash common.Hash, td *big.Int, mode // Set the requested sync mode, unless it's forbidden d.mode = mode - if d.mode == FastSync && d.noFast { + if d.mode == FastSync && d.fsPivotFails >= fsCriticalTrials { d.mode = FullSync } // Retrieve the origin peer and initiate the downloading process @@ -422,12 +435,17 @@ func (d *Downloader) syncWithPeer(p *peer, hash common.Hash, td *big.Int) (err e pivot = height case FastSync: // Calculate the new fast/slow sync pivot point - pivotOffset, err := rand.Int(rand.Reader, big.NewInt(int64(fsPivotInterval))) - if err != nil { - panic(fmt.Sprintf("Failed to access crypto random source: %v", err)) - } - if height > uint64(fsMinFullBlocks)+pivotOffset.Uint64() { - pivot = height - uint64(fsMinFullBlocks) - pivotOffset.Uint64() + if d.fsPivotLock == nil { + pivotOffset, err := rand.Int(rand.Reader, big.NewInt(int64(fsPivotInterval))) + if err != nil { + panic(fmt.Sprintf("Failed to access crypto random source: %v", err)) + } + if height > uint64(fsMinFullBlocks)+pivotOffset.Uint64() { + pivot = height - uint64(fsMinFullBlocks) - pivotOffset.Uint64() + } + } else { + // Pivot point locked in, use this and do not pick a new one! + pivot = d.fsPivotLock.Number.Uint64() } // If the point is below the origin, move origin back to ensure state download if pivot < origin { @@ -1227,8 +1245,12 @@ func (d *Downloader) fetchHeaders(p *peer, from uint64) error { // If no more headers are inbound, notify the content fetchers and return if packet.Items() == 0 { glog.V(logger.Debug).Infof("%v: no available headers", p) - d.headerProcCh <- nil - return nil + select { + case d.headerProcCh <- nil: + return nil + case <-d.cancelCh: + return errCancelHeaderFetch + } } headers := packet.(*headerPack).headers @@ -1620,9 +1642,18 @@ func (d *Downloader) processHeaders(origin uint64, td *big.Int) error { glog.V(logger.Warn).Infof("Rolled back %d headers (LH: %d->%d, FB: %d->%d, LB: %d->%d)", len(hashes), lastHeader, d.headHeader().Number, lastFastBlock, d.headFastBlock().Number(), lastBlock, d.headBlock().Number()) - // If we're already past the pivot point, this could be an attack, disable fast sync + // If we're already past the pivot point, this could be an attack, thread carefully if rollback[len(rollback)-1].Number.Uint64() > pivot { - d.noFast = true + // If we didn't ever fail, lock in te pivot header (must! not! change!) + if d.fsPivotFails == 0 { + for _, header := range rollback { + if header.Number.Uint64() == pivot { + glog.V(logger.Warn).Infof("Fast-sync critical section failure, locked pivot to header #%d [%x…]", pivot, header.Hash().Bytes()[:4]) + d.fsPivotLock = header + } + } + } + d.fsPivotFails++ } } }() @@ -1721,6 +1752,13 @@ func (d *Downloader) processHeaders(origin uint64, td *big.Int) error { rollback = append(rollback[:0], rollback[len(rollback)-fsHeaderSafetyNet:]...) } } + // If we're fast syncing and just pulled in the pivot, make sure it's the one locked in + if d.mode == FastSync && d.fsPivotLock != nil && chunk[0].Number.Uint64() <= pivot && chunk[len(chunk)-1].Number.Uint64() >= pivot { + if pivot := chunk[int(pivot-chunk[0].Number.Uint64())]; pivot.Hash() != d.fsPivotLock.Hash() { + glog.V(logger.Warn).Infof("Pivot doesn't match locked in version: have #%v [%x…], want #%v [%x…]", pivot.Number, pivot.Hash().Bytes()[:4], d.fsPivotLock.Number, d.fsPivotLock.Hash().Bytes()[:4]) + return errInvalidChain + } + } // Unless we're doing light chains, schedule the headers for associated content retrieval if d.mode == FullSync || d.mode == FastSync { // If we've reached the allowed number of pending headers, stall a bit diff --git a/eth/downloader/downloader_test.go b/eth/downloader/downloader_test.go index 9eecf87628..1e0cd81c01 100644 --- a/eth/downloader/downloader_test.go +++ b/eth/downloader/downloader_test.go @@ -149,22 +149,25 @@ type downloadTester struct { peerReceipts map[string]map[common.Hash]types.Receipts // Receipts belonging to different test peers peerChainTds map[string]map[common.Hash]*big.Int // Total difficulties of the blocks in the peer chains + peerMissingStates map[string]map[common.Hash]bool // State entries that fast sync should not return + lock sync.RWMutex } // newTester creates a new downloader test mocker. func newTester() *downloadTester { tester := &downloadTester{ - ownHashes: []common.Hash{genesis.Hash()}, - ownHeaders: map[common.Hash]*types.Header{genesis.Hash(): genesis.Header()}, - ownBlocks: map[common.Hash]*types.Block{genesis.Hash(): genesis}, - ownReceipts: map[common.Hash]types.Receipts{genesis.Hash(): nil}, - ownChainTd: map[common.Hash]*big.Int{genesis.Hash(): genesis.Difficulty()}, - peerHashes: make(map[string][]common.Hash), - peerHeaders: make(map[string]map[common.Hash]*types.Header), - peerBlocks: make(map[string]map[common.Hash]*types.Block), - peerReceipts: make(map[string]map[common.Hash]types.Receipts), - peerChainTds: make(map[string]map[common.Hash]*big.Int), + ownHashes: []common.Hash{genesis.Hash()}, + ownHeaders: map[common.Hash]*types.Header{genesis.Hash(): genesis.Header()}, + ownBlocks: map[common.Hash]*types.Block{genesis.Hash(): genesis}, + ownReceipts: map[common.Hash]types.Receipts{genesis.Hash(): nil}, + ownChainTd: map[common.Hash]*big.Int{genesis.Hash(): genesis.Difficulty()}, + peerHashes: make(map[string][]common.Hash), + peerHeaders: make(map[string]map[common.Hash]*types.Header), + peerBlocks: make(map[string]map[common.Hash]*types.Block), + peerReceipts: make(map[string]map[common.Hash]types.Receipts), + peerChainTds: make(map[string]map[common.Hash]*big.Int), + peerMissingStates: make(map[string]map[common.Hash]bool), } tester.stateDb, _ = ethdb.NewMemDatabase() tester.stateDb.Put(genesis.Root().Bytes(), []byte{0x00}) @@ -408,6 +411,7 @@ func (dl *downloadTester) newSlowPeer(id string, version int, hashes []common.Ha dl.peerBlocks[id] = make(map[common.Hash]*types.Block) dl.peerReceipts[id] = make(map[common.Hash]types.Receipts) dl.peerChainTds[id] = make(map[common.Hash]*big.Int) + dl.peerMissingStates[id] = make(map[common.Hash]bool) genesis := hashes[len(hashes)-1] if header := headers[genesis]; header != nil { @@ -648,7 +652,9 @@ func (dl *downloadTester) peerGetNodeDataFn(id string, delay time.Duration) func results := make([][]byte, 0, len(hashes)) for _, hash := range hashes { if data, err := testdb.Get(hash.Bytes()); err == nil { - results = append(results, data) + if !dl.peerMissingStates[id][hash] { + results = append(results, data) + } } } go dl.downloader.DeliverNodeData(id, results) @@ -1288,7 +1294,7 @@ func testInvalidHeaderRollback(t *testing.T, protocol int, mode SyncMode) { tester.newPeer("withhold-attack", protocol, hashes, headers, blocks, receipts) missing = 3*fsHeaderSafetyNet + MaxHeaderFetch + 1 - tester.downloader.noFast = false + tester.downloader.fsPivotFails = 0 tester.downloader.syncInitHook = func(uint64, uint64) { for i := missing; i <= len(hashes); i++ { delete(tester.peerHeaders["withhold-attack"], hashes[len(hashes)-i]) @@ -1307,6 +1313,8 @@ func testInvalidHeaderRollback(t *testing.T, protocol int, mode SyncMode) { t.Errorf("fast sync pivot block #%d not rolled back", head) } } + tester.downloader.fsPivotFails = fsCriticalTrials + // Synchronise with the valid peer and make sure sync succeeds. Since the last // rollback should also disable fast syncing for this process, verify that we // did a fresh full sync. Note, we can't assert anything about the receipts @@ -1749,3 +1757,41 @@ func testDeliverHeadersHang(t *testing.T, protocol int, mode SyncMode) { } } } + +// Tests that if fast sync aborts in the critical section, it can restart a few +// times before giving up. +func TestFastCriticalRestarts63(t *testing.T) { testFastCriticalRestarts(t, 63) } +func TestFastCriticalRestarts64(t *testing.T) { testFastCriticalRestarts(t, 64) } + +func testFastCriticalRestarts(t *testing.T, protocol int) { + t.Parallel() + + // Create a large enough blockchin to actually fast sync on + targetBlocks := fsMinFullBlocks + 2*fsPivotInterval - 15 + hashes, headers, blocks, receipts := makeChain(targetBlocks, 0, genesis, nil, false) + + // Create a tester peer with the critical section state roots missing (force failures) + tester := newTester() + tester.newPeer("peer", protocol, hashes, headers, blocks, receipts) + + for i := 0; i < fsPivotInterval; i++ { + tester.peerMissingStates["peer"][headers[hashes[fsMinFullBlocks+i]].Root] = true + } + // Synchronise with the peer a few times and make sure they fail until the retry limit + for i := 0; i < fsCriticalTrials; i++ { + // Attempt a sync and ensure it fails properly + if err := tester.sync("peer", nil, FastSync); err == nil { + t.Fatalf("failing fast sync succeeded: %v", err) + } + // If it's the first failure, pivot should be locked => reenable all others to detect pivot changes + if i == 0 { + tester.peerMissingStates["peer"] = map[common.Hash]bool{tester.downloader.fsPivotLock.Root: true} + } + time.Sleep(100 * time.Millisecond) // Make sure no in-flight requests remain + } + // Retry limit exhausted, downloader will switch to full sync, should succeed + if err := tester.sync("peer", nil, FastSync); err != nil { + t.Fatalf("failed to synchronise blocks in slow sync: %v", err) + } + assertOwnChain(t, tester, targetBlocks+1) +}