p2p/discover: send a packet and not wait for a response

Signed-off-by: thinkAfCod <[email protected]>

Author: thinkAfCod

p2p/discover/v5_udp.go

@@ -93,12 +93,14 @@ type UDPv5 struct {
 	callDoneCh    chan *callV5
 	respTimeoutCh chan *callTimeout
 	sendCh        chan sendRequest
+	sendNoRespCh  chan *sendNoRespRequest
 	unhandled     chan<- ReadPacket
 
 	// state of dispatch
 	codec            codecV5
 	activeCallByNode map[enode.ID]*callV5
 	activeCallByAuth map[v5wire.Nonce]*callV5
+	noRespCallByAuth map[v5wire.Nonce]*callV5
 	callQueue        map[enode.ID][]*callV5
 
 	// shutdown stuff
@@ -112,6 +114,13 @@ type sendRequest struct {
 	destID   enode.ID
 	destAddr netip.AddrPort
 	msg      v5wire.Packet
+	destNode *enode.Node
+}
+
+type sendNoRespRequest struct {
+	destNode *enode.Node
+	destAddr netip.AddrPort
+	msg      v5wire.Packet
 }
 
 // callV5 represents a remote procedure call against another node.
@@ -173,12 +182,14 @@ func newUDPv5(conn UDPConn, ln *enode.LocalNode, cfg Config) (*UDPv5, error) {
 		callCh:        make(chan *callV5),
 		callDoneCh:    make(chan *callV5),
 		sendCh:        make(chan sendRequest),
+		sendNoRespCh:  make(chan *sendNoRespRequest),
 		respTimeoutCh: make(chan *callTimeout),
 		unhandled:     cfg.Unhandled,
 		// state of dispatch
 		codec:            v5wire.NewCodec(ln, cfg.PrivateKey, cfg.Clock, cfg.V5ProtocolID),
 		activeCallByNode: make(map[enode.ID]*callV5),
 		activeCallByAuth: make(map[v5wire.Nonce]*callV5),
+		noRespCallByAuth: make(map[v5wire.Nonce]*callV5),
 		callQueue:        make(map[enode.ID][]*callV5),
 		// shutdown
 		closeCtx:       closeCtx,
@@ -578,12 +589,18 @@ func (t *UDPv5) dispatch() {
 		case c := <-t.callCh:
 			t.callQueue[c.id] = append(t.callQueue[c.id], c)
 			t.sendNextCall(c.id)
+		case cnr := <-t.sendNoRespCh:
+			// send a TalkReq call but not waiting for TalkResp
+			// may more used by portal network
+			t.sendCallNotWaitResp(cnr)
 
 		case ct := <-t.respTimeoutCh:
 			active := t.activeCallByNode[ct.c.id]
 			if ct.c == active && ct.timer == active.timeout {
 				ct.c.err <- errTimeout
 			}
+			delete(t.noRespCallByAuth, ct.c.nonce)
+			ct.c.timeout.Stop()
 
 		case c := <-t.callDoneCh:
 			active := t.activeCallByNode[c.id]
@@ -597,7 +614,6 @@ func (t *UDPv5) dispatch() {
 
 		case r := <-t.sendCh:
 			t.send(r.destID, r.destAddr, r.msg, nil)
-
 		case p := <-t.packetInCh:
 			t.handlePacket(p.Data, p.Addr)
 			// Arm next read.
@@ -616,6 +632,9 @@ func (t *UDPv5) dispatch() {
 				delete(t.activeCallByNode, id)
 				delete(t.activeCallByAuth, c.nonce)
 			}
+			for nonce := range t.noRespCallByAuth {
+				delete(t.activeCallByAuth, nonce)
+			}
 			return
 		}
 	}
@@ -641,6 +660,36 @@ func (t *UDPv5) startResponseTimeout(c *callV5) {
 	close(done)
 }
 
+// sendCallNotWaitResp send a talk request contains utp packet by call, call will not insert into queue
+func (t *UDPv5) sendCallNotWaitResp(r *sendNoRespRequest) {
+	// send out a TalkRequest that is a UTP packet for portal network
+	// todo If the destination node has been handshaked, it may not be necessary to use call
+	// request should be cached to handle WHOAREYOU
+	c := &callV5{id: r.destNode.ID(), addr: r.destAddr}
+	c.node = r.destNode
+	c.packet = r.msg
+	c.reqid = make([]byte, 8)
+	c.ch = make(chan v5wire.Packet, 1)
+	c.err = make(chan error, 1)
+	// Assign request ID.
+	crand.Read(c.reqid)
+	c.packet.SetRequestID(c.reqid)
+
+	nonce, _ := t.send(c.id, c.addr, c.packet, nil)
+	c.nonce = nonce
+	t.noRespCallByAuth[nonce] = c
+	t.startResponseTimeout(c)
+}
+
+// sendNoRespData send a data from a call of no wait resp.
+// The handshake just has been successful.
+func (t *UDPv5) sendNoRespData(c *callV5) {
+	// Just resend the data and not use call again
+	delete(t.noRespCallByAuth, c.nonce)
+	c.timeout.Stop()
+	t.send(c.node.ID(), c.addr, c.packet, nil)
+}
+
 // sendNextCall sends the next call in the call queue if there is no active call.
 func (t *UDPv5) sendNextCall(id enode.ID) {
 	queue := t.callQueue[id]
@@ -681,7 +730,15 @@ func (t *UDPv5) sendResponse(toID enode.ID, toAddr netip.AddrPort, packet v5wire
 
 func (t *UDPv5) sendFromAnotherThread(toID enode.ID, toAddr netip.AddrPort, packet v5wire.Packet) {
 	select {
-	case t.sendCh <- sendRequest{toID, toAddr, packet}:
+	case t.sendCh <- sendRequest{destID: toID, destAddr: toAddr, msg: packet}:
+	case <-t.closeCtx.Done():
+	}
+}
+
+// SendNoResp Send a packet but will not wait for a response
+func (t *UDPv5) SendNoResp(n *enode.Node, toAddr netip.AddrPort, packet v5wire.Packet) {
+	select {
+	case t.sendNoRespCh <- &sendNoRespRequest{n, toAddr, packet}:
 	case <-t.closeCtx.Done():
 	}
 }
@@ -874,17 +931,25 @@ func (t *UDPv5) handleWhoareyou(p *v5wire.Whoareyou, fromID enode.ID, fromAddr n
 		c.err <- errors.New("remote wants handshake, but call has no ENR")
 		return
 	}
-	// Resend the call that was answered by WHOAREYOU.
+
 	t.log.Trace("<< "+p.Name(), "id", c.node.ID(), "addr", fromAddr)
-	c.handshakeCount++
-	c.challenge = p
-	p.Node = c.node
-	t.sendCall(c)
+	if _, ok := t.noRespCallByAuth[p.Nonce]; !ok {
+		// Resend the call that was answered by WHOAREYOU.
+		c.handshakeCount++
+		c.challenge = p
+		p.Node = c.node
+		t.sendCall(c)
+	} else {
+		t.sendNoRespData(c)
+	}
 }
 
 // matchWithCall checks whether a handshake attempt matches the active call.
 func (t *UDPv5) matchWithCall(fromID enode.ID, nonce v5wire.Nonce) (*callV5, error) {
 	c := t.activeCallByAuth[nonce]
+	if c == nil {
+		c = t.noRespCallByAuth[nonce]
+	}
 	if c == nil {
 		return nil, errChallengeNoCall
 	}