Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
whqwjb
go-ethereum
提交
b40dc8a1
G
go-ethereum
项目概览
whqwjb
/
go-ethereum
与 Fork 源项目一致
从无法访问的项目Fork
通知
1
Star
0
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
G
go-ethereum
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
前往新版Gitcode,体验更适合开发者的 AI 搜索 >>
提交
b40dc8a1
编写于
2月 25, 2016
作者:
P
Péter Szilágyi
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
eth/downloader: implement concurrent header downloads
上级
fe532a98
变更
4
展开全部
隐藏空白更改
内联
并排
Showing
4 changed file
with
584 addition
and
211 deletion
+584
-211
eth/downloader/downloader.go
eth/downloader/downloader.go
+299
-181
eth/downloader/downloader_test.go
eth/downloader/downloader_test.go
+24
-23
eth/downloader/peer.go
eth/downloader/peer.go
+61
-5
eth/downloader/queue.go
eth/downloader/queue.go
+200
-2
未找到文件。
eth/downloader/downloader.go
浏览文件 @
b40dc8a1
此差异已折叠。
点击以展开。
eth/downloader/downloader_test.go
浏览文件 @
b40dc8a1
...
...
@@ -560,8 +560,8 @@ func (dl *downloadTester) peerGetAbsHeadersFn(id string, delay time.Duration) fu
hashes
:=
dl
.
peerHashes
[
id
]
headers
:=
dl
.
peerHeaders
[
id
]
result
:=
make
([]
*
types
.
Header
,
0
,
amount
)
for
i
:=
0
;
i
<
amount
&&
len
(
hashes
)
-
int
(
origin
)
-
1
-
i
>=
0
;
i
++
{
if
header
,
ok
:=
headers
[
hashes
[
len
(
hashes
)
-
int
(
origin
)
-
1
-
i
]];
ok
{
for
i
:=
0
;
i
<
amount
&&
len
(
hashes
)
-
int
(
origin
)
-
1
-
i
*
(
skip
+
1
)
>=
0
;
i
++
{
if
header
,
ok
:=
headers
[
hashes
[
len
(
hashes
)
-
int
(
origin
)
-
1
-
i
*
(
skip
+
1
)
]];
ok
{
result
=
append
(
result
,
header
)
}
}
...
...
@@ -1348,27 +1348,28 @@ func testBlockHeaderAttackerDropping(t *testing.T, protocol int) {
result
error
drop
bool
}{
{
nil
,
false
},
// Sync succeeded, all is well
{
errBusy
,
false
},
// Sync is already in progress, no problem
{
errUnknownPeer
,
false
},
// Peer is unknown, was already dropped, don't double drop
{
errBadPeer
,
true
},
// Peer was deemed bad for some reason, drop it
{
errStallingPeer
,
true
},
// Peer was detected to be stalling, drop it
{
errNoPeers
,
false
},
// No peers to download from, soft race, no issue
{
errTimeout
,
true
},
// No hashes received in due time, drop the peer
{
errEmptyHashSet
,
true
},
// No hashes were returned as a response, drop as it's a dead end
{
errEmptyHeaderSet
,
true
},
// No headers were returned as a response, drop as it's a dead end
{
errPeersUnavailable
,
true
},
// Nobody had the advertised blocks, drop the advertiser
{
errInvalidAncestor
,
true
},
// Agreed upon ancestor is not acceptable, drop the chain rewriter
{
errInvalidChain
,
true
},
// Hash chain was detected as invalid, definitely drop
{
errInvalidBlock
,
false
},
// A bad peer was detected, but not the sync origin
{
errInvalidBody
,
false
},
// A bad peer was detected, but not the sync origin
{
errInvalidReceipt
,
false
},
// A bad peer was detected, but not the sync origin
{
errCancelHashFetch
,
false
},
// Synchronisation was canceled, origin may be innocent, don't drop
{
errCancelBlockFetch
,
false
},
// Synchronisation was canceled, origin may be innocent, don't drop
{
errCancelHeaderFetch
,
false
},
// Synchronisation was canceled, origin may be innocent, don't drop
{
errCancelBodyFetch
,
false
},
// Synchronisation was canceled, origin may be innocent, don't drop
{
errCancelReceiptFetch
,
false
},
// Synchronisation was canceled, origin may be innocent, don't drop
{
errCancelProcessing
,
false
},
// Synchronisation was canceled, origin may be innocent, don't drop
{
nil
,
false
},
// Sync succeeded, all is well
{
errBusy
,
false
},
// Sync is already in progress, no problem
{
errUnknownPeer
,
false
},
// Peer is unknown, was already dropped, don't double drop
{
errBadPeer
,
true
},
// Peer was deemed bad for some reason, drop it
{
errStallingPeer
,
true
},
// Peer was detected to be stalling, drop it
{
errNoPeers
,
false
},
// No peers to download from, soft race, no issue
{
errTimeout
,
true
},
// No hashes received in due time, drop the peer
{
errEmptyHashSet
,
true
},
// No hashes were returned as a response, drop as it's a dead end
{
errEmptyHeaderSet
,
true
},
// No headers were returned as a response, drop as it's a dead end
{
errPeersUnavailable
,
true
},
// Nobody had the advertised blocks, drop the advertiser
{
errInvalidAncestor
,
true
},
// Agreed upon ancestor is not acceptable, drop the chain rewriter
{
errInvalidChain
,
true
},
// Hash chain was detected as invalid, definitely drop
{
errInvalidBlock
,
false
},
// A bad peer was detected, but not the sync origin
{
errInvalidBody
,
false
},
// A bad peer was detected, but not the sync origin
{
errInvalidReceipt
,
false
},
// A bad peer was detected, but not the sync origin
{
errCancelHashFetch
,
false
},
// Synchronisation was canceled, origin may be innocent, don't drop
{
errCancelBlockFetch
,
false
},
// Synchronisation was canceled, origin may be innocent, don't drop
{
errCancelHeaderFetch
,
false
},
// Synchronisation was canceled, origin may be innocent, don't drop
{
errCancelBodyFetch
,
false
},
// Synchronisation was canceled, origin may be innocent, don't drop
{
errCancelReceiptFetch
,
false
},
// Synchronisation was canceled, origin may be innocent, don't drop
{
errCancelHeaderProcessing
,
false
},
// Synchronisation was canceled, origin may be innocent, don't drop
{
errCancelContentProcessing
,
false
},
// Synchronisation was canceled, origin may be innocent, don't drop
}
// Run the tests and check disconnection status
tester
:=
newTester
()
...
...
eth/downloader/peer.go
浏览文件 @
b40dc8a1
...
...
@@ -58,15 +58,18 @@ type peer struct {
id
string
// Unique identifier of the peer
head
common
.
Hash
// Hash of the peers latest known block
headerIdle
int32
// Current header activity state of the peer (idle = 0, active = 1)
blockIdle
int32
// Current block activity state of the peer (idle = 0, active = 1)
receiptIdle
int32
// Current receipt activity state of the peer (idle = 0, active = 1)
stateIdle
int32
// Current node data activity state of the peer (idle = 0, active = 1)
headerThroughput
float64
// Number of headers measured to be retrievable per second
blockThroughput
float64
// Number of blocks (bodies) measured to be retrievable per second
receiptThroughput
float64
// Number of receipts measured to be retrievable per second
stateThroughput
float64
// Number of node data pieces measured to be retrievable per second
blockStarted
time
.
Time
// Time instance when the last block (body)fetch was started
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
stateStarted
time
.
Time
// Time instance when the last node data fetch was started
...
...
@@ -118,10 +121,12 @@ func (p *peer) Reset() {
p
.
lock
.
Lock
()
defer
p
.
lock
.
Unlock
()
atomic
.
StoreInt32
(
&
p
.
headerIdle
,
0
)
atomic
.
StoreInt32
(
&
p
.
blockIdle
,
0
)
atomic
.
StoreInt32
(
&
p
.
receiptIdle
,
0
)
atomic
.
StoreInt32
(
&
p
.
stateIdle
,
0
)
p
.
headerThroughput
=
0
p
.
blockThroughput
=
0
p
.
receiptThroughput
=
0
p
.
stateThroughput
=
0
...
...
@@ -151,6 +156,24 @@ func (p *peer) Fetch61(request *fetchRequest) error {
return
nil
}
// FetchHeaders sends a header retrieval request to the remote peer.
func
(
p
*
peer
)
FetchHeaders
(
from
uint64
,
count
int
)
error
{
// Sanity check the protocol version
if
p
.
version
<
62
{
panic
(
fmt
.
Sprintf
(
"header fetch [eth/62+] requested on eth/%d"
,
p
.
version
))
}
// Short circuit if the peer is already fetching
if
!
atomic
.
CompareAndSwapInt32
(
&
p
.
headerIdle
,
0
,
1
)
{
return
errAlreadyFetching
}
p
.
headerStarted
=
time
.
Now
()
// Issue the header retrieval request (absolut upwards without gaps)
go
p
.
getAbsHeaders
(
from
,
count
,
0
,
false
)
return
nil
}
// FetchBodies sends a block body retrieval request to the remote peer.
func
(
p
*
peer
)
FetchBodies
(
request
*
fetchRequest
)
error
{
// Sanity check the protocol version
...
...
@@ -217,6 +240,13 @@ func (p *peer) FetchNodeData(request *fetchRequest) error {
return
nil
}
// SetHeadersIdle sets the peer to idle, allowing it to execute new header retrieval
// requests. Its estimated header retrieval throughput is updated with that measured
// just now.
func
(
p
*
peer
)
SetHeadersIdle
(
delivered
int
)
{
p
.
setIdle
(
p
.
headerStarted
,
delivered
,
&
p
.
headerThroughput
,
&
p
.
headerIdle
)
}
// SetBlocksIdle sets the peer to idle, allowing it to execute new block retrieval
// requests. Its estimated block retrieval throughput is updated with that measured
// just now.
...
...
@@ -264,6 +294,15 @@ func (p *peer) setIdle(started time.Time, delivered int, throughput *float64, id
*
throughput
=
(
1
-
throughputImpact
)
*
(
*
throughput
)
+
throughputImpact
*
measured
}
// HeaderCapacity retrieves the peers header download allowance based on its
// previously discovered throughput.
func
(
p
*
peer
)
HeaderCapacity
()
int
{
p
.
lock
.
RLock
()
defer
p
.
lock
.
RUnlock
()
return
int
(
math
.
Max
(
1
,
math
.
Min
(
p
.
headerThroughput
*
float64
(
headerTargetRTT
)
/
float64
(
time
.
Second
),
float64
(
MaxHeaderFetch
))))
}
// BlockCapacity retrieves the peers block download allowance based on its
// previously discovered throughput.
func
(
p
*
peer
)
BlockCapacity
()
int
{
...
...
@@ -323,14 +362,15 @@ func (p *peer) String() string {
defer
p
.
lock
.
RUnlock
()
return
fmt
.
Sprintf
(
"Peer %s [%s]"
,
p
.
id
,
fmt
.
Sprintf
(
"blocks %3.2f/s, "
,
p
.
blockThroughput
)
+
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
)),
)
}
// peerSet represents the collection of active peer participating in the
block
// peerSet represents the collection of active peer participating in the
chain
// download procedure.
type
peerSet
struct
{
peers
map
[
string
]
*
peer
...
...
@@ -359,7 +399,7 @@ func (ps *peerSet) Reset() {
// peer is already known.
//
// The method also sets the starting throughput values of the new peer to the
// average of all existing peers, to give it a realistic chan
g
e of being used
// average of all existing peers, to give it a realistic chan
c
e of being used
// for data retrievals.
func
(
ps
*
peerSet
)
Register
(
p
*
peer
)
error
{
ps
.
lock
.
Lock
()
...
...
@@ -369,15 +409,17 @@ func (ps *peerSet) Register(p *peer) error {
return
errAlreadyRegistered
}
if
len
(
ps
.
peers
)
>
0
{
p
.
blockThroughput
,
p
.
receiptThroughput
,
p
.
stateThroughput
=
0
,
0
,
0
p
.
headerThroughput
,
p
.
blockThroughput
,
p
.
receiptThroughput
,
p
.
stateThroughput
=
0
,
0
,
0
,
0
for
_
,
peer
:=
range
ps
.
peers
{
peer
.
lock
.
RLock
()
p
.
headerThroughput
+=
peer
.
headerThroughput
p
.
blockThroughput
+=
peer
.
blockThroughput
p
.
receiptThroughput
+=
peer
.
receiptThroughput
p
.
stateThroughput
+=
peer
.
stateThroughput
peer
.
lock
.
RUnlock
()
}
p
.
headerThroughput
/=
float64
(
len
(
ps
.
peers
))
p
.
blockThroughput
/=
float64
(
len
(
ps
.
peers
))
p
.
receiptThroughput
/=
float64
(
len
(
ps
.
peers
))
p
.
stateThroughput
/=
float64
(
len
(
ps
.
peers
))
...
...
@@ -441,6 +483,20 @@ func (ps *peerSet) BlockIdlePeers() ([]*peer, int) {
return
ps
.
idlePeers
(
61
,
61
,
idle
,
throughput
)
}
// HeaderIdlePeers retrieves a flat list of all the currently header-idle peers
// within the active peer set, ordered by their reputation.
func
(
ps
*
peerSet
)
HeaderIdlePeers
()
([]
*
peer
,
int
)
{
idle
:=
func
(
p
*
peer
)
bool
{
return
atomic
.
LoadInt32
(
&
p
.
headerIdle
)
==
0
}
throughput
:=
func
(
p
*
peer
)
float64
{
p
.
lock
.
RLock
()
defer
p
.
lock
.
RUnlock
()
return
p
.
headerThroughput
}
return
ps
.
idlePeers
(
62
,
64
,
idle
,
throughput
)
}
// BodyIdlePeers retrieves a flat list of all the currently body-idle peers within
// the active peer set, ordered by their reputation.
func
(
ps
*
peerSet
)
BodyIdlePeers
()
([]
*
peer
,
int
)
{
...
...
eth/downloader/queue.go
浏览文件 @
b40dc8a1
...
...
@@ -39,8 +39,8 @@ import (
)
var
(
blockCacheLimit
=
8192
// Maximum number of blocks to cache before throttling the download
maxInFlightStates
=
4096
// Maximum number of state downloads to allow concurrently
blockCacheLimit
=
16384
// Maximum number of blocks to cache before throttling the download
maxInFlightStates
=
8192
// Maximum number of state downloads to allow concurrently
)
var
(
...
...
@@ -52,6 +52,7 @@ var (
// fetchRequest is a currently running data retrieval operation.
type
fetchRequest
struct
{
Peer
*
peer
// Peer to which the request was sent
From
uint64
// [eth/62] Requested chain element index (used for skeleton fills only)
Hashes
map
[
common
.
Hash
]
int
// [eth/61] Requested hashes with their insertion index (priority)
Headers
[]
*
types
.
Header
// [eth/62] Requested headers, sorted by request order
Time
time
.
Time
// Time when the request was made
...
...
@@ -79,6 +80,17 @@ type queue struct {
headerHead
common
.
Hash
// [eth/62] Hash of the last queued header to verify order
// Headers are "special", they download in batches, supported by a skeleton chain
headerTaskPool
map
[
uint64
]
*
types
.
Header
// [eth/62] Pending header retrieval tasks, mapping starting indexes to skeleton headers
headerTaskQueue
*
prque
.
Prque
// [eth/62] Priority queue of the skeleton indexes to fetch the filling headers for
headerPeerMiss
map
[
string
]
map
[
uint64
]
struct
{}
// [eth/62] Set of per-peer header batches known to be unavailable
headerPendPool
map
[
string
]
*
fetchRequest
// [eth/62] Currently pending header retrieval operations
headerDonePool
map
[
uint64
]
struct
{}
// [eth/62] Set of the completed header fetches
headerResults
[]
*
types
.
Header
// [eth/62] Result cache accumulating the completed headers
headerOffset
uint64
// [eth/62] Number of the first header in the result cache
headerContCh
chan
bool
// [eth/62] Channel to notify when header download finishes
// All data retrievals below are based on an already assembles header chain
blockTaskPool
map
[
common
.
Hash
]
*
types
.
Header
// [eth/62] Pending block (body) retrieval tasks, mapping hashes to headers
blockTaskQueue
*
prque
.
Prque
// [eth/62] Priority queue of the headers to fetch the blocks (bodies) for
blockPendPool
map
[
string
]
*
fetchRequest
// [eth/62] Currently pending block (body) retrieval operations
...
...
@@ -113,6 +125,8 @@ func newQueue(stateDb ethdb.Database) *queue {
return
&
queue
{
hashPool
:
make
(
map
[
common
.
Hash
]
int
),
hashQueue
:
prque
.
New
(),
headerPendPool
:
make
(
map
[
string
]
*
fetchRequest
),
headerContCh
:
make
(
chan
bool
),
blockTaskPool
:
make
(
map
[
common
.
Hash
]
*
types
.
Header
),
blockTaskQueue
:
prque
.
New
(),
blockPendPool
:
make
(
map
[
string
]
*
fetchRequest
),
...
...
@@ -149,6 +163,8 @@ func (q *queue) Reset() {
q
.
headerHead
=
common
.
Hash
{}
q
.
headerPendPool
=
make
(
map
[
string
]
*
fetchRequest
)
q
.
blockTaskPool
=
make
(
map
[
common
.
Hash
]
*
types
.
Header
)
q
.
blockTaskQueue
.
Reset
()
q
.
blockPendPool
=
make
(
map
[
string
]
*
fetchRequest
)
...
...
@@ -178,6 +194,14 @@ func (q *queue) Close() {
q
.
active
.
Broadcast
()
}
// PendingHeaders retrieves the number of header requests pending for retrieval.
func
(
q
*
queue
)
PendingHeaders
()
int
{
q
.
lock
.
Lock
()
defer
q
.
lock
.
Unlock
()
return
q
.
headerTaskQueue
.
Size
()
}
// PendingBlocks retrieves the number of block (body) requests pending for retrieval.
func
(
q
*
queue
)
PendingBlocks
()
int
{
q
.
lock
.
Lock
()
...
...
@@ -205,6 +229,15 @@ func (q *queue) PendingNodeData() int {
return
0
}
// InFlightHeaders retrieves whether there are header fetch requests currently
// in flight.
func
(
q
*
queue
)
InFlightHeaders
()
bool
{
q
.
lock
.
Lock
()
defer
q
.
lock
.
Unlock
()
return
len
(
q
.
headerPendPool
)
>
0
}
// InFlightBlocks retrieves whether there are block fetch requests currently in
// flight.
func
(
q
*
queue
)
InFlightBlocks
()
bool
{
...
...
@@ -317,6 +350,44 @@ func (q *queue) Schedule61(hashes []common.Hash, fifo bool) []common.Hash {
return
inserts
}
// ScheduleSkeleton adds a batch of header retrieval tasks to the queue to fill
// up an already retrieved header skeleton.
func
(
q
*
queue
)
ScheduleSkeleton
(
from
uint64
,
skeleton
[]
*
types
.
Header
)
{
q
.
lock
.
Lock
()
defer
q
.
lock
.
Unlock
()
// No skeleton retrieval can be in progress, fail hard if so (huge implementation bug)
if
q
.
headerResults
!=
nil
{
panic
(
"skeleton assembly already in progress"
)
}
// Shedule all the header retrieval tasks for the skeleton assembly
q
.
headerTaskPool
=
make
(
map
[
uint64
]
*
types
.
Header
)
q
.
headerTaskQueue
=
prque
.
New
()
q
.
headerPeerMiss
=
make
(
map
[
string
]
map
[
uint64
]
struct
{})
// Reset availability to correct invalid chains
q
.
headerResults
=
make
([]
*
types
.
Header
,
len
(
skeleton
)
*
MaxHeaderFetch
)
q
.
headerOffset
=
from
q
.
headerContCh
=
make
(
chan
bool
,
1
)
for
i
,
header
:=
range
skeleton
{
index
:=
from
+
uint64
(
i
*
MaxHeaderFetch
)
q
.
headerTaskPool
[
index
]
=
header
q
.
headerTaskQueue
.
Push
(
index
,
-
float32
(
index
))
}
}
// RetrieveHeaders retrieves the header chain assemble based on the scheduled
// skeleton.
func
(
q
*
queue
)
RetrieveHeaders
()
[]
*
types
.
Header
{
q
.
lock
.
Lock
()
defer
q
.
lock
.
Unlock
()
headers
:=
q
.
headerResults
q
.
headerResults
=
nil
return
headers
}
// Schedule adds a set of headers for the download queue for scheduling, returning
// the new headers encountered.
func
(
q
*
queue
)
Schedule
(
headers
[]
*
types
.
Header
,
from
uint64
)
[]
*
types
.
Header
{
...
...
@@ -437,6 +508,46 @@ func (q *queue) countProcessableItems() int {
return
len
(
q
.
resultCache
)
}
// ReserveHeaders reserves a set of headers for the given peer, skipping any
// previously failed batches.
func
(
q
*
queue
)
ReserveHeaders
(
p
*
peer
,
count
int
)
*
fetchRequest
{
q
.
lock
.
Lock
()
defer
q
.
lock
.
Unlock
()
// Short circuit if the peer's already downloading something (sanity check to
// not corrupt state)
if
_
,
ok
:=
q
.
headerPendPool
[
p
.
id
];
ok
{
return
nil
}
// Retrieve a batch of hashes, skipping previously failed ones
send
,
skip
:=
uint64
(
0
),
[]
uint64
{}
for
send
==
0
&&
!
q
.
headerTaskQueue
.
Empty
()
{
from
,
_
:=
q
.
headerTaskQueue
.
Pop
()
if
q
.
headerPeerMiss
[
p
.
id
]
!=
nil
{
if
_
,
ok
:=
q
.
headerPeerMiss
[
p
.
id
][
from
.
(
uint64
)];
ok
{
skip
=
append
(
skip
,
from
.
(
uint64
))
continue
}
}
send
=
from
.
(
uint64
)
}
// Merge all the skipped batches back
for
_
,
from
:=
range
skip
{
q
.
headerTaskQueue
.
Push
(
from
,
-
float32
(
from
))
}
// Assemble and return the block download request
if
send
==
0
{
return
nil
}
request
:=
&
fetchRequest
{
Peer
:
p
,
From
:
send
,
Time
:
time
.
Now
(),
}
q
.
headerPendPool
[
p
.
id
]
=
request
return
request
}
// ReserveBlocks reserves a set of block hashes for the given peer, skipping any
// previously failed download.
func
(
q
*
queue
)
ReserveBlocks
(
p
*
peer
,
count
int
)
*
fetchRequest
{
...
...
@@ -635,6 +746,11 @@ func (q *queue) reserveHeaders(p *peer, count int, taskPool map[common.Hash]*typ
return
request
,
progress
,
nil
}
// CancelHeaders aborts a fetch request, returning all pending skeleton indexes to the queue.
func
(
q
*
queue
)
CancelHeaders
(
request
*
fetchRequest
)
{
q
.
cancel
(
request
,
q
.
headerTaskQueue
,
q
.
headerPendPool
)
}
// CancelBlocks aborts a fetch request, returning all pending hashes to the queue.
func
(
q
*
queue
)
CancelBlocks
(
request
*
fetchRequest
)
{
q
.
cancel
(
request
,
q
.
hashQueue
,
q
.
blockPendPool
)
...
...
@@ -663,6 +779,9 @@ func (q *queue) cancel(request *fetchRequest, taskQueue *prque.Prque, pendPool m
q
.
lock
.
Lock
()
defer
q
.
lock
.
Unlock
()
if
request
.
From
>
0
{
taskQueue
.
Push
(
request
.
From
,
-
float32
(
request
.
From
))
}
for
hash
,
index
:=
range
request
.
Hashes
{
taskQueue
.
Push
(
hash
,
float32
(
index
))
}
...
...
@@ -702,6 +821,15 @@ func (q *queue) Revoke(peerId string) {
}
}
// ExpireHeaders checks for in flight requests that exceeded a timeout allowance,
// canceling them and returning the responsible peers for penalisation.
func
(
q
*
queue
)
ExpireHeaders
(
timeout
time
.
Duration
)
map
[
string
]
int
{
q
.
lock
.
Lock
()
defer
q
.
lock
.
Unlock
()
return
q
.
expire
(
timeout
,
q
.
headerPendPool
,
q
.
headerTaskQueue
,
headerTimeoutMeter
)
}
// ExpireBlocks checks for in flight requests that exceeded a timeout allowance,
// canceling them and returning the responsible peers for penalisation.
func
(
q
*
queue
)
ExpireBlocks
(
timeout
time
.
Duration
)
map
[
string
]
int
{
...
...
@@ -753,6 +881,9 @@ func (q *queue) expire(timeout time.Duration, pendPool map[string]*fetchRequest,
timeoutMeter
.
Mark
(
1
)
// Return any non satisfied requests to the pool
if
request
.
From
>
0
{
taskQueue
.
Push
(
request
.
From
,
-
float32
(
request
.
From
))
}
for
hash
,
index
:=
range
request
.
Hashes
{
taskQueue
.
Push
(
hash
,
float32
(
index
))
}
...
...
@@ -842,6 +973,73 @@ func (q *queue) DeliverBlocks(id string, blocks []*types.Block) (int, error) {
}
}
// DeliverHeaders injects a header retrieval response into the header results
// cache. This method either accepts all headers it received, or none of them
// if they do not map correctly to the skeleton.
func
(
q
*
queue
)
DeliverHeaders
(
id
string
,
headers
[]
*
types
.
Header
)
(
int
,
error
)
{
q
.
lock
.
Lock
()
defer
q
.
lock
.
Unlock
()
// Short circuit if the data was never requested
request
:=
q
.
headerPendPool
[
id
]
if
request
==
nil
{
return
0
,
errNoFetchesPending
}
headerReqTimer
.
UpdateSince
(
request
.
Time
)
delete
(
q
.
headerPendPool
,
id
)
// Ensure headers can be mapped onto the skeleton chain
target
:=
q
.
headerTaskPool
[
request
.
From
]
.
Hash
()
accepted
:=
len
(
headers
)
==
MaxHeaderFetch
if
accepted
{
if
headers
[
0
]
.
Number
.
Uint64
()
!=
request
.
From
{
glog
.
V
(
logger
.
Detail
)
.
Infof
(
"Peer %s: first header #%v [%x] broke chain ordering, expected %d"
,
id
,
headers
[
0
]
.
Number
,
headers
[
0
]
.
Hash
()
.
Bytes
()[
:
4
],
request
.
From
)
accepted
=
false
}
else
if
headers
[
len
(
headers
)
-
1
]
.
Hash
()
!=
target
{
glog
.
V
(
logger
.
Detail
)
.
Infof
(
"Peer %s: last header #%v [%x] broke skeleton structure, expected %x"
,
id
,
headers
[
len
(
headers
)
-
1
]
.
Number
,
headers
[
len
(
headers
)
-
1
]
.
Hash
()
.
Bytes
()[
:
4
],
target
[
:
4
])
accepted
=
false
}
}
if
accepted
{
for
i
,
header
:=
range
headers
[
1
:
]
{
hash
:=
header
.
Hash
()
if
want
:=
request
.
From
+
1
+
uint64
(
i
);
header
.
Number
.
Uint64
()
!=
want
{
glog
.
V
(
logger
.
Warn
)
.
Infof
(
"Peer %s: header #%v [%x] broke chain ordering, expected %d"
,
id
,
header
.
Number
,
hash
[
:
4
],
want
)
accepted
=
false
break
}
if
headers
[
i
]
.
Hash
()
!=
header
.
ParentHash
{
glog
.
V
(
logger
.
Warn
)
.
Infof
(
"Peer %s: header #%v [%x] broke chain ancestry"
,
id
,
header
.
Number
,
hash
[
:
4
])
accepted
=
false
break
}
}
}
// If the batch of headers wasn't accepted, mark as unavailable
if
!
accepted
{
glog
.
V
(
logger
.
Detail
)
.
Infof
(
"Peer %s: skeleton filling from header #%d not accepted"
,
id
,
request
.
From
)
miss
:=
q
.
headerPeerMiss
[
id
]
if
miss
==
nil
{
q
.
headerPeerMiss
[
id
]
=
make
(
map
[
uint64
]
struct
{})
miss
=
q
.
headerPeerMiss
[
id
]
}
miss
[
request
.
From
]
=
struct
{}{}
q
.
headerTaskQueue
.
Push
(
request
.
From
,
-
float32
(
request
.
From
))
return
0
,
errors
.
New
(
"delivery not accepted"
)
}
// Clean up a successful fetch, check for termination and return
copy
(
q
.
headerResults
[
request
.
From
-
q
.
headerOffset
:
],
headers
)
delete
(
q
.
headerTaskPool
,
request
.
From
)
if
len
(
q
.
headerTaskPool
)
==
0
{
q
.
headerContCh
<-
false
}
return
len
(
headers
),
nil
}
// DeliverBodies injects a block body retrieval response into the results queue.
// The method returns the number of blocks bodies accepted from the delivery and
// also wakes any threads waiting for data delivery.
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录