Commit ae11545b authored by Nick Johnson's avatar Nick Johnson

eth, les: Refactor downloader peer to use structs

parent 05509579
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
...@@ -41,7 +41,7 @@ var ( ...@@ -41,7 +41,7 @@ var (
// fetchRequest is a currently running data retrieval operation. // fetchRequest is a currently running data retrieval operation.
type fetchRequest struct { type fetchRequest struct {
Peer *peer // Peer to which the request was sent Peer *peerConnection // Peer to which the request was sent
From uint64 // [eth/62] Requested chain element index (used for skeleton fills only) 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) 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 Headers []*types.Header // [eth/62] Requested headers, sorted by request order
...@@ -391,7 +391,7 @@ func (q *queue) countProcessableItems() int { ...@@ -391,7 +391,7 @@ func (q *queue) countProcessableItems() int {
// ReserveHeaders reserves a set of headers for the given peer, skipping any // ReserveHeaders reserves a set of headers for the given peer, skipping any
// previously failed batches. // previously failed batches.
func (q *queue) ReserveHeaders(p *peer, count int) *fetchRequest { func (q *queue) ReserveHeaders(p *peerConnection, count int) *fetchRequest {
q.lock.Lock() q.lock.Lock()
defer q.lock.Unlock() defer q.lock.Unlock()
...@@ -432,7 +432,7 @@ func (q *queue) ReserveHeaders(p *peer, count int) *fetchRequest { ...@@ -432,7 +432,7 @@ func (q *queue) ReserveHeaders(p *peer, count int) *fetchRequest {
// ReserveBodies reserves a set of body fetches for the given peer, skipping any // ReserveBodies reserves a set of body fetches for the given peer, skipping any
// previously failed downloads. Beside the next batch of needed fetches, it also // previously failed downloads. Beside the next batch of needed fetches, it also
// returns a flag whether empty blocks were queued requiring processing. // returns a flag whether empty blocks were queued requiring processing.
func (q *queue) ReserveBodies(p *peer, count int) (*fetchRequest, bool, error) { func (q *queue) ReserveBodies(p *peerConnection, count int) (*fetchRequest, bool, error) {
isNoop := func(header *types.Header) bool { isNoop := func(header *types.Header) bool {
return header.TxHash == types.EmptyRootHash && header.UncleHash == types.EmptyUncleHash return header.TxHash == types.EmptyRootHash && header.UncleHash == types.EmptyUncleHash
} }
...@@ -445,7 +445,7 @@ func (q *queue) ReserveBodies(p *peer, count int) (*fetchRequest, bool, error) { ...@@ -445,7 +445,7 @@ func (q *queue) ReserveBodies(p *peer, count int) (*fetchRequest, bool, error) {
// ReserveReceipts reserves a set of receipt fetches for the given peer, skipping // ReserveReceipts reserves a set of receipt fetches for the given peer, skipping
// any previously failed downloads. Beside the next batch of needed fetches, it // any previously failed downloads. Beside the next batch of needed fetches, it
// also returns a flag whether empty receipts were queued requiring importing. // also returns a flag whether empty receipts were queued requiring importing.
func (q *queue) ReserveReceipts(p *peer, count int) (*fetchRequest, bool, error) { func (q *queue) ReserveReceipts(p *peerConnection, count int) (*fetchRequest, bool, error) {
isNoop := func(header *types.Header) bool { isNoop := func(header *types.Header) bool {
return header.ReceiptHash == types.EmptyRootHash return header.ReceiptHash == types.EmptyRootHash
} }
...@@ -462,7 +462,7 @@ func (q *queue) ReserveReceipts(p *peer, count int) (*fetchRequest, bool, error) ...@@ -462,7 +462,7 @@ func (q *queue) ReserveReceipts(p *peer, count int) (*fetchRequest, bool, error)
// Note, this method expects the queue lock to be already held for writing. The // Note, this method expects the queue lock to be already held for writing. The
// reason the lock is not obtained in here is because the parameters already need // reason the lock is not obtained in here is because the parameters already need
// to access the queue, so they already need a lock anyway. // to access the queue, so they already need a lock anyway.
func (q *queue) reserveHeaders(p *peer, count int, taskPool map[common.Hash]*types.Header, taskQueue *prque.Prque, func (q *queue) reserveHeaders(p *peerConnection, count int, taskPool map[common.Hash]*types.Header, taskQueue *prque.Prque,
pendPool map[string]*fetchRequest, donePool map[common.Hash]struct{}, isNoop func(*types.Header) bool) (*fetchRequest, bool, error) { pendPool map[string]*fetchRequest, donePool map[common.Hash]struct{}, isNoop func(*types.Header) bool) (*fetchRequest, bool, error) {
// Short circuit if the pool has been depleted, or if the peer's already // Short circuit if the pool has been depleted, or if the peer's already
// downloading something (sanity check not to corrupt state) // downloading something (sanity check not to corrupt state)
......
...@@ -37,7 +37,7 @@ type stateReq struct { ...@@ -37,7 +37,7 @@ type stateReq struct {
tasks map[common.Hash]*stateTask // Download tasks to track previous attempts tasks map[common.Hash]*stateTask // Download tasks to track previous attempts
timeout time.Duration // Maximum round trip time for this to complete timeout time.Duration // Maximum round trip time for this to complete
timer *time.Timer // Timer to fire when the RTT timeout expires timer *time.Timer // Timer to fire when the RTT timeout expires
peer *peer // Peer that we're requesting from peer *peerConnection // Peer that we're requesting from
response [][]byte // Response data of the peer (nil for timeouts) response [][]byte // Response data of the peer (nil for timeouts)
} }
...@@ -246,7 +246,7 @@ func (s *stateSync) Cancel() error { ...@@ -246,7 +246,7 @@ func (s *stateSync) Cancel() error {
// and timeouts. // and timeouts.
func (s *stateSync) loop() error { func (s *stateSync) loop() error {
// Listen for new peer events to assign tasks to them // Listen for new peer events to assign tasks to them
newPeer := make(chan *peer, 1024) newPeer := make(chan *peerConnection, 1024)
peerSub := s.d.peers.SubscribeNewPeers(newPeer) peerSub := s.d.peers.SubscribeNewPeers(newPeer)
defer peerSub.Unsubscribe() defer peerSub.Unsubscribe()
......
...@@ -265,7 +265,7 @@ func (pm *ProtocolManager) handle(p *peer) error { ...@@ -265,7 +265,7 @@ func (pm *ProtocolManager) handle(p *peer) error {
defer pm.removePeer(p.id) defer pm.removePeer(p.id)
// Register the peer in the downloader. If the downloader considers it banned, we disconnect // Register the peer in the downloader. If the downloader considers it banned, we disconnect
if err := pm.downloader.RegisterPeer(p.id, p.version, p.Head, p.RequestHeadersByHash, p.RequestHeadersByNumber, p.RequestBodies, p.RequestReceipts, p.RequestNodeData); err != nil { if err := pm.downloader.RegisterPeer(p.id, p.version, p); err != nil {
return err return err
} }
// Propagate existing transactions. new transactions appearing // Propagate existing transactions. new transactions appearing
......
...@@ -838,57 +838,83 @@ func (self *ProtocolManager) NodeInfo() *eth.EthNodeInfo { ...@@ -838,57 +838,83 @@ func (self *ProtocolManager) NodeInfo() *eth.EthNodeInfo {
// downloaderPeerNotify implements peerSetNotify // downloaderPeerNotify implements peerSetNotify
type downloaderPeerNotify ProtocolManager type downloaderPeerNotify ProtocolManager
func (d *downloaderPeerNotify) registerPeer(p *peer) { type peerConnection struct {
pm := (*ProtocolManager)(d) manager *ProtocolManager
peer *peer
}
requestHeadersByHash := func(origin common.Hash, amount int, skip int, reverse bool) error { func (pc *peerConnection) Head() (common.Hash, *big.Int) {
reqID := genReqID() return pc.peer.HeadAndTd()
rq := &distReq{ }
getCost: func(dp distPeer) uint64 {
peer := dp.(*peer) func (pc *peerConnection) RequestHeadersByHash(origin common.Hash, amount int, skip int, reverse bool) error {
return peer.GetRequestCost(GetBlockHeadersMsg, amount) reqID := genReqID()
}, rq := &distReq{
canSend: func(dp distPeer) bool { getCost: func(dp distPeer) uint64 {
return dp.(*peer) == p peer := dp.(*peer)
}, return peer.GetRequestCost(GetBlockHeadersMsg, amount)
request: func(dp distPeer) func() { },
peer := dp.(*peer) canSend: func(dp distPeer) bool {
cost := peer.GetRequestCost(GetBlockHeadersMsg, amount) return dp.(*peer) == pc.peer
peer.fcServer.QueueRequest(reqID, cost) },
return func() { peer.RequestHeadersByHash(reqID, cost, origin, amount, skip, reverse) } request: func(dp distPeer) func() {
}, peer := dp.(*peer)
} cost := peer.GetRequestCost(GetBlockHeadersMsg, amount)
_, ok := <-pm.reqDist.queue(rq) peer.fcServer.QueueRequest(reqID, cost)
if !ok { return func() { peer.RequestHeadersByHash(reqID, cost, origin, amount, skip, reverse) }
return ErrNoPeers },
}
return nil
} }
requestHeadersByNumber := func(origin uint64, amount int, skip int, reverse bool) error { _, ok := <-pc.manager.reqDist.queue(rq)
reqID := genReqID() if !ok {
rq := &distReq{ return ErrNoPeers
getCost: func(dp distPeer) uint64 { }
peer := dp.(*peer) return nil
return peer.GetRequestCost(GetBlockHeadersMsg, amount) }
},
canSend: func(dp distPeer) bool { func (pc *peerConnection) RequestHeadersByNumber(origin uint64, amount int, skip int, reverse bool) error {
return dp.(*peer) == p reqID := genReqID()
}, rq := &distReq{
request: func(dp distPeer) func() { getCost: func(dp distPeer) uint64 {
peer := dp.(*peer) peer := dp.(*peer)
cost := peer.GetRequestCost(GetBlockHeadersMsg, amount) return peer.GetRequestCost(GetBlockHeadersMsg, amount)
peer.fcServer.QueueRequest(reqID, cost) },
return func() { peer.RequestHeadersByNumber(reqID, cost, origin, amount, skip, reverse) } canSend: func(dp distPeer) bool {
}, return dp.(*peer) == pc.peer
} },
_, ok := <-pm.reqDist.queue(rq) request: func(dp distPeer) func() {
if !ok { peer := dp.(*peer)
return ErrNoPeers cost := peer.GetRequestCost(GetBlockHeadersMsg, amount)
} peer.fcServer.QueueRequest(reqID, cost)
return nil return func() { peer.RequestHeadersByNumber(reqID, cost, origin, amount, skip, reverse) }
},
}
_, ok := <-pc.manager.reqDist.queue(rq)
if !ok {
return ErrNoPeers
} }
return nil
}
func (pc *peerConnection) RequestBodies(hashes []common.Hash) error {
panic("RequestBodies not supported in light client mode sync")
}
func (pc *peerConnection) RequestReceipts(hashes []common.Hash) error {
panic("RequestReceipts not supported in light client mode sync")
}
func (pc *peerConnection) RequestNodeData(hashes []common.Hash) error {
panic("RequestNodeData not supported in light client mode sync")
}
pm.downloader.RegisterPeer(p.id, ethVersion, p.HeadAndTd, requestHeadersByHash, requestHeadersByNumber, nil, nil, nil) func (d *downloaderPeerNotify) registerPeer(p *peer) {
pm := (*ProtocolManager)(d)
pc := &peerConnection{
manager: pm,
peer: p,
}
pm.downloader.RegisterPeer(p.id, ethVersion, pc)
} }
func (d *downloaderPeerNotify) unregisterPeer(p *peer) { func (d *downloaderPeerNotify) unregisterPeer(p *peer) {
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment