Skip to content

Commit 04c5048

Browse files
qcloudGrapeBaBa
qcloud
authored and
GrapeBaBa
committed
refactor: contentLookup and traceContentLookup
1 parent 1ae723d commit 04c5048

File tree

4 files changed

+100
-122
lines changed

4 files changed

+100
-122
lines changed

‎go.sum‎

Lines changed: 0 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -429,12 +429,6 @@ github.com/onsi/gomega v1.10.1 h1:o0+MgICZLuZ7xjH7Vx6zS/zcu93/BEp1VwkIW1mEXCE=
429429
github.com/onsi/gomega v1.10.1/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo=
430430
github.com/opentracing/opentracing-go v1.1.0 h1:pWlfV3Bxv7k65HYwkikxat0+s3pV4bsqf19k25Ur8rU=
431431
github.com/opentracing/opentracing-go v1.1.0/go.mod h1:UkNAQd3GIcIGf0SeVgPpRdFStlNbqXla1AfSYxPUl2o=
432-
github.com/optimism-java/utp-go v0.0.0-20240309041853-b6b3a0dea581 h1:ZxgrtI0xIw+clB32iDDDWaiTcCizTeN7rNyzH9YorPI=
433-
github.com/optimism-java/utp-go v0.0.0-20240309041853-b6b3a0dea581/go.mod h1:DZ0jYzLzt4ZsCmhI/iqYgGFoNx45OfpEoKzXB8HVALQ=
434-
github.com/optimism-java/utp-go v0.0.0-20240530085325-d8dd9d262631 h1:01AecSuOSS6fsIU/oTVG/C70hIl3xPen99qy2hGr57w=
435-
github.com/optimism-java/utp-go v0.0.0-20240530085325-d8dd9d262631/go.mod h1:DZ0jYzLzt4ZsCmhI/iqYgGFoNx45OfpEoKzXB8HVALQ=
436-
github.com/optimism-java/utp-go v0.0.0-20240531021243-e12d25b6be38 h1:t0gRqfM7wUrFyryagUpw4TmYY0DLt+rjPaBd92i+W2M=
437-
github.com/optimism-java/utp-go v0.0.0-20240531021243-e12d25b6be38/go.mod h1:DZ0jYzLzt4ZsCmhI/iqYgGFoNx45OfpEoKzXB8HVALQ=
438432
github.com/optimism-java/utp-go v0.0.0-20240531024756-00da67044c50 h1:I1jGQkNEWq7BTFZkCJKLDrqFLC1jR3EC7jz3to4kpLg=
439433
github.com/optimism-java/utp-go v0.0.0-20240531024756-00da67044c50/go.mod h1:DZ0jYzLzt4ZsCmhI/iqYgGFoNx45OfpEoKzXB8HVALQ=
440434
github.com/optimism-java/utp-go v0.0.0-20240603010819-75be99daf402 h1:jssfGQq6xdzgs0ZI/O/S2dKAyh2fIDiOTWdrbqat1Ls=
@@ -715,8 +709,6 @@ golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
715709
golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
716710
golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
717711
golang.org/x/sys v0.11.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
718-
golang.org/x/sys v0.19.0 h1:q5f1RH2jigJ1MoAWp2KTp3gm5zAGFUTarQZ5U386+4o=
719-
golang.org/x/sys v0.19.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
720712
golang.org/x/sys v0.20.0 h1:Od9JTbYCk261bKm4M/mw7AklTlFYIa0bIp9BgSm1S8Y=
721713
golang.org/x/sys v0.20.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
722714
golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=

‎p2p/discover/api.go‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -56,7 +56,7 @@ type Trace struct {
5656
Origin string `json:"origin"` // local node id
5757
TargetId string `json:"targetId"` // target content id
5858
ReceivedFrom string `json:"receivedFrom"` // the node id of which content from
59-
Responses map[string][]string `json:"responses"` // the node id and there response nodeIds
59+
Responses map[string]RespByNode `json:"responses"` // the node id and there response nodeIds
6060
Metadata map[string]*NodeMetadata `json:"metadata"` // node id and there metadata object
6161
StartedAtMs int `json:"startedAtMs"` // timestamp of the beginning of this request in milliseconds
6262
Cancelled []string `json:"cancelled"` // the node ids which are send but cancelled
@@ -67,6 +67,11 @@ type NodeMetadata struct {
6767
Distance string `json:"distance"`
6868
}
6969

70+
type RespByNode struct {
71+
DurationMs int32 `json:"durationMs"`
72+
RespondedWith []string `json:"respondedWith"`
73+
}
74+
7075
type Enrs struct {
7176
Enrs []string `json:"enrs"`
7277
}

‎p2p/discover/portal_protocol.go‎

Lines changed: 87 additions & 106 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@ import (
1616
"slices"
1717
"sort"
1818
"sync"
19+
"sync/atomic"
1920
"time"
2021

2122
"github.com/ethereum/go-ethereum/common/hexutil"
@@ -1488,64 +1489,44 @@ func (p *PortalProtocol) collectTableNodes(rip net.IP, distances []uint, limit i
14881489

14891490
func (p *PortalProtocol) ContentLookup(contentKey, contentId []byte) ([]byte, bool, error) {
14901491
lookupContext, cancel := context.WithCancel(context.Background())
1491-
defer cancel()
1492-
resChan := make(chan *ContentInfoResp, 1)
1493-
defer close(resChan)
1492+
1493+
resChan := make(chan *traceContentInfoResp, alpha)
1494+
hasResult := int32(0)
1495+
1496+
result := ContentInfoResp{}
1497+
1498+
var wg sync.WaitGroup
1499+
wg.Add(1)
1500+
1501+
go func() {
1502+
defer wg.Done()
1503+
for res := range resChan {
1504+
if res.Flag != portalwire.ContentEnrsSelector {
1505+
result.Content = res.Content.([]byte)
1506+
result.UtpTransfer = res.UtpTransfer
1507+
}
1508+
}
1509+
}()
1510+
14941511
newLookup(lookupContext, p.table, enode.ID(contentId), func(n *node) ([]*node, error) {
1495-
return p.contentLookupWorker(unwrapNode(n), contentKey, resChan)
1512+
return p.contentLookupWorker(unwrapNode(n), contentKey, resChan, cancel, &hasResult)
14961513
}).run()
1514+
close(resChan)
14971515

1498-
if len(resChan) > 0 {
1499-
res := <-resChan
1500-
return res.Content, res.UtpTransfer, nil
1516+
wg.Wait()
1517+
if hasResult == 1 {
1518+
return result.Content, result.UtpTransfer, nil
15011519
}
1520+
defer cancel()
15021521
return nil, false, ContentNotFound
15031522
}
15041523

1505-
func (p *PortalProtocol) contentLookupWorker(n *enode.Node, contentKey []byte, resChan chan<- *ContentInfoResp) ([]*node, error) {
1506-
wrapedNode := make([]*node, 0)
1507-
flag, content, err := p.findContent(n, contentKey)
1508-
if err != nil {
1509-
p.Log.Error("contentLookupWorker failed", "ip", n.IP().String(), "err", err)
1510-
return nil, err
1511-
}
1512-
p.Log.Debug("contentLookupWorker reveice response", "ip", n.IP().String(), "flag", flag)
1513-
// has find content
1514-
if len(resChan) > 0 {
1515-
return []*node{}, nil
1516-
}
1517-
switch flag {
1518-
case portalwire.ContentRawSelector, portalwire.ContentConnIdSelector:
1519-
content, ok := content.([]byte)
1520-
if !ok {
1521-
return wrapedNode, fmt.Errorf("failed to assert to raw content, value is: %v", content)
1522-
}
1523-
res := &ContentInfoResp{
1524-
Content: content,
1525-
}
1526-
if flag == portalwire.ContentConnIdSelector {
1527-
res.UtpTransfer = true
1528-
}
1529-
resChan <- res
1530-
return wrapedNode, err
1531-
case portalwire.ContentEnrsSelector:
1532-
nodes, ok := content.([]*enode.Node)
1533-
if !ok {
1534-
return wrapedNode, fmt.Errorf("failed to assert to enrs content, value is: %v", content)
1535-
}
1536-
return wrapNodes(nodes), nil
1537-
}
1538-
return wrapedNode, nil
1539-
}
1540-
15411524
func (p *PortalProtocol) TraceContentLookup(contentKey, contentId []byte) (*TraceContentResult, error) {
15421525
lookupContext, cancel := context.WithCancel(context.Background())
1543-
defer cancel()
1544-
requestNodeChan := make(chan *enode.Node, 3)
1545-
resChan := make(chan *traceContentInfoResp, 3)
1526+
// resp channel
1527+
resChan := make(chan *traceContentInfoResp, alpha)
15461528

1547-
requestNode := make([]*enode.Node, 0)
1548-
requestRes := make(map[string]*traceContentInfoResp)
1529+
hasResult := int32(0)
15491530

15501531
traceContentRes := &TraceContentResult{}
15511532

@@ -1555,7 +1536,7 @@ func (p *PortalProtocol) TraceContentLookup(contentKey, contentId []byte) (*Trac
15551536
Origin: selfHexId,
15561537
TargetId: hexutil.Encode(contentId),
15571538
StartedAtMs: int(time.Now().UnixMilli()),
1558-
Responses: make(map[string][]string),
1539+
Responses: make(map[string]RespByNode),
15591540
Metadata: make(map[string]*NodeMetadata),
15601541
Cancelled: make([]string, 0),
15611542
}
@@ -1567,7 +1548,10 @@ func (p *PortalProtocol) TraceContentLookup(contentKey, contentId []byte) (*Trac
15671548
id := "0x" + node.ID().String()
15681549
localResponse = append(localResponse, id)
15691550
}
1570-
trace.Responses[selfHexId] = localResponse
1551+
trace.Responses[selfHexId] = RespByNode{
1552+
DurationMs: 0,
1553+
RespondedWith: localResponse,
1554+
}
15711555

15721556
dis := p.Distance(p.Self().ID(), enode.ID(contentId))
15731557

@@ -1577,82 +1561,73 @@ func (p *PortalProtocol) TraceContentLookup(contentKey, contentId []byte) (*Trac
15771561
}
15781562

15791563
var wg sync.WaitGroup
1580-
wg.Add(2)
1581-
go func() {
1582-
defer wg.Done()
1583-
for node := range requestNodeChan {
1584-
requestNode = append(requestNode, node)
1585-
}
1586-
}()
1564+
wg.Add(1)
15871565

15881566
go func() {
15891567
defer wg.Done()
15901568
for res := range resChan {
1591-
key := res.Node.ID().String()
1592-
requestRes[key] = res
1593-
if res.Flag == portalwire.ContentRawSelector || res.Flag == portalwire.ContentConnIdSelector {
1594-
// get the content
1595-
return
1569+
node := res.Node
1570+
hexId := "0x" + node.ID().String()
1571+
dis := p.Distance(node.ID(), enode.ID(contentId))
1572+
p.Log.Debug("reveice res", "id", hexId, "flag", res.Flag)
1573+
trace.Metadata[hexId] = &NodeMetadata{
1574+
Enr: node.String(),
1575+
Distance: hexutil.Encode(dis[:]),
1576+
}
1577+
// 没有返回 content
1578+
if traceContentRes.Content == "" {
1579+
if res.Flag == portalwire.ContentRawSelector || res.Flag == portalwire.ContentConnIdSelector {
1580+
trace.ReceivedFrom = hexId
1581+
content := res.Content.([]byte)
1582+
traceContentRes.Content = hexutil.Encode(content)
1583+
traceContentRes.UtpTransfer = res.UtpTransfer
1584+
trace.Responses[hexId] = RespByNode{}
1585+
} else {
1586+
nodes := res.Content.([]*enode.Node)
1587+
respByNode := RespByNode{
1588+
RespondedWith: make([]string, 0, len(nodes)),
1589+
}
1590+
for _, node := range nodes {
1591+
idInner := "0x" + node.ID().String()
1592+
respByNode.RespondedWith = append(respByNode.RespondedWith, idInner)
1593+
if _, ok := trace.Metadata[idInner]; !ok {
1594+
dis := p.Distance(node.ID(), enode.ID(contentId))
1595+
trace.Metadata[idInner] = &NodeMetadata{
1596+
Enr: node.String(),
1597+
Distance: hexutil.Encode(dis[:]),
1598+
}
1599+
}
1600+
trace.Responses[hexId] = respByNode
1601+
}
1602+
}
1603+
} else {
1604+
trace.Cancelled = append(trace.Cancelled, hexId)
15961605
}
15971606
}
15981607
}()
15991608

1600-
newLookup(lookupContext, p.table, enode.ID(contentId), func(n *node) ([]*node, error) {
1601-
node := unwrapNode(n)
1602-
requestNodeChan <- node
1603-
return p.traceContentLookupWorker(node, contentKey, resChan)
1604-
}).run()
1605-
1606-
close(requestNodeChan)
1609+
lookup := newLookup(lookupContext, p.table, enode.ID(contentId), func(n *node) ([]*node, error) {
1610+
return p.contentLookupWorker(unwrapNode(n), contentKey, resChan, cancel, &hasResult)
1611+
})
1612+
lookup.run()
16071613
close(resChan)
16081614

16091615
wg.Wait()
1610-
1611-
for _, node := range requestNode {
1612-
id := node.ID().String()
1613-
hexId := "0x" + id
1614-
dis := p.Distance(node.ID(), enode.ID(contentId))
1615-
trace.Metadata[hexId] = &NodeMetadata{
1616-
Enr: node.String(),
1617-
Distance: hexutil.Encode(dis[:]),
1618-
}
1619-
if res, ok := requestRes[id]; ok {
1620-
if res.Flag == portalwire.ContentRawSelector || res.Flag == portalwire.ContentConnIdSelector {
1621-
trace.ReceivedFrom = hexId
1622-
content := res.Content.([]byte)
1623-
traceContentRes.Content = hexutil.Encode(content)
1624-
traceContentRes.UtpTransfer = res.UtpTransfer
1625-
trace.Responses[hexId] = make([]string, 0)
1626-
} else {
1627-
content := res.Content.([]*enode.Node)
1628-
ids := make([]string, 0)
1629-
for _, n := range content {
1630-
hexId := "0x" + n.ID().String()
1631-
ids = append(ids, hexId)
1632-
}
1633-
trace.Responses[hexId] = ids
1634-
}
1635-
} else {
1636-
trace.Cancelled = append(trace.Cancelled, id)
1637-
}
1616+
if hasResult == 0 {
1617+
cancel()
16381618
}
1639-
16401619
traceContentRes.Trace = *trace
16411620

16421621
return traceContentRes, nil
16431622
}
16441623

1645-
func (p *PortalProtocol) traceContentLookupWorker(n *enode.Node, contentKey []byte, resChan chan<- *traceContentInfoResp) ([]*node, error) {
1624+
func (p *PortalProtocol) contentLookupWorker(n *enode.Node, contentKey []byte, resChan chan<- *traceContentInfoResp, cancel context.CancelFunc, done *int32) ([]*node, error) {
16461625
wrapedNode := make([]*node, 0)
16471626
flag, content, err := p.findContent(n, contentKey)
16481627
if err != nil {
16491628
return nil, err
16501629
}
16511630
p.Log.Debug("traceContentLookupWorker reveice response", "ip", n.IP().String(), "flag", flag)
1652-
// has find content
1653-
if len(resChan) > 0 {
1654-
return []*node{}, nil
1655-
}
16561631

16571632
switch flag {
16581633
case portalwire.ContentRawSelector, portalwire.ContentConnIdSelector:
@@ -1669,17 +1644,23 @@ func (p *PortalProtocol) traceContentLookupWorker(n *enode.Node, contentKey []by
16691644
if flag == portalwire.ContentConnIdSelector {
16701645
res.UtpTransfer = true
16711646
}
1672-
resChan <- res
1647+
if atomic.CompareAndSwapInt32(done, 0, 1) {
1648+
p.Log.Debug("contentLookupWorker find content", "ip", n.IP().String(), "port", n.UDP())
1649+
resChan <- res
1650+
cancel()
1651+
}
16731652
return wrapedNode, err
16741653
case portalwire.ContentEnrsSelector:
16751654
nodes, ok := content.([]*enode.Node)
16761655
if !ok {
16771656
return wrapedNode, fmt.Errorf("failed to assert to enrs content, value is: %v", content)
16781657
}
1679-
resChan <- &traceContentInfoResp{Node: n,
1658+
resChan <- &traceContentInfoResp{
1659+
Node: n,
16801660
Flag: flag,
16811661
Content: content,
1682-
UtpTransfer: false}
1662+
UtpTransfer: false,
1663+
}
16831664
return wrapNodes(nodes), nil
16841665
}
16851666
return wrapedNode, nil

‎p2p/discover/portal_protocol_test.go‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -458,6 +458,10 @@ func TestTraceContentLookup(t *testing.T) {
458458
err = node3.Start()
459459
assert.NoError(t, err)
460460

461+
defer node1.Stop()
462+
defer node2.Stop()
463+
defer node3.Stop()
464+
461465
contentKey := []byte{0x3, 0x4}
462466
content := []byte{0x1, 0x2}
463467
contentId := node1.toContentId(contentKey)
@@ -495,15 +499,11 @@ func TestTraceContentLookup(t *testing.T) {
495499

496500
// check response
497501
node3Response := res.Trace.Responses[node3Id]
498-
assert.Equal(t, node3Response, []string{node2Id})
502+
assert.Equal(t, node3Response.RespondedWith, []string{node2Id})
499503

500504
node2Response := res.Trace.Responses[node2Id]
501-
assert.Equal(t, node2Response, []string{node1Id})
505+
assert.Equal(t, node2Response.RespondedWith, []string{node1Id})
502506

503507
node1Response := res.Trace.Responses[node1Id]
504-
assert.Equal(t, node1Response, []string{})
505-
506-
// res, _, err = node1.ContentLookup([]byte{0x2, 0x4})
507-
// assert.Equal(t, ContentNotFound, err)
508-
// assert.Nil(t, res)
508+
assert.Equal(t, node1Response.RespondedWith, ([]string)(nil))
509509
}

0 commit comments

Comments
 (0)