From 95a05d0d9a45a86613e66aa4461d8f38e9bae89b Mon Sep 17 00:00:00 2001 From: Zsolt Felfoldi Date: Wed, 20 Nov 2019 19:53:56 +0100 Subject: [PATCH] p2p/discv5: talkRequest works --- les/server_handler.go | 2 +- p2p/discv5/net.go | 102 +++++++++++++++++++++++++++++++++--------- p2p/discv5/udp.go | 6 +-- 3 files changed, 85 insertions(+), 25 deletions(-) diff --git a/les/server_handler.go b/les/server_handler.go index 5c80b9ffb1..c4a18b2945 100644 --- a/les/server_handler.go +++ b/les/server_handler.go @@ -956,6 +956,6 @@ func (h *serverHandler) broadcastHeaders() { } } -func (h *serverHandler) talkRequestHandler(id enode.ID, addr *net.UDPAddr, payload rlp.RawValue) (rlp.RawValue, bool) { +func (h *serverHandler) talkRequestHandler(id enode.ID, addr *net.UDPAddr, payload []byte) ([]byte, bool) { return payload, true } diff --git a/p2p/discv5/net.go b/p2p/discv5/net.go index 1c407d1d87..d020c48adf 100644 --- a/p2p/discv5/net.go +++ b/p2p/discv5/net.go @@ -66,7 +66,7 @@ type Network struct { refreshResp chan (<-chan struct{}) // ...and get the channel to block on from this one read chan ingressPacket // ingress packets arrive here timeout chan timeoutEvent - queryReq chan *findnodeQuery // lookups submit findnode queries on this channel + queryReq chan deferredQuery // lookups submit findnode queries on this channel tableOpReq chan func() tableOpResp chan struct{} topicRegisterReq chan topicRegisterReq @@ -87,8 +87,8 @@ type Network struct { } type ( - TalkRequestHandler func(enode.ID, *net.UDPAddr, rlp.RawValue) (rlp.RawValue, bool) - TalkResponseHandler func(rlp.RawValue) bool + TalkRequestHandler func(enode.ID, *net.UDPAddr, []byte) ([]byte, bool) + TalkResponseHandler func([]byte) bool ) // transport is implemented by the UDP transport. @@ -113,6 +113,14 @@ type findnodeQuery struct { reply chan<- []*Node } +type talkQuery struct { + remote *Node + talkID string + payload rlp.RawValue + key string + handler TalkResponseHandler +} + type topicRegisterReq struct { add bool topic Topic @@ -163,7 +171,7 @@ func newNetwork(conn transport, ourPubkey ecdsa.PublicKey, dbPath string, netres timeoutTimers: make(map[timeoutEvent]*time.Timer), tableOpReq: make(chan func()), tableOpResp: make(chan struct{}), - queryReq: make(chan *findnodeQuery), + queryReq: make(chan deferredQuery), topicRegisterReq: make(chan topicRegisterReq), topicSearchReq: make(chan topicSearchReq), nodes: make(map[NodeID]*Node), @@ -460,7 +468,7 @@ loop: case q := <-net.queryReq: log.Trace("<-net.queryReq") if !q.start(net) { - q.remote.deferQuery(q) + q.deferQuery() } // Interacting with the table. @@ -781,15 +789,21 @@ type nodeNetGuts struct { // State machine fields. Access to these fields // is restricted to the Network.loop goroutine. state *nodeState - pingEcho []byte // hash of last ping sent by us - pingTopics []Topic // topic set sent by us in last ping - deferredQueries []*findnodeQuery // queries that can't be sent yet - pendingNeighbours *findnodeQuery // current query, waiting for reply + pingEcho []byte // hash of last ping sent by us + pingTopics []Topic // topic set sent by us in last ping + deferredQueries []deferredQuery // queries that can't be sent yet + pendingNeighbours *findnodeQuery // current query, waiting for reply queryTimeouts int talkFailures int } -func (n *nodeNetGuts) deferQuery(q *findnodeQuery) { +type deferredQuery interface { + start(net *Network) bool + cancel() + deferQuery() +} + +func (n *nodeNetGuts) deferQuery(q deferredQuery) { n.deferredQueries = append(n.deferredQueries, q) } @@ -825,6 +839,14 @@ func (q *findnodeQuery) start(net *Network) bool { return false } +func (q *findnodeQuery) cancel() { + q.reply <- nil +} + +func (q *findnodeQuery) deferQuery() { + q.remote.deferQuery(q) +} + // Node Events (the input to the state machine). type nodeEvent uint @@ -886,7 +908,7 @@ func init() { n.pingEcho = nil // Abort active queries. for _, q := range n.deferredQueries { - q.reply <- nil + q.cancel() } n.deferredQueries = nil if n.pendingNeighbours != nil { @@ -1335,22 +1357,62 @@ func (net *Network) RemoveTalkHandler(talkID string) { net.talkRequestSubLock.Unlock() } -func (net *Network) SendTalkRequest(to *enode.Node, talkID string, payload rlp.RawValue, handler TalkResponseHandler) func() bool { +func (q *talkQuery) start(net *Network) bool { + if q.remote == net.tab.self { + return false + } + if q.remote.state.canQuery { + net.talkResponseSubLock.Lock() + hash := net.conn.send(q.remote, talkRequestPacket, talkRequest{TalkID: []byte(q.talkID), Payload: q.payload}) + q.key = string(q.remote.sha[:]) + string(hash[:]) + net.talkResponseSubs[q.key] = q.handler + net.talkResponseSubLock.Unlock() + fmt.Println("sent") + return true + } + // If the node is not known yet, it won't accept queries. + // Initiate the transition to known. + // The request will be sent later when the node reaches known state. + if q.remote.state == unknown { + net.transition(q.remote, verifyinit) + } + return false +} + +func (q *talkQuery) cancel() { + q.handler(nil) +} + +func (q *talkQuery) deferQuery() { + q.remote.deferQuery(q) +} + +func (net *Network) SendTalkRequest(to *enode.Node, talkID string, payload []byte, handler TalkResponseHandler) func() bool { var nodeID NodeID copy(nodeID[:], crypto.FromECDSAPub(to.Pubkey())[1:]) - node := NewNode(nodeID, to.IP(), uint16(to.UDP()), uint16(to.TCP())) - net.talkResponseSubLock.Lock() - hash := net.conn.send(node, talkRequestPacket, talkRequest{TalkID: []byte(talkID), Payload: payload}) - key := string(node.sha[:]) + string(hash[:]) - net.talkResponseSubs[key] = handler - net.talkResponseSubLock.Unlock() + node := net.nodes[nodeID] + if node == nil { + node = NewNode(nodeID, to.IP(), uint16(to.UDP()), uint16(to.TCP())) + node.state = unknown + net.nodes[nodeID] = node + } + q := &talkQuery{remote: node, talkID: talkID, payload: payload, handler: handler} + select { + case net.queryReq <- q: + case <-net.closed: + return nil + } + return func() bool { net.talkResponseSubLock.Lock() - cancel := net.talkResponseSubs[key] != nil + cancel := q.key != "" && net.talkResponseSubs[q.key] != nil if cancel { - delete(net.talkResponseSubs, key) + delete(net.talkResponseSubs, q.key) } net.talkResponseSubLock.Unlock() + if cancel { + handler(nil) + } return cancel } } diff --git a/p2p/discv5/udp.go b/p2p/discv5/udp.go index b7f272e702..4f630dab1f 100644 --- a/p2p/discv5/udp.go +++ b/p2p/discv5/udp.go @@ -120,13 +120,11 @@ type ( } talkRequest struct { - TalkID []byte - Payload rlp.RawValue + TalkID, Payload []byte } talkResponse struct { - ReplyTok []byte - Payload rlp.RawValue + ReplyTok, Payload []byte } rpcNode struct {